Skip to content

Commit 922b519

Browse files
vkhanhquivkhanhqui
committed
Refactor
1 parent ebcd217 commit 922b519

File tree

6 files changed

+189
-94
lines changed

6 files changed

+189
-94
lines changed

‎013-cloud-gaming/main.go

+8-1
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,22 @@
11
package main
22

33
import (
4+
"cloud-gaming/service"
45
"cloud-gaming/transport"
56
"fmt"
67
"log"
78
"net/http"
89
)
910

1011
func main() {
12+
wk := service.NewWorker()
13+
wk.Run()
14+
15+
svc := service.NewService()
16+
handler := transport.NewHandler(*svc)
17+
1118
http.Handle("/", http.FileServer(http.Dir("assets")))
12-
http.HandleFunc("/ws", transport.WebsocketHandler)
19+
http.HandleFunc("/ws", handler.Websocket)
1320

1421
port := 8080
1522
log.Printf("starting server on port %v", port)

‎013-cloud-gaming/service/service.go

+62
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
package service
2+
3+
import (
4+
"log"
5+
6+
"github.com/gorilla/websocket"
7+
"github.com/pion/webrtc/v4"
8+
)
9+
10+
var (
11+
peerConns []*PeerConnState
12+
peerConnCh = make(chan *PeerConnState)
13+
)
14+
15+
func NewService() *Service {
16+
return &Service{}
17+
}
18+
19+
type Service struct{}
20+
21+
func (s *Service) HandleMessage(p *PeerConnState) {
22+
p.PeerConnection().OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
23+
log.Println("Connection state changed:", state)
24+
if state == webrtc.PeerConnectionStateConnected {
25+
peerConnCh <- p
26+
log.Println("Added new peer connection to channel")
27+
}
28+
})
29+
30+
for {
31+
msg, err := p.ReadMessage()
32+
if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseAbnormalClosure) {
33+
log.Printf("Websocket closed: %v", err)
34+
break
35+
} else if err != nil {
36+
log.Printf("Error reading from websocket: %v", err)
37+
continue
38+
}
39+
40+
switch msg.Type {
41+
case OFFER:
42+
offer := msg.Value.(string)
43+
err = p.CreateOffer(offer)
44+
if err != nil {
45+
log.Printf("Error when creating offer: %v", err)
46+
continue
47+
}
48+
49+
case ICE:
50+
ice := msg.Value.(string)
51+
err = p.AddICECandidate(ice)
52+
if err != nil {
53+
log.Printf("Error when adding ICE: %v", err)
54+
continue
55+
}
56+
57+
default:
58+
log.Printf("Unknown message type: %+v", msg)
59+
continue
60+
}
61+
}
62+
}

‎013-cloud-gaming/service/webrtc.go

+83
Original file line numberDiff line numberDiff line change
@@ -1 +1,84 @@
11
package service
2+
3+
import (
4+
"sync"
5+
6+
"github.com/gorilla/websocket"
7+
"github.com/pion/webrtc/v4"
8+
)
9+
10+
type MessageType string
11+
12+
const (
13+
OFFER MessageType = "offer"
14+
ICE MessageType = "ice"
15+
ANSWER MessageType = "answer"
16+
ERROR MessageType = "error"
17+
)
18+
19+
type Message struct {
20+
Type MessageType `json:"type"`
21+
Value any `json:"value"`
22+
}
23+
24+
type PeerConnState struct {
25+
peerConnection *webrtc.PeerConnection
26+
ws *threadSafeWriter
27+
}
28+
29+
func NewPeerConnection(peerConnection *webrtc.PeerConnection, ws *threadSafeWriter) *PeerConnState {
30+
return &PeerConnState{peerConnection, ws}
31+
}
32+
33+
func (p *PeerConnState) PeerConnection() *webrtc.PeerConnection {
34+
return p.peerConnection
35+
}
36+
37+
func (p *PeerConnState) WS() *threadSafeWriter {
38+
return p.ws
39+
}
40+
41+
func (p *PeerConnState) ReadMessage() (Message, error) {
42+
var msg Message
43+
err := p.ws.ReadJSON(&msg)
44+
return msg, err
45+
}
46+
47+
func (p *PeerConnState) CreateOffer(offer string) error {
48+
err := p.PeerConnection().SetRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: offer})
49+
if err != nil {
50+
return err
51+
}
52+
53+
answer, err := p.PeerConnection().CreateAnswer(nil)
54+
if err != nil {
55+
return err
56+
}
57+
58+
err = p.PeerConnection().SetLocalDescription(answer)
59+
if err != nil {
60+
return err
61+
}
62+
63+
err = p.WS().WriteJSON(ANSWER, answer.SDP)
64+
return err
65+
}
66+
67+
func (p *PeerConnState) AddICECandidate(ice string) error {
68+
return p.peerConnection.AddICECandidate(webrtc.ICECandidateInit{Candidate: ice})
69+
}
70+
71+
type threadSafeWriter struct {
72+
*websocket.Conn
73+
sync.Mutex
74+
}
75+
76+
func NewSafeWriter(unsafeConn *websocket.Conn) *threadSafeWriter {
77+
return &threadSafeWriter{unsafeConn, sync.Mutex{}}
78+
}
79+
80+
func (t *threadSafeWriter) WriteJSON(mt MessageType, value string) error {
81+
t.Lock()
82+
defer t.Unlock()
83+
return t.Conn.WriteJSON(Message{mt, value})
84+
}

