feat: RPC. (#682)

* feat: RPC.

* wip.

* add livekit_metrics to proto generate.

* wip.

* move rpc methods to room.

* update.

* wip.

* wip.

* fix.

* error handle.

* more test case.

* test.

* wip.

* update.

* fix flutter analyzer.

* raises exception when handler already existed.

* mark rpc acks/responses as private.

* update.

Author: cloudwebrtc

Makefile

@@ -4,7 +4,7 @@ proto:
 	@{ \
 	if [ -d "$(PROTO_DIR)" ]; \
 	then \
-		protoc --dart_out=lib/src/proto -I$(PROTO_DIR) $(PROTO_DIR)/livekit_rtc.proto $(PROTO_DIR)/livekit_models.proto; \
+		protoc --dart_out=lib/src/proto -I$(PROTO_DIR) $(PROTO_DIR)/livekit_rtc.proto $(PROTO_DIR)/livekit_models.proto $(PROTO_DIR)/livekit_metrics.proto; \
 	else \
 		echo "../protocol/protobufs is not found. github.com/livekit/protocol must be checked out"; \
 	fi \

lib/livekit_client.dart

@@ -50,5 +50,6 @@ export 'src/types/participant_permissions.dart';
 export 'src/types/video_dimensions.dart';
 export 'src/types/video_encoding.dart';
 export 'src/types/video_parameters.dart';
+export 'src/types/rpc.dart';
 export 'src/widgets/screen_select_dialog.dart';
 export 'src/widgets/video_track_renderer.dart';

lib/src/core/engine.dart

@@ -36,6 +36,7 @@ import '../proto/livekit_models.pb.dart' as lk_models;
 import '../proto/livekit_rtc.pb.dart' as lk_rtc;
 import '../publication/local.dart';
 import '../support/disposable.dart';
+import '../support/platform.dart';
 import '../support/region_url_provider.dart';
 import '../support/websocket.dart';
 import '../track/local/video.dart';
@@ -343,12 +344,13 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
             rtc.RTCPeerConnectionState.RTCPeerConnectionStateConnecting) {
           await negotiate();
         }
-
-        logger.fine('Waiting for publisher to ice-connect...');
-        await events.waitFor<EnginePublisherPeerStateUpdatedEvent>(
-          filter: (event) => event.state.isConnected(),
-          duration: connectOptions.timeouts.peerConnection,
-        );
+        if (!lkPlatformIsTest()) {
+          logger.fine('Waiting for publisher to ice-connect...');
+          await events.waitFor<EnginePublisherPeerStateUpdatedEvent>(
+            filter: (event) => event.state.isConnected(),
+            duration: connectOptions.timeouts.peerConnection,
+          );
+        }
       }
 
       // wait for data channel to open (if not already)
@@ -638,6 +640,24 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
       events.emit(EngineSipDtmfReceivedEvent(
         dtmf: dp.sipDtmf,
       ));
+    } else if (dp.whichValue() == lk_models.DataPacket_Value.rpcRequest) {
+      // RPC Request
+      events.emit(EngineRPCRequestReceivedEvent(
+        request: dp.rpcRequest,
+        identity: dp.participantIdentity,
+      ));
+    } else if (dp.whichValue() == lk_models.DataPacket_Value.rpcResponse) {
+      // RPC Response
+      events.emit(EngineRPCResponseReceivedEvent(
+        response: dp.rpcResponse,
+        identity: dp.participantIdentity,
+      ));
+    } else if (dp.whichValue() == lk_models.DataPacket_Value.rpcAck) {
+      // RPC Ack
+      events.emit(EngineRPCAckReceivedEvent(
+        ack: dp.rpcAck,
+        identity: dp.participantIdentity,
+      ));
     } else {
       logger.warning('Unknown data packet type: ${dp.whichValue()}');
     }

lib/src/core/room.dart

@@ -45,6 +45,7 @@ import '../track/local/audio.dart';
 import '../track/local/video.dart';
 import '../track/track.dart';
 import '../types/other.dart';
+import '../types/rpc.dart';
 import '../utils.dart';
 import 'engine.dart';
 
