package main import ( "bytes" "context" "encoding/json" "flag" "io" "log" "strings" "time" proto "git.jeffthecoder.xyz/guochao/meow-signaling.jeffthecoder.xyz/pkg/proto/signal-server" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "github.com/pion/webrtc/v3" ) func main() { flag.Parse() if flag.NArg() != 2 { panic("invalid usage") } room := flag.Arg(0) clientId := flag.Arg(1) log.Println("dialing...") client, err := grpc.Dial("127.0.0.1:4444", grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { panic(err) } defer client.Close() log.Println("connecting...client id: ", clientId) signal_server := proto.NewSignalingClient(client) stream, err := signal_server.Connect(context.Background()) if err != nil { panic(err) } log.Println("connected. discovering ", clientId, " -> ", room) stream.Send(&proto.SignalingMessage{ Room: room, Sender: clientId, Message: &proto.SignalingMessage_Bootstrap{}, }) webrtcConfig := webrtc.Configuration{ ICEServers: []webrtc.ICEServer{ { URLs: []string{"stun:nhz.jeffthecoder.xyz:3478", "stun:nhz.jeffthecoder.xyz:3479"}, }, }, } connections := make(map[string]*webrtc.PeerConnection) channels := make(map[string]*webrtc.DataChannel) for { msg, err := stream.Recv() if err == io.EOF { break } switch inner := msg.Message.(type) { case *proto.SignalingMessage_Bootstrap: stream.Send(&proto.SignalingMessage{ Room: room, Sender: clientId, Message: &proto.SignalingMessage_DiscoverRequest{}, }) case *proto.SignalingMessage_DiscoverRequest: time.Sleep(time.Second * 3) log.Println("received discover request from ", msg.Sender, ", responding") stream.Send(&proto.SignalingMessage{ Room: room, Sender: clientId, Receiver: &msg.Sender, Message: &proto.SignalingMessage_DiscoverResponse{}, }) case *proto.SignalingMessage_DiscoverResponse: log.Println("received discover response from ", msg.Sender, ", offering") peerConnection, ok := connections[msg.Sender] if !ok { pc, err := webrtc.NewPeerConnection(webrtcConfig) if err != nil { log.Println("failed to create peer connection: ", err) continue } pc.OnConnectionStateChange(func(pcs webrtc.PeerConnectionState) { log.Printf("Peer Connection(%v) State has changed: %s\n", msg.Sender, pcs) }) pc.OnICEConnectionStateChange(func(is webrtc.ICEConnectionState) { log.Printf("ICE Connection(%v) State has changed: %s\n", msg.Sender, is) }) pc.OnICECandidate(func(i *webrtc.ICECandidate) { log.Printf("ICE Candidate for %v: %s\n", msg.Sender, i) }) dataChannel, err := pc.CreateDataChannel("chan", nil) if err != nil { log.Println("failed to create answer: ", err) continue } channels[msg.Sender] = dataChannel dataChannel.OnOpen(func() { log.Println("data channel opened") }) peerConnection = pc } sdp, err := peerConnection.CreateOffer(&webrtc.OfferOptions{}) if err != nil { log.Println("failed to create offer: ", err) peerConnection.Close() continue } log.Print("set local: ", sdp) peerConnection.SetLocalDescription(sdp) buffer := &bytes.Buffer{} if err := json.NewEncoder(buffer).Encode(sdp); err != nil { log.Println("failed to encode offer: ", err) peerConnection.Close() continue } connections[msg.Sender] = peerConnection stream.Send(&proto.SignalingMessage{ Room: room, Sender: clientId, Receiver: &msg.Sender, Message: &proto.SignalingMessage_SessionOffer{ SessionOffer: &proto.SDPMessage{ SDP: buffer.String(), Type: proto.SDPMessageType_Data, Sender: clientId, }, }, }) defer peerConnection.Close() case *proto.SignalingMessage_SessionOffer: log.Println("received session offer: ", inner.SessionOffer.SDP) peerConnection, ok := connections[msg.Sender] if !ok { pc, err := webrtc.NewPeerConnection(webrtcConfig) if err != nil { log.Println("failed to create peer connection: ", err) continue } pc.OnConnectionStateChange(func(pcs webrtc.PeerConnectionState) { log.Printf("Peer Connection(%v) State has changed: %s\n", msg.Sender, pcs) }) pc.OnICEConnectionStateChange(func(is webrtc.ICEConnectionState) { log.Printf("ICE Connection(%v) State has changed: %s\n", msg.Sender, is) }) pc.OnICECandidate(func(i *webrtc.ICECandidate) { log.Printf("ICE Candidate for %v: %s\n", msg.Sender, i) }) pc.OnDataChannel(func(dc *webrtc.DataChannel) { log.Println("DataChannel ", dc.Label()) channels[msg.Sender] = dc }) connections[msg.Sender] = pc peerConnection = pc } var offer webrtc.SessionDescription if err := json.NewDecoder(strings.NewReader(inner.SessionOffer.SDP)).Decode(&offer); err != nil { log.Println("failed to decode offer: ", err) continue } log.Println("set remote: ", offer) peerConnection.SetRemoteDescription(offer) sdp, err := peerConnection.CreateAnswer(nil) if err != nil { log.Println("failed to create answer: ", err) continue } log.Println("set local: ", sdp) peerConnection.SetLocalDescription(sdp) gatherComplete := webrtc.GatheringCompletePromise(peerConnection) <-gatherComplete buffer := &bytes.Buffer{} if err := json.NewEncoder(buffer).Encode(peerConnection.LocalDescription()); err != nil { log.Println("failed to encode answer: ", err) continue } log.Println("answering: ", buffer.String()) stream.Send(&proto.SignalingMessage{ Room: room, Sender: clientId, Receiver: &msg.Sender, Message: &proto.SignalingMessage_SessionAnswer{ SessionAnswer: &proto.SDPMessage{ SDP: buffer.String(), Type: proto.SDPMessageType_Data, Sender: clientId, }, }, }) case *proto.SignalingMessage_SessionAnswer: log.Println("received session anser: ", inner.SessionAnswer.SDP) peerConnection, ok := connections[msg.Sender] if !ok { log.Println("no connection found. there might be some mistakes") continue } var answer webrtc.SessionDescription if err := json.NewDecoder(strings.NewReader(inner.SessionAnswer.SDP)).Decode(&answer); err != nil { log.Println("failed to decode answer: ", err) continue } log.Println("set remote: ", answer) peerConnection.SetRemoteDescription(answer) gatherComplete := webrtc.GatheringCompletePromise(peerConnection) <-gatherComplete } } }