‎013-cloud-gaming/service/worker.go

+19
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
package service
2+
3+
import "log"
4+
5+
func NewWorker() *Worker {
6+
return &Worker{}
7+
}
8+
9+
type Worker struct {
10+
}
11+
12+
func (w *Worker) Run() {
13+
go func() {
14+
for p := range peerConnCh {
15+
peerConns = append(peerConns, p)
16+
log.Println("Added to connections")
17+
}
18+
}()
19+
}

‎013-cloud-gaming/transport/ssr.go

-9
This file was deleted.

‎013-cloud-gaming/transport/websocket.go

+17-84
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
package transport
22

33
import (
4+
"cloud-gaming/service"
45
"log"
56
"net/http"
6-
"sync"
77

88
"github.com/gorilla/websocket"
99
"github.com/pion/webrtc/v4"
@@ -15,104 +15,37 @@ var (
1515
WriteBufferSize: 1024,
1616
CheckOrigin: func(r *http.Request) bool { return true },
1717
}
18-
// peerConns []*PeerConnState
19-
peerConnCh = make(chan *PeerConnState)
2018
)
2119

22-
func WebsocketHandler(w http.ResponseWriter, r *http.Request) {
20+
const STUN = "stun:stun.l.google.com:19302"
21+
22+
func NewHandler(svc service.Service) *Handler {
23+
return &Handler{svc}
24+
}
25+
26+
type Handler struct {
27+
svc service.Service
28+
}
29+
30+
func (h *Handler) Websocket(w http.ResponseWriter, r *http.Request) {
2331
unsafeConn, err := upgrader.Upgrade(w, r, nil)
2432
if err != nil {
2533
log.Fatal(err)
2634
}
2735

28-
conn := &threadSafeWriter{unsafeConn, sync.Mutex{}}
36+
conn := service.NewSafeWriter(unsafeConn)
2937
pc, err := webrtc.NewPeerConnection(webrtc.Configuration{
3038
ICEServers: []webrtc.ICEServer{
3139
{
32-
URLs: []string{"stun:stun.l.google.com:19302"},
40+
URLs: []string{STUN},
3341
},
3442
},
3543
})
3644
if err != nil {
37-
conn.WriteJSON(Message{Type: "error", Value: "Error creating peer connection"})
45+
conn.WriteJSON(service.ERROR, "Error creating peer connection")
3846
return
3947
}
4048

41-
pcs := &PeerConnState{peerConnection: pc, ws: conn}
42-
pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
43-
log.Println("Connection state changed:", state)
44-
if state == webrtc.PeerConnectionStateConnected {
45-
pcs = &PeerConnState{peerConnection: pc, ws: conn}
46-
peerConnCh <- pcs
47-
log.Println("Added new peer connection to channel")
48-
}
49-
})
50-
51-
go handleMessage(pcs)
52-
}
53-
54-
func handleMessage(pcs *PeerConnState) {
55-
for {
56-
var msg Message
57-
err := pcs.ws.ReadJSON(&msg)
58-
if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseAbnormalClosure) {
59-
log.Printf("Websocket closed: %v", err)
60-
break
61-
} else if err != nil {
62-
log.Printf("Error reading from websocket: %v", err)
63-
continue
64-
}
65-
66-
switch msg.Type {
67-
case "offer":
68-
offer := msg.Value.(string)
69-
err = pcs.peerConnection.SetRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: offer})
70-
if err != nil {
71-
log.Printf("Error reading from websocket: %v", err)
72-
continue
73-
}
74-
75-
answer, err := pcs.peerConnection.CreateAnswer(nil)
76-
if err != nil {
77-
log.Printf("Error reading from websocket: %v", err)
78-
continue
79-
}
80-
81-
err = pcs.peerConnection.SetLocalDescription(answer)
82-
if err != nil {
83-
log.Printf("Error reading from websocket: %v", err)
84-
continue
85-
}
86-
pcs.ws.WriteJSON(Message{Type: "answer", Value: answer.SDP})
87-
88-
case "ice":
89-
ice := msg.Value.(string)
90-
pcs.peerConnection.AddICECandidate(webrtc.ICECandidateInit{Candidate: ice})
91-
92-
default:
93-
log.Printf("Unknown message type: %+v", msg)
94-
continue
95-
}
96-
}
97-
}
98-
99-
type Message struct {
100-
Type string `json:"type"`
101-
Value any `json:"value"`
102-
}
103-
104-
type PeerConnState struct {
105-
peerConnection *webrtc.PeerConnection
106-
ws *threadSafeWriter
107-
}
108-
109-
type threadSafeWriter struct {
110-
*websocket.Conn
111-
sync.Mutex
112-
}
113-
114-
func (t *threadSafeWriter) WriteJSON(v interface{}) error {
115-
t.Lock()
116-
defer t.Unlock()
117-
return t.Conn.WriteJSON(v)
49+
p := service.NewPeerConnection(pc, conn)
50+
go h.svc.HandleMessage(p)
11851
}

0 commit comments

Comments
 (0)