@@ -119,6 +120,12 @@ class Room extends DisposableChangeNotifier with EventsEmittable<RoomEvent> {
   // Agents
   final Map<String, DateTime> _transcriptionReceivedTimes = {};
 
+  // RPC Handlers
+  final Map<String, RpcRequestHandler> _rpcHandlers = {};
+
+  // for testing
+  Map<String, RpcRequestHandler> get rpcHandlers => _rpcHandlers;
+
   Room({
     @Deprecated('deprecated, please use connectOptions in room.connect()')
     ConnectOptions connectOptions = const ConnectOptions(),
@@ -141,6 +148,8 @@ class Room extends DisposableChangeNotifier with EventsEmittable<RoomEvent> {
       notifyListeners();
     });
 
+    _setupRpcListeners();
+
     onDispose(() async {
       // clean up routine
       await _cleanUp();
@@ -1129,3 +1138,54 @@ extension RoomHardwareManagementMethods on Room {
     events.emit(const AudioPlaybackStatusChanged(isPlaying: false));
   }
 }
+
+extension RoomRPCMethods on Room {
+  void _setupRpcListeners() {
+    // listen for incoming requests
+    _engineListener
+      ..on<EngineRPCRequestReceivedEvent>((event) async {
+        final request = event.request;
+        await _localParticipant?.handleIncomingRpcRequest(
+          event.identity,
+          request.id,
+          request.method,
+          request.payload,
+          request.responseTimeoutMs,
+          request.version,
+        );
+      })
+      ..on<EngineRPCAckReceivedEvent>((event) {
+        _localParticipant?.handleIncomingRpcAck(event.requestId);
+      })
+      ..on<EngineRPCResponseReceivedEvent>((event) {
+        String? payload;
+        RpcError? error;
+
+        if (event.payload.isNotEmpty) {
+          payload = event.response.payload;
+        } else if (event.error != null) {
+          error = RpcError.fromProto(event.error!);
+        }
+        _localParticipant?.handleIncomingRpcResponse(
+            event.requestId, payload, error);
+      });
+  }
+
+  /// Register a handler for incoming RPC requests.
+  /// @param method, the method name to listen for.
+  /// When a request with this method name is received, the handler will be called.
+  /// The handler should return a string payload to send back to the caller.
+  /// If the handler returns null, an error will be sent back to the caller.
+  void registerRpcMethod(String method, RpcRequestHandler handler) {
+    if (rpcHandlers.containsKey(method)) {
+      throw Exception('Method $method already registered');
+    }
+    rpcHandlers[method] = handler;
+  }
+
+  /// Unregister a handler for incoming RPC requests.
+  /// @param method, the method name to unregister.
+  void unregisterRpcMethod(String method) {
+    rpcHandlers.remove(method);
+  }
+}

lib/src/internal/events.dart

@@ -446,6 +446,41 @@ class EngineSipDtmfReceivedEvent with EngineEvent, InternalEvent {
   });
 }
 
+@internal
+class EngineRPCRequestReceivedEvent with EngineEvent, InternalEvent {
+  final lk_models.RpcRequest request;
+  String get requestId => request.id;
+  final String identity;
+  const EngineRPCRequestReceivedEvent({
+    required this.request,
+    required this.identity,
+  });
+}
+
+@internal
+class EngineRPCResponseReceivedEvent with EngineEvent, InternalEvent {
+  final lk_models.RpcResponse response;
+  String get requestId => response.requestId;
+  final String identity;
+  lk_models.RpcError? get error => response.error;
+  String get payload => response.payload;
+  const EngineRPCResponseReceivedEvent({
+    required this.response,
+    required this.identity,
+  });
+}
+
+@internal
+class EngineRPCAckReceivedEvent with EngineEvent, InternalEvent {
+  final lk_models.RpcAck ack;
+  String get requestId => ack.requestId;
+  final String identity;
+  const EngineRPCAckReceivedEvent({
+    required this.ack,
+    required this.identity,
+  });
+}
+
 @internal
 abstract class DataChannelStateUpdatedEvent with EngineEvent, InternalEvent {
   final bool isPrimary;