public/chat-signaling-server
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

reorganize the code

Browse Source
This commit is contained in:
guochao 2023-03-05 23:16:46 +08:00
parent 61ad7fed26
commit e59276b2a8
Signed by: guochao
GPG Key ID: 79F7306D2AA32FC3
1 changed files with 193 additions and 373 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

566
cmd/demo-signal-client/main.go
View File

@ -8,6 +8,7 @@ import (
"fmt"
"io"
"strings"
"sync"
proto "git.jeffthecoder.xyz/guochao/meow-signaling.jeffthecoder.xyz/pkg/proto/signal-server"
"github.com/charmbracelet/bubbles/textarea"
@ -47,6 +48,8 @@ type SignalClient struct {
UI SignalClientUI
Program *tea.Program
webrtcConfig webrtc.Configuration
Room string
Name string
@ -61,6 +64,7 @@ type SignalClient struct {
PeerConns map[string]*webrtc.PeerConnection
Channels map[string]*webrtc.DataChannel
Lock sync.Locker
}
func New(room, name string) *SignalClient {
@ -68,7 +72,7 @@ func New(room, name string) *SignalClient {
ta.Prompt = "Send a message"
ta.Focus()
ta.Prompt = " | "
ta.Prompt = " ! "
ta.ShowLineNumbers = false
ta.SetHeight(1)
@ -86,11 +90,20 @@ Type a message and press Enter to send.`)
textarea: ta,
},
webrtcConfig: webrtc.Configuration{
ICEServers: []webrtc.ICEServer{
{
URLs: []string{"stun:nhz.jeffthecoder.xyz:3478", "stun:nhz.jeffthecoder.xyz:3479"},
},
},
},
Room: room,
Name: name,
PeerConns: make(map[string]*webrtc.PeerConnection),
Channels: make(map[string]*webrtc.DataChannel),
Lock: &sync.Mutex{},
}
return client
@ -162,7 +175,7 @@ func (client *SignalClient) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
if client.MessageIsValid {
client.UI.textarea.Prompt = " | "
} else {
client.UI.textarea.Prompt = " ? "
client.UI.textarea.Prompt = " ! "
}
return client, tea.Batch(tiCmd, vpCmd)
@ -208,14 +221,6 @@ func (client *SignalClient) HandleConnection(ctx context.Context, grpcClient *gr
client.Program.Send(systemMsg("Bootstrapped."))
webrtcConfig := webrtc.Configuration{
ICEServers: []webrtc.ICEServer{
{
URLs: []string{"stun:nhz.jeffthecoder.xyz:3478", "stun:nhz.jeffthecoder.xyz:3479"},
},
},
}
for {
msg, err := stream.Recv()
if err == io.EOF {
@ -223,176 +228,200 @@ func (client *SignalClient) HandleConnection(ctx context.Context, grpcClient *gr
}
switch inner := msg.Message.(type) {
case *proto.SignalingMessage_Bootstrap:
stream.Send(&proto.SignalingMessage{
Room: room,
Sender: clientId,
Message: &proto.SignalingMessage_DiscoverRequest{},
})
client.OnBootstrapReady(ctx, stream, room, clientId)
case *proto.SignalingMessage_DiscoverRequest:
stream.Send(&proto.SignalingMessage{
Room: room,
Sender: clientId,
Receiver: &msg.Sender,
Message: &proto.SignalingMessage_DiscoverResponse{},
})
client.OnDiscoverRequest(ctx, stream, room, clientId, msg.Sender)
case *proto.SignalingMessage_DiscoverResponse:
peerConnection, ok := client.PeerConns[msg.Sender]
if !ok {
pc, err := webrtc.NewPeerConnection(webrtcConfig)
if err != nil {
continue
}
dataChannel, err := pc.CreateDataChannel("chan", nil)
if err != nil {
client.Program.Send(systemMsg(fmt.Sprint("failed to create answer: ", err)))
continue
}
client.Channels[msg.Sender] = dataChannel
dataChannel.OnOpen(func() {
client.Program.Send(systemMsg(fmt.Sprint("Connected to client: ", msg.Sender)))
})
dataChannel.OnMessage(func(dcMsg webrtc.DataChannelMessage) {
if dcMsg.IsString {
client.Program.Send(peerMsg{
Peer: msg.Sender,
Message: string(dcMsg.Data),
})
}
})
peerConnection = pc
}
sdp, err := peerConnection.CreateOffer(&webrtc.OfferOptions{})
if err != nil {
client.Program.Send(systemMsg(fmt.Sprint("Failed to create offer for peer "+msg.Sender+": ", err)))
peerConnection.Close()
continue
}
peerConnection.SetLocalDescription(sdp)
buffer := &bytes.Buffer{}
if err := json.NewEncoder(buffer).Encode(sdp); err != nil {
client.Program.Send(systemMsg(fmt.Sprint("Failed to encode offer for peer "+msg.Sender+": ", err)))
peerConnection.Close()
continue
}
client.PeerConns[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()
client.OnDiscoverResponse(ctx, stream, room, clientId, msg.Sender)
case *proto.SignalingMessage_SessionOffer:
peerConnection, ok := client.PeerConns[msg.Sender]
if !ok {
pc, err := webrtc.NewPeerConnection(webrtcConfig)
if err != nil {
client.Program.Send(systemMsg(fmt.Sprint("Failed to create peer connection for peer "+msg.Sender+": ", err)))
continue
}
pc.OnDataChannel(func(dc *webrtc.DataChannel) {
client.Program.Send(systemMsg(fmt.Sprint("Connected to peer " + msg.Sender + ": " + dc.Label())))
client.Channels[msg.Sender] = dc
dc.OnMessage(func(dcMsg webrtc.DataChannelMessage) {
if dcMsg.IsString {
client.Program.Send(peerMsg{
Peer: msg.Sender,
Message: string(dcMsg.Data),
})
}
})
})
client.PeerConns[msg.Sender] = pc
peerConnection = pc
}
var offer webrtc.SessionDescription
if err := json.NewDecoder(strings.NewReader(inner.SessionOffer.SDP)).Decode(&offer); err != nil {
client.Program.Send(systemMsg(fmt.Sprint("Failed to decode offer for peer"+msg.Sender+": ", err)))
continue
}
peerConnection.SetRemoteDescription(offer)
sdp, err := peerConnection.CreateAnswer(nil)
if err != nil {
client.Program.Send(systemMsg(fmt.Sprint("Failed to create answer for peer "+msg.Sender+": ", err)))
continue
}
peerConnection.SetLocalDescription(sdp)
gatherComplete := webrtc.GatheringCompletePromise(peerConnection)
<-gatherComplete
buffer := &bytes.Buffer{}
if err := json.NewEncoder(buffer).Encode(peerConnection.LocalDescription()); err != nil {
client.Program.Send(systemMsg(fmt.Sprint("Failed to encode answer for peer"+msg.Sender+": ", err)))
continue
}
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,
},
},
})
client.OnOffer(ctx, stream, room, clientId, msg.Sender, inner.SessionOffer.SDP)
case *proto.SignalingMessage_SessionAnswer:
peerConnection, ok := client.PeerConns[msg.Sender]
if !ok {
continue
}
var answer webrtc.SessionDescription
if err := json.NewDecoder(strings.NewReader(inner.SessionAnswer.SDP)).Decode(&answer); err != nil {
client.Program.Send(systemMsg(fmt.Sprint("Failed to decode answer for peer"+msg.Sender+": ", err)))
continue
}
peerConnection.SetRemoteDescription(answer)
gatherComplete := webrtc.GatheringCompletePromise(peerConnection)
<-gatherComplete
client.OnAnswer(ctx, msg.Sender, inner.SessionAnswer.SDP)
}
}
}
func (client *SignalClient) OnBootstrapReady(ctx context.Context) error {
panic("not implemented")
func (client *SignalClient) OnBootstrapReady(ctx context.Context, stream proto.Signaling_ConnectClient, room, name string) {
client.Program.Send(systemMsg("Server ready!"))
stream.Send(&proto.SignalingMessage{
Room: room,
Sender: name,
Message: &proto.SignalingMessage_DiscoverRequest{},
})
}
func (client *SignalClient) OnDiscoverRequest(ctx context.Context, sender string) error {
panic("not implemented")
func (client *SignalClient) OnDiscoverRequest(ctx context.Context, stream proto.Signaling_ConnectClient, room, name, sender string) {
client.Program.Send(systemMsg("Client " + sender + " is joining into the room " + room))
stream.Send(&proto.SignalingMessage{
Room: room,
Sender: name,
Receiver: &sender,
Message: &proto.SignalingMessage_DiscoverResponse{},
})
}
func (client *SignalClient) OnDiscoverResponse(ctx context.Context, sender string) error {
panic("not implemented")
func (client *SignalClient) OnDiscoverResponse(ctx context.Context, stream proto.Signaling_ConnectClient, room, name, sender string) {
client.Program.Send(systemMsg("Client " + sender + " ponged"))
peerConnection, err := client.GetOrCreatePeerConnection(sender)
if err != nil {
return
}
dataChannel, err := peerConnection.CreateDataChannel("chan", nil)
if err != nil {
client.Program.Send(systemMsg(fmt.Sprint("failed to create answer: ", err)))
return
}
client.Channels[sender] = dataChannel
client.SetupDataChannel(peerConnection, dataChannel, sender)
sdp, err := peerConnection.CreateOffer(&webrtc.OfferOptions{})
if err != nil {
client.Program.Send(systemMsg(fmt.Sprint("Failed to create offer for peer "+sender+": ", err)))
peerConnection.Close()
return
}
peerConnection.SetLocalDescription(sdp)
buffer := &bytes.Buffer{}
if err := json.NewEncoder(buffer).Encode(sdp); err != nil {
client.Program.Send(systemMsg(fmt.Sprint("Failed to encode offer for peer "+sender+": ", err)))
peerConnection.Close()
return
}
client.PeerConns[sender] = peerConnection
stream.Send(&proto.SignalingMessage{
Room: room,
Sender: name,
Receiver: &sender,
Message: &proto.SignalingMessage_SessionOffer{
SessionOffer: &proto.SDPMessage{
SDP: buffer.String(),
Type: proto.SDPMessageType_Data,
Sender: name,
},
},
})
}
func (client *SignalClient) OnOffer(ctx context.Context, sender, sdp string) error {
panic("not implemented")
func (client *SignalClient) OnOffer(ctx context.Context, stream proto.Signaling_ConnectClient, room, name, sender, sdp string) {
client.Program.Send(systemMsg("Client " + sender + " is offering"))
peerConnection, err := client.GetOrCreatePeerConnection(sender)
if err != nil {
return
}
var offer webrtc.SessionDescription
if err := json.NewDecoder(strings.NewReader(sdp)).Decode(&offer); err != nil {
client.Program.Send(systemMsg(fmt.Sprint("Failed to decode offer for peer"+sender+": ", err)))
return
}
peerConnection.SetRemoteDescription(offer)
answer, err := peerConnection.CreateAnswer(nil)
if err != nil {
client.Program.Send(systemMsg(fmt.Sprint("Failed to create answer for peer "+sender+": ", err)))
return
}
peerConnection.SetLocalDescription(answer)
gatherComplete := webrtc.GatheringCompletePromise(peerConnection)
<-gatherComplete
buffer := &bytes.Buffer{}
if err := json.NewEncoder(buffer).Encode(peerConnection.LocalDescription()); err != nil {
client.Program.Send(systemMsg(fmt.Sprint("Failed to encode answer for peer"+sender+": ", err)))
return
}
stream.Send(&proto.SignalingMessage{
Room: room,
Sender: name,
Receiver: &sender,
Message: &proto.SignalingMessage_SessionAnswer{
SessionAnswer: &proto.SDPMessage{
SDP: buffer.String(),
Type: proto.SDPMessageType_Data,
Sender: name,
},
},
})
}
func (client *SignalClient) OnAnswer(ctx context.Context, sender, sdp string) error {
panic("not implemented")
func (client *SignalClient) OnAnswer(ctx context.Context, sender, sdp string) {
client.Program.Send(systemMsg("Client " + sender + " has answered the offer"))
peerConnection, ok := client.PeerConns[sender]
if !ok {
return
}
var answer webrtc.SessionDescription
if err := json.NewDecoder(strings.NewReader(sdp)).Decode(&answer); err != nil {
client.Program.Send(systemMsg(fmt.Sprint("Failed to decode answer for peer"+sender+": ", err)))
return
}
peerConnection.SetRemoteDescription(answer)
}
func (client SignalClient) GetOrCreatePeerConnection(sender string) (*webrtc.PeerConnection, error) {
client.Lock.Lock()
defer client.Lock.Unlock()
peerConnection, ok := client.PeerConns[sender]
if ok {
return peerConnection, nil
}
peerConnection, err := webrtc.NewPeerConnection(client.webrtcConfig)
if err != nil {
return nil, err
}
client.PeerConns[sender] = peerConnection
closeOnceAndNoMore := &sync.Once{}
peerConnection.OnConnectionStateChange(func(pcs webrtc.PeerConnectionState) {
if pcs == webrtc.PeerConnectionStateDisconnected || pcs == webrtc.PeerConnectionStateClosed || pcs == webrtc.PeerConnectionStateFailed {
closeOnceAndNoMore.Do(func() {
if peerConnection != nil {
client.Program.Send(systemMsg(fmt.Sprint("Closing connection with ", sender, " for state changed to: ", pcs.String())))
peerConnection.Close()
delete(client.Channels, sender)
delete(client.PeerConns, sender)
peerConnection = nil
}
})
}
})
peerConnection.OnDataChannel(func(dc *webrtc.DataChannel) {
client.Program.Send(systemMsg(fmt.Sprint("Connected to client " + sender + ": " + dc.Label())))
client.Channels[sender] = dc
client.SetupDataChannel(peerConnection, dc, sender)
})
return peerConnection, nil
}
func (client *SignalClient) SetupDataChannel(pc *webrtc.PeerConnection, dc *webrtc.DataChannel, sender string) {
dc.OnOpen(func() {
client.Program.Send(systemMsg(fmt.Sprint("Channel opened: ", sender)))
})
dc.OnMessage(func(dcMsg webrtc.DataChannelMessage) {
if dcMsg.IsString {
client.Program.Send(peerMsg{
Peer: sender,
Message: string(dcMsg.Data),
})
}
})
}
func main() {
@ -409,213 +438,4 @@ func main() {
if _, err := signalClient.Program.Run(); err != nil {
panic("err")
}
// 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
// }
// }
}