Merge main into hiroshi/fix-stream-progress

Resolve conflicts by adopting enum-based TextStreamOperationType approach:
- Use TextStreamOperationType.fromPBType() for incoming streams
- Keep enum type for TextStreamInfo.operationType
- Update protocol conversion to use .toPBType()
- Maintain consistency with main branch enum approach

Author: hiroshihorie

lib/src/core/room.dart

@@ -1340,7 +1340,7 @@ extension DataStreamRoomMethods on Room {
             ? streamHeader.textHeader.generated
             : false,
         operationType: streamHeader.textHeader.hasOperationType()
-            ? TextStreamOperationType.fromString(streamHeader.textHeader.operationType.name)
+            ? TextStreamOperationType.fromPBType(streamHeader.textHeader.operationType)
             : null,
       );
 

lib/src/participant/local.dart

@@ -1152,22 +1152,6 @@ extension RPCMethods on LocalParticipant {
   }
 }
 
-/// Helper function to convert string operation type to enum
-lk_models.DataStream_OperationType _stringToOperationType(String? type) {
-  switch (type?.toLowerCase()) {
-    case 'create':
-      return lk_models.DataStream_OperationType.CREATE;
-    case 'update':
-      return lk_models.DataStream_OperationType.UPDATE;
-    case 'delete':
-      return lk_models.DataStream_OperationType.DELETE;
-    case 'reaction':
-      return lk_models.DataStream_OperationType.REACTION;
-    default:
-      return lk_models.DataStream_OperationType.CREATE;
-  }
-}
-
 extension DataStreamParticipantMethods on LocalParticipant {
   Future<TextStreamInfo> sendText(String text,
       {SendTextOptions? options}) async {
@@ -1231,40 +1215,43 @@ extension DataStreamParticipantMethods on LocalParticipant {
 
   Future<TextStreamWriter> streamText(StreamTextOptions? options) async {
     final streamId = options?.streamId ?? Uuid().v4();
+    final timestamp = DateTime.timestamp().millisecondsSinceEpoch;
 
     final info = TextStreamInfo(
       id: streamId,
       mimeType: 'text/plain',
-      timestamp: DateTime.timestamp().millisecondsSinceEpoch,
+      timestamp: timestamp,
       topic: options?.topic ?? '',
       size: options?.totalSize ?? 0,
       replyToStreamId: options?.replyToStreamId,
       attachedStreamIds: options?.attachedStreamIds ?? [],
       version: options?.version,
       generated: options?.generated ?? false,
-      operationType: TextStreamOperationType.fromString(options?.type),
+      operationType: options?.type,
     );
 
     final header = lk_models.DataStream_Header(
       streamId: streamId,
       mimeType: info.mimeType,
       topic: info.topic,
-      timestamp: Int64(info.timestamp),
+      timestamp: Int64(timestamp),
       totalLength: Int64(options?.totalSize ?? 0),
       attributes: options?.attributes.entries,
       textHeader: lk_models.DataStream_TextHeader(
         version: options?.version,
         attachedStreamIds: options?.attachedStreamIds,
         replyToStreamId: options?.replyToStreamId,
         generated: options?.generated ?? false,
-        operationType: _stringToOperationType(options?.type),
+        operationType: options?.type?.toPBType(),
       ),
     );
+
     final destinationIdentities = options?.destinationIdentities;
     final packet = lk_models.DataPacket(
       destinationIdentities: destinationIdentities,
       streamHeader: header,
     );
+
     await room.engine.sendDataPacket(packet, reliability: true);
 
     final writableStream = WritableStream<String>(
@@ -1332,12 +1319,13 @@ extension DataStreamParticipantMethods on LocalParticipant {
 
   Future<ByteStreamWriter> streamBytes(StreamBytesOptions? options) async {
     final streamId = options?.streamId ?? Uuid().v4();
+    final timestamp = DateTime.timestamp().millisecondsSinceEpoch;
 
     final info = ByteStreamInfo(
       name: options?.name ?? 'unknown',
       id: streamId,
       mimeType: options?.mimeType ?? 'application/octet-stream',
-      timestamp: DateTime.timestamp().millisecondsSinceEpoch,
+      timestamp: timestamp,
       topic: options?.topic ?? '',
       size: options?.totalSize ?? 0,
       attributes: options?.attributes ?? {},
@@ -1349,7 +1337,7 @@ extension DataStreamParticipantMethods on LocalParticipant {
       streamId: streamId,
       topic: options?.topic,
       encryptionType: options?.encryptionType,
-      timestamp: Int64(info.timestamp),
+      timestamp: Int64(timestamp),
       byteHeader: lk_models.DataStream_ByteHeader(
         name: info.name,
       ),

lib/src/types/data_stream.dart

@@ -2,6 +2,7 @@ import 'dart:async';
 import 'dart:io' show File;
 
 import '../data_stream/stream_reader.dart';
+import '../proto/livekit_models.pb.dart' as lk_models;
 import '../proto/livekit_models.pb.dart' show Encryption_Type, DataStream_Chunk;
 
 const kStreamChunkSize = 15_000;
@@ -53,7 +54,7 @@ class StreamTextOptions {
   int? totalSize;
 
   /// 'create' | 'update' | 'delete' | 'reaction'
-  String? type;
+  TextStreamOperationType? type;
 
   /// true if the text has been generated by an agent from a participant's audio transcription
   bool generated;
@@ -172,21 +173,34 @@ enum TextStreamOperationType {
   delete,
   reaction;
 
-  static TextStreamOperationType? fromString(String? stringValue) {
-    if (stringValue == null) return null;
-    switch (stringValue.toUpperCase()) {
-      case 'CREATE':
+  static TextStreamOperationType? fromPBType(lk_models.DataStream_OperationType? type) {
+    if (type == null) return TextStreamOperationType.create;
+    switch (type) {
+      case lk_models.DataStream_OperationType.CREATE:
         return TextStreamOperationType.create;
-      case 'UPDATE':
+      case lk_models.DataStream_OperationType.UPDATE:
         return TextStreamOperationType.update;
-      case 'DELETE':
+      case lk_models.DataStream_OperationType.DELETE:
         return TextStreamOperationType.delete;
-      case 'REACTION':
+      case lk_models.DataStream_OperationType.REACTION:
         return TextStreamOperationType.reaction;
       default:
         return null;
     }
   }
+
+  lk_models.DataStream_OperationType toPBType() {
+    switch (this) {
+      case TextStreamOperationType.create:
+        return lk_models.DataStream_OperationType.CREATE;
+      case TextStreamOperationType.update:
+        return lk_models.DataStream_OperationType.UPDATE;
+      case TextStreamOperationType.delete:
+        return lk_models.DataStream_OperationType.DELETE;
+      case TextStreamOperationType.reaction:
+        return lk_models.DataStream_OperationType.REACTION;
+    }
+  }
 }
 
 class TextStreamInfo extends BaseStreamInfo {

test/core/data_stream_test.dart

@@ -157,7 +157,12 @@ void main() {
     });
 
     test('Text Stream With Operation Types', () async {
-      final operationTypes = ['create', 'update', 'delete', 'reaction'];
+      final operationTypes = [
+        TextStreamOperationType.create,
+        TextStreamOperationType.update,
+        TextStreamOperationType.delete,
+        TextStreamOperationType.reaction,
+      ];
       final receivedMessages = <String>[];
 
       for (var operationType in operationTypes) {
@@ -176,7 +181,7 @@ void main() {
         final stream = await room.localParticipant?.streamText(StreamTextOptions(
           topic: 'chat-operations',
           type: operationType,
-          version: operationType == 'update' ? 2 : null,
+          version: operationType == TextStreamOperationType.update ? 2 : null,
         ));
         await stream?.write('Streamed ${operationType}');
         await stream?.close();
@@ -229,7 +234,7 @@ void main() {
       // Send a reply to an existing stream
       final stream = await room.localParticipant?.streamText(StreamTextOptions(
         topic: 'chat-replies',
-        type: 'create',
+        type: TextStreamOperationType.create,
         streamId: replyStreamId,
         replyToStreamId: originalStreamId,
         version: 1,
@@ -388,7 +393,7 @@ void main() {
           final stream = await room.localParticipant?.streamText(StreamTextOptions(
             topic: 'concurrent-streams',
             streamId: 'stream-${i}',
-            type: 'create',
+            type: TextStreamOperationType.create,
           ));
           await stream?.write('Concurrent message ${i}');
           await stream?.close();
@@ -454,7 +459,7 @@ void main() {
       // Send a message with comprehensive options
       final stream = await room.localParticipant?.streamText(StreamTextOptions(
         topic: 'header-validation',
-        type: 'create',
+        type: TextStreamOperationType.create,
         version: 1,
         generated: false,
         attributes: {