From 0b00b26d056f14bc600b8c2c8b68d28fe1ce279e Mon Sep 17 00:00:00 2001 From: David Zhao Date: Tue, 19 Jan 2021 16:18:03 -0800 Subject: [PATCH] terminate RTC sessions properly --- pkg/routing/interfaces.go | 5 +- pkg/routing/messagechannel.go | 11 +- .../routingfakes/fake_message_source.go | 93 +++++++------ .../routingfakes/fake_node_selector.go | 124 ++++++++++++++++++ pkg/rtc/participant.go | 4 - pkg/rtc/room.go | 3 + pkg/rtc/types/interfaces.go | 2 +- .../types/typesfakes/fake_peer_connection.go | 39 ------ pkg/service/roommanager.go | 109 ++++++++------- pkg/service/rtcservice.go | 38 +++--- 10 files changed, 254 insertions(+), 174 deletions(-) create mode 100644 pkg/routing/routingfakes/fake_node_selector.go diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index 409ab3070..ec34e9e54 100644 --- a/pkg/routing/interfaces.go +++ b/pkg/routing/interfaces.go @@ -19,7 +19,8 @@ type MessageSink interface { //counterfeiter:generate . MessageSource type MessageSource interface { - ReadMessage() (proto.Message, error) + // source exposes a one way channel to make it easier to use with select + ReadChan() <-chan proto.Message } type ParticipantCallback func(roomId, participantId, participantName string, requestSource MessageSource, responseSink MessageSink) @@ -48,7 +49,7 @@ type Router interface { } // NodeSelector selects an appropriate node to run the current session -//counterfeiter:generate . NodeBalancer +//counterfeiter:generate . NodeSelector type NodeSelector interface { SelectNode(nodes []*livekit.Node, room *livekit.Room) (*livekit.Node, error) } diff --git a/pkg/routing/messagechannel.go b/pkg/routing/messagechannel.go index 285dacfdb..0a635a069 100644 --- a/pkg/routing/messagechannel.go +++ b/pkg/routing/messagechannel.go @@ -1,8 +1,6 @@ package routing import ( - "io" - "google.golang.org/protobuf/proto" ) @@ -26,13 +24,8 @@ func (m *MessageChannel) WriteMessage(msg proto.Message) error { return nil } -func (m *MessageChannel) ReadMessage() (proto.Message, error) { - msg := <-m.msgChan - // channel closed - if msg == nil { - return nil, io.EOF - } - return msg, nil +func (m *MessageChannel) ReadChan() <-chan proto.Message { + return m.msgChan } func (m *MessageChannel) Close() { diff --git a/pkg/routing/routingfakes/fake_message_source.go b/pkg/routing/routingfakes/fake_message_source.go index 4f2cd28a1..fe39c5c25 100644 --- a/pkg/routing/routingfakes/fake_message_source.go +++ b/pkg/routing/routingfakes/fake_message_source.go @@ -9,83 +9,78 @@ import ( ) type FakeMessageSource struct { - ReadMessageStub func() (protoreflect.ProtoMessage, error) - readMessageMutex sync.RWMutex - readMessageArgsForCall []struct { + ReadChanStub func() <-chan protoreflect.ProtoMessage + readChanMutex sync.RWMutex + readChanArgsForCall []struct { } - readMessageReturns struct { - result1 protoreflect.ProtoMessage - result2 error + readChanReturns struct { + result1 <-chan protoreflect.ProtoMessage } - readMessageReturnsOnCall map[int]struct { - result1 protoreflect.ProtoMessage - result2 error + readChanReturnsOnCall map[int]struct { + result1 <-chan protoreflect.ProtoMessage } invocations map[string][][]interface{} invocationsMutex sync.RWMutex } -func (fake *FakeMessageSource) ReadMessage() (protoreflect.ProtoMessage, error) { - fake.readMessageMutex.Lock() - ret, specificReturn := fake.readMessageReturnsOnCall[len(fake.readMessageArgsForCall)] - fake.readMessageArgsForCall = append(fake.readMessageArgsForCall, struct { +func (fake *FakeMessageSource) ReadChan() <-chan protoreflect.ProtoMessage { + fake.readChanMutex.Lock() + ret, specificReturn := fake.readChanReturnsOnCall[len(fake.readChanArgsForCall)] + fake.readChanArgsForCall = append(fake.readChanArgsForCall, struct { }{}) - stub := fake.ReadMessageStub - fakeReturns := fake.readMessageReturns - fake.recordInvocation("ReadMessage", []interface{}{}) - fake.readMessageMutex.Unlock() + stub := fake.ReadChanStub + fakeReturns := fake.readChanReturns + fake.recordInvocation("ReadChan", []interface{}{}) + fake.readChanMutex.Unlock() if stub != nil { return stub() } if specificReturn { - return ret.result1, ret.result2 + return ret.result1 } - return fakeReturns.result1, fakeReturns.result2 + return fakeReturns.result1 } -func (fake *FakeMessageSource) ReadMessageCallCount() int { - fake.readMessageMutex.RLock() - defer fake.readMessageMutex.RUnlock() - return len(fake.readMessageArgsForCall) +func (fake *FakeMessageSource) ReadChanCallCount() int { + fake.readChanMutex.RLock() + defer fake.readChanMutex.RUnlock() + return len(fake.readChanArgsForCall) } -func (fake *FakeMessageSource) ReadMessageCalls(stub func() (protoreflect.ProtoMessage, error)) { - fake.readMessageMutex.Lock() - defer fake.readMessageMutex.Unlock() - fake.ReadMessageStub = stub +func (fake *FakeMessageSource) ReadChanCalls(stub func() <-chan protoreflect.ProtoMessage) { + fake.readChanMutex.Lock() + defer fake.readChanMutex.Unlock() + fake.ReadChanStub = stub } -func (fake *FakeMessageSource) ReadMessageReturns(result1 protoreflect.ProtoMessage, result2 error) { - fake.readMessageMutex.Lock() - defer fake.readMessageMutex.Unlock() - fake.ReadMessageStub = nil - fake.readMessageReturns = struct { - result1 protoreflect.ProtoMessage - result2 error - }{result1, result2} +func (fake *FakeMessageSource) ReadChanReturns(result1 <-chan protoreflect.ProtoMessage) { + fake.readChanMutex.Lock() + defer fake.readChanMutex.Unlock() + fake.ReadChanStub = nil + fake.readChanReturns = struct { + result1 <-chan protoreflect.ProtoMessage + }{result1} } -func (fake *FakeMessageSource) ReadMessageReturnsOnCall(i int, result1 protoreflect.ProtoMessage, result2 error) { - fake.readMessageMutex.Lock() - defer fake.readMessageMutex.Unlock() - fake.ReadMessageStub = nil - if fake.readMessageReturnsOnCall == nil { - fake.readMessageReturnsOnCall = make(map[int]struct { - result1 protoreflect.ProtoMessage - result2 error +func (fake *FakeMessageSource) ReadChanReturnsOnCall(i int, result1 <-chan protoreflect.ProtoMessage) { + fake.readChanMutex.Lock() + defer fake.readChanMutex.Unlock() + fake.ReadChanStub = nil + if fake.readChanReturnsOnCall == nil { + fake.readChanReturnsOnCall = make(map[int]struct { + result1 <-chan protoreflect.ProtoMessage }) } - fake.readMessageReturnsOnCall[i] = struct { - result1 protoreflect.ProtoMessage - result2 error - }{result1, result2} + fake.readChanReturnsOnCall[i] = struct { + result1 <-chan protoreflect.ProtoMessage + }{result1} } func (fake *FakeMessageSource) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() - fake.readMessageMutex.RLock() - defer fake.readMessageMutex.RUnlock() + fake.readChanMutex.RLock() + defer fake.readChanMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} for key, value := range fake.invocations { copiedInvocations[key] = value diff --git a/pkg/routing/routingfakes/fake_node_selector.go b/pkg/routing/routingfakes/fake_node_selector.go new file mode 100644 index 000000000..9b7f6302b --- /dev/null +++ b/pkg/routing/routingfakes/fake_node_selector.go @@ -0,0 +1,124 @@ +// Code generated by counterfeiter. DO NOT EDIT. +package routingfakes + +import ( + "sync" + + "github.com/livekit/livekit-server/pkg/routing" + "github.com/livekit/livekit-server/proto/livekit" +) + +type FakeNodeSelector struct { + SelectNodeStub func([]*livekit.Node, *livekit.Room) (*livekit.Node, error) + selectNodeMutex sync.RWMutex + selectNodeArgsForCall []struct { + arg1 []*livekit.Node + arg2 *livekit.Room + } + selectNodeReturns struct { + result1 *livekit.Node + result2 error + } + selectNodeReturnsOnCall map[int]struct { + result1 *livekit.Node + result2 error + } + invocations map[string][][]interface{} + invocationsMutex sync.RWMutex +} + +func (fake *FakeNodeSelector) SelectNode(arg1 []*livekit.Node, arg2 *livekit.Room) (*livekit.Node, error) { + var arg1Copy []*livekit.Node + if arg1 != nil { + arg1Copy = make([]*livekit.Node, len(arg1)) + copy(arg1Copy, arg1) + } + fake.selectNodeMutex.Lock() + ret, specificReturn := fake.selectNodeReturnsOnCall[len(fake.selectNodeArgsForCall)] + fake.selectNodeArgsForCall = append(fake.selectNodeArgsForCall, struct { + arg1 []*livekit.Node + arg2 *livekit.Room + }{arg1Copy, arg2}) + stub := fake.SelectNodeStub + fakeReturns := fake.selectNodeReturns + fake.recordInvocation("SelectNode", []interface{}{arg1Copy, arg2}) + fake.selectNodeMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeNodeSelector) SelectNodeCallCount() int { + fake.selectNodeMutex.RLock() + defer fake.selectNodeMutex.RUnlock() + return len(fake.selectNodeArgsForCall) +} + +func (fake *FakeNodeSelector) SelectNodeCalls(stub func([]*livekit.Node, *livekit.Room) (*livekit.Node, error)) { + fake.selectNodeMutex.Lock() + defer fake.selectNodeMutex.Unlock() + fake.SelectNodeStub = stub +} + +func (fake *FakeNodeSelector) SelectNodeArgsForCall(i int) ([]*livekit.Node, *livekit.Room) { + fake.selectNodeMutex.RLock() + defer fake.selectNodeMutex.RUnlock() + argsForCall := fake.selectNodeArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeNodeSelector) SelectNodeReturns(result1 *livekit.Node, result2 error) { + fake.selectNodeMutex.Lock() + defer fake.selectNodeMutex.Unlock() + fake.SelectNodeStub = nil + fake.selectNodeReturns = struct { + result1 *livekit.Node + result2 error + }{result1, result2} +} + +func (fake *FakeNodeSelector) SelectNodeReturnsOnCall(i int, result1 *livekit.Node, result2 error) { + fake.selectNodeMutex.Lock() + defer fake.selectNodeMutex.Unlock() + fake.SelectNodeStub = nil + if fake.selectNodeReturnsOnCall == nil { + fake.selectNodeReturnsOnCall = make(map[int]struct { + result1 *livekit.Node + result2 error + }) + } + fake.selectNodeReturnsOnCall[i] = struct { + result1 *livekit.Node + result2 error + }{result1, result2} +} + +func (fake *FakeNodeSelector) Invocations() map[string][][]interface{} { + fake.invocationsMutex.RLock() + defer fake.invocationsMutex.RUnlock() + fake.selectNodeMutex.RLock() + defer fake.selectNodeMutex.RUnlock() + copiedInvocations := map[string][][]interface{}{} + for key, value := range fake.invocations { + copiedInvocations[key] = value + } + return copiedInvocations +} + +func (fake *FakeNodeSelector) recordInvocation(key string, args []interface{}) { + fake.invocationsMutex.Lock() + defer fake.invocationsMutex.Unlock() + if fake.invocations == nil { + fake.invocations = map[string][][]interface{}{} + } + if fake.invocations[key] == nil { + fake.invocations[key] = [][]interface{}{} + } + fake.invocations[key] = append(fake.invocations[key], args) +} + +var _ routing.NodeSelector = new(FakeNodeSelector) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index ac1af70fd..ad7846fe0 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -133,10 +133,6 @@ func NewParticipant(participantId, name string, pc types.PeerConnection, rs rout } }) - //pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) { - // logger.Debugw("PeerConnection state changed", "state", state.String()) - //}) - pc.OnDataChannel(participant.onDataChannel) // only set after answered diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 900bb6988..a7c3fb487 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -80,6 +80,9 @@ func (r *Room) Join(participant types.Participant) error { } // start the workers once connectivity is established p.Start() + } else if p.State() == livekit.ParticipantInfo_DISCONNECTED { + // remove participant from room + go r.RemoveParticipant(p.ID()) } }) participant.OnTrackUpdated(r.onTrackUpdated) diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index dac67a647..2f2f3b81e 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -24,7 +24,7 @@ type WebsocketClient interface { type PeerConnection interface { OnICECandidate(f func(*webrtc.ICECandidate)) OnICEConnectionStateChange(func(webrtc.ICEConnectionState)) - OnConnectionStateChange(f func(webrtc.PeerConnectionState)) + //OnConnectionStateChange(f func(webrtc.PeerConnectionState)) OnTrack(f func(*webrtc.TrackRemote, *webrtc.RTPReceiver)) OnDataChannel(func(d *webrtc.DataChannel)) OnNegotiationNeeded(f func()) diff --git a/pkg/rtc/types/typesfakes/fake_peer_connection.go b/pkg/rtc/types/typesfakes/fake_peer_connection.go index 86eaac8c2..8abb47508 100644 --- a/pkg/rtc/types/typesfakes/fake_peer_connection.go +++ b/pkg/rtc/types/typesfakes/fake_peer_connection.go @@ -95,11 +95,6 @@ type FakePeerConnection struct { result1 webrtc.SessionDescription result2 error } - OnConnectionStateChangeStub func(func(webrtc.PeerConnectionState)) - onConnectionStateChangeMutex sync.RWMutex - onConnectionStateChangeArgsForCall []struct { - arg1 func(webrtc.PeerConnectionState) - } OnDataChannelStub func(func(d *webrtc.DataChannel)) onDataChannelMutex sync.RWMutex onDataChannelArgsForCall []struct { @@ -608,38 +603,6 @@ func (fake *FakePeerConnection) CreateOfferReturnsOnCall(i int, result1 webrtc.S }{result1, result2} } -func (fake *FakePeerConnection) OnConnectionStateChange(arg1 func(webrtc.PeerConnectionState)) { - fake.onConnectionStateChangeMutex.Lock() - fake.onConnectionStateChangeArgsForCall = append(fake.onConnectionStateChangeArgsForCall, struct { - arg1 func(webrtc.PeerConnectionState) - }{arg1}) - stub := fake.OnConnectionStateChangeStub - fake.recordInvocation("OnConnectionStateChange", []interface{}{arg1}) - fake.onConnectionStateChangeMutex.Unlock() - if stub != nil { - fake.OnConnectionStateChangeStub(arg1) - } -} - -func (fake *FakePeerConnection) OnConnectionStateChangeCallCount() int { - fake.onConnectionStateChangeMutex.RLock() - defer fake.onConnectionStateChangeMutex.RUnlock() - return len(fake.onConnectionStateChangeArgsForCall) -} - -func (fake *FakePeerConnection) OnConnectionStateChangeCalls(stub func(func(webrtc.PeerConnectionState))) { - fake.onConnectionStateChangeMutex.Lock() - defer fake.onConnectionStateChangeMutex.Unlock() - fake.OnConnectionStateChangeStub = stub -} - -func (fake *FakePeerConnection) OnConnectionStateChangeArgsForCall(i int) func(webrtc.PeerConnectionState) { - fake.onConnectionStateChangeMutex.RLock() - defer fake.onConnectionStateChangeMutex.RUnlock() - argsForCall := fake.onConnectionStateChangeArgsForCall[i] - return argsForCall.arg1 -} - func (fake *FakePeerConnection) OnDataChannel(arg1 func(d *webrtc.DataChannel)) { fake.onDataChannelMutex.Lock() fake.onDataChannelArgsForCall = append(fake.onDataChannelArgsForCall, struct { @@ -1119,8 +1082,6 @@ func (fake *FakePeerConnection) Invocations() map[string][][]interface{} { defer fake.createDataChannelMutex.RUnlock() fake.createOfferMutex.RLock() defer fake.createOfferMutex.RUnlock() - fake.onConnectionStateChangeMutex.RLock() - defer fake.onConnectionStateChangeMutex.RUnlock() fake.onDataChannelMutex.RLock() defer fake.onDataChannelMutex.RUnlock() fake.onICECandidateMutex.RLock() diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index e4369d8a3..1d78ea6ae 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -1,7 +1,6 @@ package service import ( - "io" "sync" "time" @@ -157,7 +156,7 @@ func (r *RoomManager) StartSession(roomName, participantId, participantName stri return } - go r.sessionWorker(room, participant, requestSource) + go r.rtcSessionWorker(room, participant, requestSource) } // create the actual room object @@ -189,73 +188,73 @@ func (r *RoomManager) getOrCreateRoom(roomName string) (*rtc.Room, error) { return room, nil } -func (r *RoomManager) sessionWorker(room *rtc.Room, participant types.Participant, requestSource routing.MessageSource) { +// manages a RTC session for a participant, runs on the RTC node +func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Participant, requestSource routing.MessageSource) { defer func() { logger.Debugw("RTC session finishing", "participant", participant.Name(), "room", room.Name, ) - // remove peer from room when participant leaves room - room.RemoveParticipant(participant.ID()) - - // TODO: notify router to cleanup? }() defer rtc.Recover() - for { - obj, err := requestSource.ReadMessage() - if err == io.EOF { - // TODO: when request is EOF, we might be better off requesting a new connection and waiting - // RTC connection terminating should be the only case that we exit - return - } - - req := obj.(*livekit.SignalRequest) - - switch msg := req.Message.(type) { - case *livekit.SignalRequest_Offer: - _, err := participant.Answer(rtc.FromProtoSessionDescription(msg.Offer)) - if err != nil { - logger.Errorw("could not handle join", "err", err, "participant", participant.ID()) + select { + case <-time.After(time.Millisecond * 100): + // periodic check to ensure participant didn't become disconnected + if participant.State() == livekit.ParticipantInfo_DISCONNECTED { return } - case *livekit.SignalRequest_AddTrack: - logger.Debugw("publishing track", "participant", participant.ID(), - "track", msg.AddTrack.Cid) - participant.AddTrack(msg.AddTrack.Cid, msg.AddTrack.Name, msg.AddTrack.Type) - case *livekit.SignalRequest_Answer: - if participant.State() == livekit.ParticipantInfo_JOINING { - logger.Errorw("cannot negotiate before peer offer", "participant", participant.ID()) - //conn.WriteJSON(jsonError(http.StatusNotAcceptable, "cannot negotiate before peer offer")) - return - } - sd := rtc.FromProtoSessionDescription(msg.Answer) - err = participant.HandleAnswer(sd) - if err != nil { - logger.Errorw("could not handle answer", "participant", participant.ID(), "err", err) - //conn.WriteJSON( - // jsonError(http.StatusInternalServerError, "could not handle negotiate", err.Error())) - return - } - case *livekit.SignalRequest_Negotiate: - participant.HandleClientNegotiation() - case *livekit.SignalRequest_Trickle: - if participant.State() == livekit.ParticipantInfo_JOINING { - logger.Errorw("cannot trickle before peer offer", "participant", participant.ID()) - //conn.WriteJSON(jsonError(http.StatusNotAcceptable, "cannot trickle before peer offer")) + case obj := <-requestSource.ReadChan(): + if obj == nil { return } - candidateInit := rtc.FromProtoTrickle(msg.Trickle) - //logger.Debugw("adding peer candidate", "participant", participant.ID()) - if err := participant.AddICECandidate(candidateInit); err != nil { - logger.Errorw("could not handle trickle", "participant", participant.ID(), "err", err) - //conn.WriteJSON( - // jsonError(http.StatusInternalServerError, "could not handle trickle", err.Error())) - return + req := obj.(*livekit.SignalRequest) + + switch msg := req.Message.(type) { + case *livekit.SignalRequest_Offer: + _, err := participant.Answer(rtc.FromProtoSessionDescription(msg.Offer)) + if err != nil { + logger.Errorw("could not handle join", "err", err, "participant", participant.ID()) + return + } + case *livekit.SignalRequest_AddTrack: + logger.Debugw("publishing track", "participant", participant.ID(), + "track", msg.AddTrack.Cid) + participant.AddTrack(msg.AddTrack.Cid, msg.AddTrack.Name, msg.AddTrack.Type) + case *livekit.SignalRequest_Answer: + if participant.State() == livekit.ParticipantInfo_JOINING { + logger.Errorw("cannot negotiate before peer offer", "participant", participant.ID()) + //conn.WriteJSON(jsonError(http.StatusNotAcceptable, "cannot negotiate before peer offer")) + return + } + sd := rtc.FromProtoSessionDescription(msg.Answer) + if err := participant.HandleAnswer(sd); err != nil { + logger.Errorw("could not handle answer", "participant", participant.ID(), "err", err) + //conn.WriteJSON( + // jsonError(http.StatusInternalServerError, "could not handle negotiate", err.Error())) + return + } + case *livekit.SignalRequest_Negotiate: + participant.HandleClientNegotiation() + case *livekit.SignalRequest_Trickle: + if participant.State() == livekit.ParticipantInfo_JOINING { + logger.Errorw("cannot trickle before peer offer", "participant", participant.ID()) + //conn.WriteJSON(jsonError(http.StatusNotAcceptable, "cannot trickle before peer offer")) + return + } + + candidateInit := rtc.FromProtoTrickle(msg.Trickle) + //logger.Debugw("adding peer candidate", "participant", participant.ID()) + if err := participant.AddICECandidate(candidateInit); err != nil { + logger.Errorw("could not handle trickle", "participant", participant.ID(), "err", err) + //conn.WriteJSON( + // jsonError(http.StatusInternalServerError, "could not handle trickle", err.Error())) + return + } + case *livekit.SignalRequest_Mute: + participant.SetTrackMuted(msg.Mute.Sid, msg.Mute.Muted) } - case *livekit.SignalRequest_Mute: - participant.SetTrackMuted(msg.Mute.Sid, msg.Mute.Muted) } } } diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index 3e4b01460..23af27b4b 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -101,30 +101,38 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { handleError(w, http.StatusInternalServerError, "could not get response source"+err.Error()) return } + done := make(chan bool, 1) + defer func() { + logger.Infow("WS connection closed", "participant", pName) + reqSink.Close() + close(done) + }() + // handle responses go func() { for { - msg, err := resSource.ReadMessage() - if err == io.EOF { + select { + case <-done: return - } - res, ok := msg.(*livekit.SignalResponse) - if !ok { - logger.Errorw("unexpected message type", "type", fmt.Sprintf("%T", msg)) - continue - } + case msg := <-resSource.ReadChan(): + if msg == nil { + return + } + res, ok := msg.(*livekit.SignalResponse) + if !ok { + logger.Errorw("unexpected message type", "type", fmt.Sprintf("%T", msg)) + continue + } - if err = sigConn.WriteResponse(res); err != nil { - logger.Warnw("error writing to websocket", "error", err) - return + if err = sigConn.WriteResponse(res); err != nil { + logger.Warnw("error writing to websocket", "error", err) + return + } } } }() - defer func() { - logger.Infow("WS connection closed", "participant", pName) - reqSink.Close() - }() + // handle incoming requests from websocket for { req, err := sigConn.ReadRequest() // normal closure