diff --git a/pkg/rtc/datatrack.go b/pkg/rtc/datatrack.go new file mode 100644 index 000000000..e2564f2a2 --- /dev/null +++ b/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 +} diff --git a/pkg/rtc/track.go b/pkg/rtc/mediatrack.go similarity index 82% rename from pkg/rtc/track.go rename to pkg/rtc/mediatrack.go index be53fdb6b..70f651e4d 100644 --- a/pkg/rtc/track.go +++ b/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? diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 2ff0d5ebf..8234645cf 100644 --- a/pkg/rtc/participant.go +++ b/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 - // } - //} } diff --git a/pkg/rtc/publishedtrack.go b/pkg/rtc/publishedtrack.go new file mode 100644 index 000000000..143fe1372 --- /dev/null +++ b/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(), + } +} diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 3aad14a3e..0914c8e96 100644 --- a/pkg/rtc/room.go +++ b/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()) } } diff --git a/pkg/rtc/utils.go b/pkg/rtc/utils.go index 18f911704..b13e323e2 100644 --- a/pkg/rtc/utils.go +++ b/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 { diff --git a/proto/livekit/model.pb.go b/proto/livekit/model.pb.go index 5b3c0e60b..62c8cc9d7 100644 --- a/proto/livekit/model.pb.go +++ b/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{ diff --git a/proto/model.proto b/proto/model.proto index 235cb5d16..b58b1bb9e 100644 --- a/proto/model.proto +++ b/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; + } }