datatrack implementation

pkg/rtc/datatrack.go

@@ -0,0 +1,175 @@
+package rtc
+
+import (
+	"sync"
+
+	"github.com/pion/webrtc/v3"
+
+	"github.com/livekit/livekit-server/pkg/logger"
+	"github.com/livekit/livekit-server/pkg/utils"
+	"github.com/livekit/livekit-server/proto/livekit"
+)
+
+// DataTrack wraps a WebRTC DataChannel to satisfy the PublishedTrack interface
+// it shall forward tracks to all of its subscribers
+type DataTrack struct {
+	id            string
+	participantId string
+	dataChannel   *webrtc.DataChannel
+	lock          sync.RWMutex
+	once          sync.Once
+	msgChan       chan livekit.DataMessage
+
+	// map of target participantId -> DownDataChannel
+	subscribers map[string]*DownDataChannel
+}
+
+func NewDataTrack(participantId string, dc *webrtc.DataChannel) *DataTrack {
+	t := &DataTrack{
+		//ctx:           context.Background(),
+		id:            utils.NewGuid(utils.TrackPrefix),
+		participantId: participantId,
+		dataChannel:   dc,
+		msgChan:       make(chan livekit.DataMessage),
+		lock:          sync.RWMutex{},
+		subscribers:   make(map[string]*DownDataChannel),
+	}
+
+	dc.OnMessage(func(msg webrtc.DataChannelMessage) {
+		dm := messageFromDataChannelMessage(msg)
+		t.msgChan <- dm
+	})
+
+	return t
+}
+
+func (t *DataTrack) Start() {
+	t.once.Do(func() {
+		go t.forwardWorker()
+	})
+}
+
+func (t *DataTrack) ID() string {
+	return t.id
+}
+
+func (t *DataTrack) Kind() livekit.TrackInfo_Type {
+	return livekit.TrackInfo_DATA
+}
+
+func (t *DataTrack) StreamID() string {
+	return t.dataChannel.Label()
+}
+
+func (t *DataTrack) AddSubscriber(participant *Participant) error {
+	label := PackDataTrackLabel(t.participantId, t.ID(), t.dataChannel.Label())
+	downChannel, err := participant.peerConn.CreateDataChannel(label, t.dataChannelOptions())
+	if err != nil {
+		return err
+	}
+
+	sub := &DownDataChannel{
+		participantId: participant.ID(),
+		dataChannel:   downChannel,
+	}
+
+	t.lock.Lock()
+	t.subscribers[participant.ID()] = sub
+	t.lock.Unlock()
+
+	downChannel.OnClose(func() {
+		t.RemoveSubscriber(sub.participantId)
+	})
+	return nil
+}
+
+func (t *DataTrack) RemoveSubscriber(participantId string) {
+	t.lock.Lock()
+	sub := t.subscribers[participantId]
+	delete(t.subscribers, participantId)
+	t.lock.Unlock()
+
+	if sub != nil {
+		go sub.dataChannel.Close()
+	}
+}
+
+func (t *DataTrack) RemoveAllSubscribers() {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	for _, sub := range t.subscribers {
+		go sub.dataChannel.Close()
+	}
+	t.subscribers = make(map[string]*DownDataChannel)
+}
+
+func (t *DataTrack) forwardWorker() {
+	defer func() {
+		t.RemoveAllSubscribers()
+	}()
+
+	for {
+		msg := <-t.msgChan
+		if msg.Value == nil {
+			// track closed
+			return
+		}
+
+		t.lock.RLock()
+		for _, sub := range t.subscribers {
+			err := sub.SendMessage(msg)
+			if err != nil {
+				logger.GetLogger().Errorw("could not send data message",
+					"err", err,
+					"source", t.participantId,
+					"dest", sub.participantId)
+			}
+		}
+		t.lock.RUnlock()
+	}
+}
+
+func (t *DataTrack) dataChannelOptions() *webrtc.DataChannelInit {
+	ordered := t.dataChannel.Ordered()
+	protocol := t.dataChannel.Protocol()
+	negotiated := false
+
+	return &webrtc.DataChannelInit{
+		Ordered:           &ordered,
+		MaxPacketLifeTime: t.dataChannel.MaxPacketLifeTime(),
+		MaxRetransmits:    t.dataChannel.MaxRetransmits(),
+		Protocol:          &protocol,
+		Negotiated:        &negotiated,
+	}
+}
+
+type DownDataChannel struct {
+	participantId string
+	dataChannel   *webrtc.DataChannel
+}
+
+func (d *DownDataChannel) SendMessage(msg livekit.DataMessage) error {
+	var err error
+	switch val := msg.Value.(type) {
+	case *livekit.DataMessage_Binary:
+		err = d.dataChannel.Send(val.Binary)
+	case *livekit.DataMessage_Text:
+		err = d.dataChannel.SendText(val.Text)
+	}
+	return err
+}
+
+func messageFromDataChannelMessage(msg webrtc.DataChannelMessage) livekit.DataMessage {
+	dm := livekit.DataMessage{}
+	if msg.IsString {
+		dm.Value = &livekit.DataMessage_Text{
+			Text: string(msg.Data),
+		}
+	} else {
+		dm.Value = &livekit.DataMessage_Binary{
+			Binary: msg.Data,
+		}
+
+	}
+	return dm
+}

