terminate RTC sessions properly

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)
 }

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() {

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

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)

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

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)

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())

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()

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)
 		}
 	}
 }

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