Add sequence for reliable data message (#692)

Author: cnderrauber

engine.go

@@ -46,11 +46,13 @@ type RTCEngine struct {
 	subscriber *PCTransport
 	client     *SignalClient
 
-	dclock        sync.RWMutex
-	reliableDC    *webrtc.DataChannel
-	lossyDC       *webrtc.DataChannel
-	reliableDCSub *webrtc.DataChannel
-	lossyDCSub    *webrtc.DataChannel
+	dclock          sync.RWMutex
+	reliableDC      *webrtc.DataChannel
+	lossyDC         *webrtc.DataChannel
+	reliableDCSub   *webrtc.DataChannel
+	lossyDCSub      *webrtc.DataChannel
+	reliableMsgLock sync.Mutex
+	reliableMsgSeq  uint32
 
 	trackPublishedListenersLock sync.Mutex
 	trackPublishedListeners     map[string]chan *livekit.TrackPublishedResponse
@@ -107,6 +109,7 @@ func NewRTCEngine() *RTCEngine {
 		client:                  NewSignalClient(),
 		trackPublishedListeners: make(map[string]chan *livekit.TrackPublishedResponse),
 		JoinTimeout:             15 * time.Second,
+		reliableMsgSeq:          1,
 	}
 
 	e.client.OnParticipantUpdate = func(info []*livekit.ParticipantInfo) {
@@ -284,6 +287,12 @@ func (e *RTCEngine) configure(
 	subscriberPrimary *bool) error {
 
 	configuration := e.makeRTCConfiguration(iceServers, clientConfig)
+
+	// reset reliable message sequence
+	e.reliableMsgLock.Lock()
+	e.reliableMsgSeq = 1
+	e.reliableMsgLock.Unlock()
+
 	e.pclock.Lock()
 	defer e.pclock.Unlock()
 
@@ -835,13 +844,7 @@ func (e *RTCEngine) makeRTCConfiguration(iceServers []*livekit.ICEServer, client
 }
 
 func (e *RTCEngine) publishDataPacket(pck *livekit.DataPacket, kind livekit.DataPacket_Kind) error {
-	data, err := proto.Marshal(pck)
-	if err != nil {
-		e.log.Errorw("could not marshal data packet", err)
-		return err
-	}
-
-	err = e.ensurePublisherConnected(true)
+	err := e.ensurePublisherConnected(true)
 	if err != nil {
 		e.log.Errorw("could not ensure publisher connected", err)
 		return err
@@ -853,6 +856,20 @@ func (e *RTCEngine) publishDataPacket(pck *livekit.DataPacket, kind livekit.Data
 		return errors.New("datachannel not found")
 	}
 
+	if kind == livekit.DataPacket_RELIABLE {
+		e.reliableMsgLock.Lock()
+		defer e.reliableMsgLock.Unlock()
+
+		pck.Sequence = e.reliableMsgSeq
+		e.reliableMsgSeq++
+	}
+
+	data, err := proto.Marshal(pck)
+	if err != nil {
+		e.log.Errorw("could not marshal data packet", err)
+		return err
+	}
+
 	dc.Send(data)
 	return nil
 }