Minor fixes

Signed-off-by: shishir gowda <shishir@livekit.io>

pkg/rtc/participant_signal.go

@@ -31,7 +31,7 @@ func (p *ParticipantImpl) SetResponseSink(sink routing.MessageSink) {
 
 func (p *ParticipantImpl) HandleDataStreamRequest(request *livekit.GetDataStreamRequest) error {
 	handler := p.onDataStreamRequest
-	if handler == nil {
+	if handler != nil {
 		rsp, err := handler(request)
 		if err != nil {
 			return err

pkg/rtc/room.go

@@ -68,6 +68,7 @@ type Room struct {
 	onMetadataUpdate     func(metadata string)
 	onClose              func()
 	onGetDataStream      func(bucket string) ([]*livekit.DataPacket_Stream, error)
+	onDataStreamSave     func(bucket string, key string, value *livekit.DataPacket_Stream) error
 }
 
 type ParticipantOptions struct {
@@ -112,12 +113,22 @@ func NewRoom(
 	return r
 }
 
+func (r *Room) getBucketName(name string) string {
+	return string(r.ID()) + "_" + name
+}
+
 func (r *Room) OnGetDataStream(callback func(bucket string) ([]*livekit.DataPacket_Stream, error)) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 	r.onGetDataStream = callback
 }
 
+func (r *Room) OnDataStreamSave(callback func(bucket string, key string, value *livekit.DataPacket_Stream) error) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	r.onDataStreamSave = callback
+}
+
 func (r *Room) ToProto() *livekit.Room {
 	r.lock.RLock()
 	defer r.lock.RUnlock()
@@ -226,12 +237,14 @@ func (r *Room) OnDataStreamRequest(request *livekit.GetDataStreamRequest) (*live
 	if handler == nil {
 		return nil, errNotFound
 	}
-	data, err := handler(request.GetName())
+	data, err := handler(r.getBucketName(request.GetName()))
 	if err != nil {
 		return nil, err
 	}
 	var packets []*livekit.StreamPacket
 	for _, d := range data {
+		//logger.Infow("PACKETS sending", "name", d.Stream.GetName(),
+		//	"key", d.Stream.GetKey())
 		packets = append(packets, &livekit.StreamPacket{
 			Name:  d.Stream.GetName(),
 			Key:   d.Stream.GetKey(),
@@ -825,6 +838,16 @@ func (r *Room) onDataPacket(source types.LocalParticipant, dp *livekit.DataPacke
 			r.Logger.Infow("send data packet error", "error", err, "participant", op.Identity())
 		}
 	}
+
+	switch v := dp.GetValue().(type) {
+	case *livekit.DataPacket_Stream:
+		if v.Stream.Persistent {
+			err := r.onDataStreamSave(r.getBucketName(v.Stream.GetName()), v.Stream.GetKey(), v)
+			if err != nil {
+				logger.Errorw("failed to save dataStream", err)
+			}
+		}
+	}
 }
 
 func (r *Room) subscribeToExistingTracks(p types.LocalParticipant) int {

pkg/service/datastreamstore.go

@@ -58,7 +58,8 @@ func (ds *LocalDataStreamStore) Get(bucket string, key string) (*livekit.DataPac
 	if err != nil {
 		return nil, err
 	}
-	return v.(*livekit.DataPacket_Stream), nil
+	s := v.(livekit.DataPacket_Stream)
+	return &s, nil
 }
 
 func (ds *LocalDataStreamStore) GetAll(bucket string) ([]*livekit.DataPacket_Stream, error) {
@@ -66,12 +67,15 @@ func (ds *LocalDataStreamStore) GetAll(bucket string) ([]*livekit.DataPacket_Str
 	ds.RLock()
 	defer ds.RUnlock()
 	for _, key := range ds.Cache.GetKeys() {
+		//logger.Infow("==========GET FOUND", "key", key, "bucket", bucket)
 		if bucket == getBucketNameFromKey(key) {
+			//	logger.Infow("==========GOT MATCHED", "key", key, "bucket", bucket)
 			d, err := ds.Cache.Get(key)
 			if err != nil {
 				continue
 			}
-			rsp = append(rsp, d.(*livekit.DataPacket_Stream))
+			s := d.(livekit.DataPacket_Stream)
+			rsp = append(rsp, &s)
 		}
 	}
 	return rsp, nil
@@ -80,5 +84,7 @@ func (ds *LocalDataStreamStore) GetAll(bucket string) ([]*livekit.DataPacket_Str
 func (ds *LocalDataStreamStore) Put(bucket string, key string, value *livekit.DataPacket_Stream) error {
 	ds.Lock()
 	defer ds.Unlock()
+	//	logger.Infow("==========PUT", "key", value.Stream.GetKey(), "name", value.Stream.GetName(),
+	//		"bucket", bucket)
 	return ds.Cache.SetWithTTL(generateKeyName(bucket, key), *value, ds.ttl)
 }

pkg/service/roommanager.go

@@ -404,6 +404,7 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.Room
 	newRoom := rtc.NewRoom(ri, internal, *r.rtcConfig, &r.config.Audio, r.serverInfo, r.telemetry, r.egressLauncher)
 
 	newRoom.OnGetDataStream(r.dataStreamManager.GetAll)
+	newRoom.OnDataStreamSave(r.dataStreamManager.Put)
 	newRoom.OnClose(func() {
 		roomInfo := newRoom.ToProto()
 		r.telemetry.RoomEnded(ctx, roomInfo)