pkg/rtc/mediatrack.go

@@ -17,13 +17,13 @@ import (
 )
 
 var (
-	creationDelay   = 500 * time.Millisecond
 	maxPLIFrequency = 1 * time.Second
 	feedbackTypes   = []webrtc.RTCPFeedback{{"goog-remb", ""}, {"nack", ""}, {"nack", "pli"}}
 )
 
-// Track represents a remoteTrack that needs to be forwarded
-type Track struct {
+// MediaTrack represents a WebRTC track that needs to be forwarded
+// Implements the PublishedTrack interface
+type MediaTrack struct {
 	ctx           context.Context
 	id            string
 	participantId string
@@ -32,21 +32,23 @@ type Track struct {
 	// channel to send RTCP packets to the source
 	rtcpCh chan []rtcp.Packet
 	lock   sync.RWMutex
+	once   sync.Once
 	// map of target participantId -> forwarder
 	forwarders map[string]Forwarder
 	receiver   *Receiver
-	lastNack   int64
-	lastPLI    time.Time
+	//lastNack   int64
+	lastPLI time.Time
 }
 
-func NewTrack(pId string, rtcpCh chan []rtcp.Packet, track *webrtc.TrackRemote, receiver *Receiver) *Track {
-	t := &Track{
+func NewMediaTrack(pId string, rtcpCh chan []rtcp.Packet, track *webrtc.TrackRemote, receiver *Receiver) *MediaTrack {
+	t := &MediaTrack{
 		ctx:           context.Background(),
 		id:            utils.NewGuid(utils.TrackPrefix),
 		participantId: pId,
 		remoteTrack:   track,
 		rtcpCh:        rtcpCh,
 		lock:          sync.RWMutex{},
+		once:          sync.Once{},
 		forwarders:    make(map[string]Forwarder),
 		receiver:      receiver,
 	}
@@ -54,23 +56,35 @@ func NewTrack(pId string, rtcpCh chan []rtcp.Packet, track *webrtc.TrackRemote,
 	return t
 }
 
-func (t *Track) Start() {
-	t.receiver.Start()
-	// start worker
-	go t.forwardRTPWorker()
+func (t *MediaTrack) Start() {
+	t.once.Do(func() {
+		t.receiver.Start()
+		// start worker
+		go t.forwardRTPWorker()
+	})
 }
 
-func (t *Track) Kind() webrtc.RTPCodecType {
-	return t.remoteTrack.Kind()
+func (t *MediaTrack) ID() string {
+	return t.id
 }
 
-func (t *Track) StreamID() string {
+func (t *MediaTrack) Kind() livekit.TrackInfo_Type {
+	switch t.remoteTrack.Kind() {
+	case webrtc.RTPCodecTypeVideo:
+		return livekit.TrackInfo_VIDEO
+	case webrtc.RTPCodecTypeAudio:
+		return livekit.TrackInfo_AUDIO
+	}
+	panic("unsupported track kind")
+}
+
+func (t *MediaTrack) StreamID() string {
 	return t.remoteTrack.StreamID()
 }
 
 // subscribes participant to current remoteTrack
 // creates and add necessary forwarders and starts them
-func (t *Track) AddSubscriber(participant *Participant) error {
+func (t *MediaTrack) AddSubscriber(participant *Participant) error {
 	codec := t.remoteTrack.Codec()
 	// pack ID to identify all tracks
 	packedId := PackTrackId(t.participantId, t.id)
@@ -136,7 +150,7 @@ func (t *Track) AddSubscriber(participant *Participant) error {
 
 // removes peer from subscription
 // stop all forwarders to the peer
-func (t *Track) RemoveSubscriber(participantId string) {
+func (t *MediaTrack) RemoveSubscriber(participantId string) {
 	t.lock.RLock()
 	defer t.lock.RUnlock()
 
@@ -145,33 +159,18 @@ func (t *Track) RemoveSubscriber(participantId string) {
 	}
 }
 
-func (t *Track) RemoveAllSubscribers() {
+func (t *MediaTrack) RemoveAllSubscribers() {
 	logger.GetLogger().Debugw("removing all subscribers", "track", t.id)
 	t.lock.RLock()
 	defer t.lock.RUnlock()
 	for _, f := range t.forwarders {
 		go f.Close()
 	}
-}
-
-func (t *Track) ToProto() *livekit.TrackInfo {
-	var kind livekit.TrackInfo_Type
-	switch t.Kind() {
-	case webrtc.RTPCodecTypeAudio:
-		kind = livekit.TrackInfo_AUDIO
-	case webrtc.RTPCodecTypeVideo:
-		kind = livekit.TrackInfo_VIDEO
-	}
-
-	return &livekit.TrackInfo{
-		Sid:  t.id,
-		Type: kind,
-		Name: t.remoteTrack.StreamID(),
-	}
+	t.forwarders = make(map[string]Forwarder)
 }
 
 // forwardRTPWorker reads from the receiver and writes to each sender
-func (t *Track) forwardRTPWorker() {
+func (t *MediaTrack) forwardRTPWorker() {
 	defer func() {
 		t.RemoveAllSubscribers()
 		// TODO: send unpublished events?

pkg/rtc/participant.go

@@ -37,12 +37,12 @@ type Participant struct {
 
 	lock           sync.RWMutex
 	receiverConfig ReceiverConfig
-	tracks         map[string]*Track // tracks that the peer is publishing
+	tracks         map[string]PublishedTrack // tracks that the peer is publishing
 	once           sync.Once
 
 	// callbacks & handlers
-	// OnParticipantTrack - remote peer added a remoteTrack
-	OnParticipantTrack func(*Participant, *Track)
+	// OnTrackPublished - remote peer added a remoteTrack
+	OnTrackPublished func(*Participant, PublishedTrack)
 	// OnOffer - offer is ready for remote peer
 	OnOffer func(webrtc.SessionDescription)
 	// OnIceCandidate - ice candidate discovered for local peer
@@ -78,13 +78,13 @@ func NewParticipant(conf *WebRTCConfig, sc SignalConnection, name string) (*Part
 		state:          livekit.ParticipantInfo_JOINING,
 		lock:           sync.RWMutex{},
 		receiverConfig: conf.receiver,
-		tracks:         make(map[string]*Track, 0),
+		tracks:         make(map[string]PublishedTrack, 0),
 		mediaEngine:    me,
 	}
 
 	log := logger.GetLogger()
 
-	pc.OnTrack(participant.onTrack)
+	pc.OnTrack(participant.onMediaTrack)
 
 	pc.OnICECandidate(func(c *webrtc.ICECandidate) {
 		if c == nil {
@@ -116,6 +116,7 @@ func NewParticipant(conf *WebRTCConfig, sc SignalConnection, name string) (*Part
 	})
 
 	// TODO: handle data channel
+	pc.OnDataChannel(participant.onDataChannel)
 
 	return participant, nil
 }
@@ -140,7 +141,7 @@ func (p *Participant) ToProto() *livekit.ParticipantInfo {
 	}
 
 	for _, t := range p.tracks {
-		info.Tracks = append(info.Tracks, t.ToProto())
+		info.Tracks = append(info.Tracks, TrackToProto(t))
 	}
 	return info
 }
@@ -299,7 +300,7 @@ func (p *Participant) AddSubscriber(op *Participant) error {
 		logger.GetLogger().Debugw("subscribing to remoteTrack",
 			"srcParticipant", p.ID(),
 			"dstParticipant", op.ID(),
-			"remoteTrack", track.id)
+			"remoteTrack", track.ID())
 		if err := track.AddSubscriber(op); err != nil {
 			return err
 		}
@@ -354,30 +355,44 @@ func (p *Participant) updateState(state livekit.ParticipantInfo_State) {
 }
 
 // when a new remoteTrack is created, creates a Track and adds it to room
-func (p *Participant) onTrack(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
-	track.StreamID()
+func (p *Participant) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
 	logger.GetLogger().Debugw("remoteTrack added", "participantId", p.ID(), "remoteTrack", track.ID())
 
 	// create Receiver
-	// p.mediaEngine.TCCExt
 	receiver := NewReceiver(p.ctx, p.id, rtpReceiver, p.bi)
-	pt := NewTrack(p.id, p.rtcpCh, track, receiver)
+	mt := NewMediaTrack(p.id, p.rtcpCh, track, receiver)
 
+	p.handleTrackPublished(mt)
+}
+
+func (p *Participant) onDataChannel(dc *webrtc.DataChannel) {
+	logger.GetLogger().Debugw("dataChannel added", "participantId", p.ID(), "id", dc.ID())
+
+	dt := NewDataTrack(p.id, dc)
 	p.lock.Lock()
-	p.tracks[pt.id] = pt
+	p.tracks[dt.id] = dt
 	p.lock.Unlock()
 
-	pt.Start()
+	dt.Start()
+
+	p.handleTrackPublished(dt)
+}
+
+func (p *Participant) handleTrackPublished(track PublishedTrack) {
+	p.lock.Lock()
+	p.tracks[track.ID()] = track
+	p.lock.Unlock()
+
+	track.Start()
 
 	// confirm publication
 	p.sigConn.WriteResponse(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_TrackPublished{
-			TrackPublished: pt.ToProto(),
+			TrackPublished: TrackToProto(track),
 		},
 	})
-	if p.OnParticipantTrack != nil {
-		// caller should hook up what happens when the peer remoteTrack is available
-		go p.OnParticipantTrack(p, pt)
+	if p.OnTrackPublished != nil {
+		go p.OnTrackPublished(p, track)
 	}
 }
 
@@ -478,33 +493,4 @@ func (p *Participant) rtcpSendWorker() {
 				"err", err)
 		}
 	}
-	//t := time.NewTicker(time.Second)
-	//for {
-	//	select {
-	//	case <-t.C:
-	//		pkts := make([]rtcp.Packet, 0)
-	//		p.lock.RLock()
-	//		for _, r := range p.tracks {
-	//			rr, ps := r.receiver.BuildRTCP()
-	//			if rr.SSRC != 0 {
-	//				ps = append(ps, &rtcp.ReceiverReport{
-	//					Reports: []rtcp.ReceptionReport{rr},
-	//				})
-	//			}
-	//			pkts = append(pkts, ps...)
-	//		}
-	//		p.lock.RUnlock()
-	//		if len(pkts) > 0 {
-	//			if err := p.peerConn.WriteRTCP(pkts); err != nil {
-	//				logger.GetLogger().Errorw("error writing RTCP to peer",
-	//					"peer", p.id,
-	//					"err", err,
-	//				)
-	//			}
-	//		}
-	//	case <-p.ctx.Done():
-	//		t.Stop()
-	//		return
-	//	}
-	//}
 }

pkg/rtc/publishedtrack.go

@@ -0,0 +1,25 @@
+package rtc
+
+import (
+	"github.com/livekit/livekit-server/proto/livekit"
+)
+
+// PublishedTrack is the main interface representing a track published to the room
+// it's responsible for managing subscribers and forwarding data from the input track to all subscribers
+type PublishedTrack interface {
+	Start()
+	ID() string
+	Kind() livekit.TrackInfo_Type
+	StreamID() string
+	AddSubscriber(participant *Participant) error
+	RemoveSubscriber(participantId string)
+	RemoveAllSubscribers()
+}
+
+func TrackToProto(t PublishedTrack) *livekit.TrackInfo {
+	return &livekit.TrackInfo{
+		Sid:  t.ID(),
+		Type: t.Kind(),
+		Name: t.StreamID(),
+	}
+}

pkg/rtc/room.go

@@ -63,7 +63,7 @@ func (r *Room) Join(participant *Participant) error {
 	log := logger.GetLogger()
 
 	// it's important to set this before connection, we don't want to miss out on any tracks
-	participant.OnParticipantTrack = r.onTrackAdded
+	participant.OnTrackPublished = r.onTrackAdded
 	participant.OnStateChange = func(p *Participant, oldState livekit.ParticipantInfo_State) {
 		log.Debugw("participant state changed", "state", p.state, "participant", p.id)
 		r.broadcastParticipantState(p)
@@ -122,15 +122,15 @@ func (r *Room) RemoveParticipant(id string) {
 	delete(r.participants, id)
 }
 
-// a peer in the room added a new remoteTrack, subscribe other participants to it
-func (r *Room) onTrackAdded(participant *Participant, track *Track) {
+// a Participant in the room added a new remoteTrack, subscribe other participants to it
+func (r *Room) onTrackAdded(participant *Participant, track PublishedTrack) {
 	// publish participant update, since track state is changed
 	r.broadcastParticipantState(participant)
 
 	r.lock.RLock()
 	defer r.lock.RUnlock()
 
-	// subscribe all existing participants to this remoteTrack
+	// subscribe all existing participants to this PublishedTrack
 	// this is the default behavior. in the future this could be more selective
 	for _, existingParticipant := range r.participants {
 		if existingParticipant == participant {
@@ -140,7 +140,7 @@ func (r *Room) onTrackAdded(participant *Participant, track *Track) {
 		if err := track.AddSubscriber(existingParticipant); err != nil {
 			logger.GetLogger().Errorw("could not subscribe to remoteTrack",
 				"srcParticipant", participant.ID(),
-				"remoteTrack", track.id,
+				"remoteTrack", track.ID(),
 				"dstParticipant", existingParticipant.ID())
 		}
 	}

pkg/rtc/utils.go

@@ -25,6 +25,21 @@ func PackTrackId(participantId, trackId string) string {
 	return participantId + trackIdSeparator + trackId
 }
 
+func PackDataTrackLabel(participantId, trackId string, label string) string {
+	return participantId + trackIdSeparator + trackId + trackIdSeparator + label
+}
+
+func UnpackDataTrackLabel(packed string) (peerId string, trackId string, label string) {
+	parts := strings.Split(packed, trackIdSeparator)
+	if len(parts) != 3 {
+		return "", packed, ""
+	}
+	peerId = parts[0]
+	trackId = parts[1]
+	label = parts[2]
+	return
+}
+
 func ToProtoParticipants(participants []*Participant) []*livekit.ParticipantInfo {
 	infos := make([]*livekit.ParticipantInfo, 0, len(participants))
 	for _, op := range participants {

proto/livekit/model.pb.go

@@ -542,17 +542,19 @@ func (x *TrackInfo) GetName() string {
 	return ""
 }
 
-type DataChannel struct {
+type DataMessage struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	SessionId string `protobuf:"bytes,1,opt,name=session_id,json=sessionId,proto3" json:"session_id,omitempty"`
-	Payload   []byte `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"`
+	// Types that are assignable to Value:
+	//	*DataMessage_Text
+	//	*DataMessage_Binary
+	Value isDataMessage_Value `protobuf_oneof:"value"`
 }
 
-func (x *DataChannel) Reset() {
-	*x = DataChannel{}
+func (x *DataMessage) Reset() {
+	*x = DataMessage{}
 	if protoimpl.UnsafeEnabled {
 		mi := &file_model_proto_msgTypes[6]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -560,13 +562,13 @@ func (x *DataChannel) Reset() {
 	}
 }
 
-func (x *DataChannel) String() string {
+func (x *DataMessage) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*DataChannel) ProtoMessage() {}
+func (*DataMessage) ProtoMessage() {}
 
-func (x *DataChannel) ProtoReflect() protoreflect.Message {
+func (x *DataMessage) ProtoReflect() protoreflect.Message {
 	mi := &file_model_proto_msgTypes[6]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -578,25 +580,48 @@ func (x *DataChannel) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use DataChannel.ProtoReflect.Descriptor instead.
-func (*DataChannel) Descriptor() ([]byte, []int) {
+// Deprecated: Use DataMessage.ProtoReflect.Descriptor instead.
+func (*DataMessage) Descriptor() ([]byte, []int) {
 	return file_model_proto_rawDescGZIP(), []int{6}
 }
 
-func (x *DataChannel) GetSessionId() string {
-	if x != nil {
-		return x.SessionId
+func (m *DataMessage) GetValue() isDataMessage_Value {
+	if m != nil {
+		return m.Value
+	}
+	return nil
+}
+
+func (x *DataMessage) GetText() string {
+	if x, ok := x.GetValue().(*DataMessage_Text); ok {
+		return x.Text
 	}
 	return ""
 }
 
-func (x *DataChannel) GetPayload() []byte {
-	if x != nil {
-		return x.Payload
+func (x *DataMessage) GetBinary() []byte {
+	if x, ok := x.GetValue().(*DataMessage_Binary); ok {
+		return x.Binary
 	}
 	return nil
 }
 
+type isDataMessage_Value interface {
+	isDataMessage_Value()
+}
+
+type DataMessage_Text struct {
+	Text string `protobuf:"bytes,1,opt,name=text,proto3,oneof"`
+}
+
+type DataMessage_Binary struct {
+	Binary []byte `protobuf:"bytes,2,opt,name=binary,proto3,oneof"`
+}
+
+func (*DataMessage_Text) isDataMessage_Value() {}
+
+func (*DataMessage_Binary) isDataMessage_Value() {}
+
 var File_model_proto protoreflect.FileDescriptor
 
 var file_model_proto_rawDesc = []byte{
@@ -653,11 +678,11 @@ var file_model_proto_rawDesc = []byte{
 	0x61, 0x6d, 0x65, 0x22, 0x26, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x12, 0x09, 0x0a, 0x05, 0x41,
 	0x55, 0x44, 0x49, 0x4f, 0x10, 0x00, 0x12, 0x09, 0x0a, 0x05, 0x56, 0x49, 0x44, 0x45, 0x4f, 0x10,
 	0x01, 0x12, 0x08, 0x0a, 0x04, 0x44, 0x41, 0x54, 0x41, 0x10, 0x02, 0x22, 0x46, 0x0a, 0x0b, 0x44,
-	0x61, 0x74, 0x61, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x12, 0x1d, 0x0a, 0x0a, 0x73, 0x65,
-	0x73, 0x73, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09,
-	0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x70, 0x61, 0x79,
-	0x6c, 0x6f, 0x61, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x70, 0x61, 0x79, 0x6c,
-	0x6f, 0x61, 0x64, 0x42, 0x31, 0x5a, 0x2f, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f,
+	0x61, 0x74, 0x61, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x14, 0x0a, 0x04, 0x74, 0x65,
+	0x78, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x48, 0x00, 0x52, 0x04, 0x74, 0x65, 0x78, 0x74,
+	0x12, 0x18, 0x0a, 0x06, 0x62, 0x69, 0x6e, 0x61, 0x72, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c,
+	0x48, 0x00, 0x52, 0x06, 0x62, 0x69, 0x6e, 0x61, 0x72, 0x79, 0x42, 0x07, 0x0a, 0x05, 0x76, 0x61,
+	0x6c, 0x75, 0x65, 0x42, 0x31, 0x5a, 0x2f, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f,
 	0x6d, 0x2f, 0x6c, 0x69, 0x76, 0x65, 0x6b, 0x69, 0x74, 0x2f, 0x6c, 0x69, 0x76, 0x65, 0x6b, 0x69,
 	0x74, 0x2d, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x6c,
 	0x69, 0x76, 0x65, 0x6b, 0x69, 0x74, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
@@ -686,7 +711,7 @@ var file_model_proto_goTypes = []interface{}{
 	(*RoomInfo)(nil),           // 5: livekit.RoomInfo
 	(*ParticipantInfo)(nil),    // 6: livekit.ParticipantInfo
 	(*TrackInfo)(nil),          // 7: livekit.TrackInfo
-	(*DataChannel)(nil),        // 8: livekit.DataChannel
+	(*DataMessage)(nil),        // 8: livekit.DataMessage
 }
 var file_model_proto_depIdxs = []int32{
 	3, // 0: livekit.Node.stats:type_name -> livekit.NodeStats
@@ -779,7 +804,7 @@ func file_model_proto_init() {
 			}
 		}
 		file_model_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*DataChannel); i {
+			switch v := v.(*DataMessage); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -791,6 +816,10 @@ func file_model_proto_init() {
 			}
 		}
 	}
+	file_model_proto_msgTypes[6].OneofWrappers = []interface{}{
+		(*DataMessage_Text)(nil),
+		(*DataMessage_Binary)(nil),
+	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{

proto/model.proto

@@ -61,7 +61,9 @@ message TrackInfo {
 }
 
 
-message DataChannel {
-  string session_id = 1;
-  bytes payload = 2;
+message DataMessage {
+  oneof value {
+    string text = 1;
+    bytes binary = 2;
+  }
 }