RPC 機能に対応する

Author: zztkm

CHANGES.md

@@ -33,6 +33,19 @@
     - デフォルト値の `unspecified` の場合はシグナリングパラメータに `simulcast_request_rid` を含めない
   - role が sendrecv または recvonly の場合、かつ simulcast が true の場合にのみ有効
   - @zztkm
+- [ADD] RPC 機能を追加する
+  - `MediaChannel` に `rpc` メソッドを追加する
+  - RPC メソッドを定義するための `RPCMethodProtocol` プロトコルを追加する
+  - RPC の ID を表す `RPCID` 列挙型を追加する
+    - `int(Int)` と `string(String)` の 2 つのケースをサポート
+  - RPC エラー応答の詳細を表す `RPCErrorDetail` 構造体を追加する
+  - RPC 成功応答を表す `RPCResponse<Result>` ジェネリック構造体を追加する
+  - DataChannel 経由の RPC を扱う `RPCChannel` クラスを追加する
+  - `SoraError` に RPC 関連のエラーケースを追加する
+    - `rpcUnavailable(reason: String)`
+    - `rpcCallFailed(errorDetail: RPCErrorDetail)`
+    - `rpcInvalidResponse`
+  - @zztkm
 
 ### misc
 

Sora/DataChannel.swift

@@ -207,6 +207,8 @@ class BasicDataChannelDelegate: NSObject, RTCDataChannelDelegate {
             type: .dataChannel,
             message: "decode failed (\(error.localizedDescription)) => ")
         }
+      case "rpc":
+        peerChannel.handleRPCMessage(data)
       case "e2ee":
         Logger.error(
           type: .dataChannel, message: "NOT IMPLEMENTED: label => \(dataChannel.label)")

Sora/MediaChannel.swift

@@ -154,6 +154,21 @@ public final class MediaChannel {
   /// サーバーから通知を受信可能であり、接続中にのみ取得可能です。
   public private(set) var subscriberCount: Int?
 
+  /// RPC を扱うチャネル
+  public var rpcChannel: RPCChannel? {
+    peerChannel.rpcChannel
+  }
+
+  /// RPC で利用可能なメソッド一覧
+  public var rpcMethods: [RPCMethod] {
+    peerChannel.rpcChannel?.allowedMethods.compactMap { RPCMethod(name: $0) } ?? []
+  }
+
+  /// RPC で利用可能なサイマルキャスト rid
+  public var rpcSimulcastRids: [String] {
+    peerChannel.rpcChannel?.simulcastRpcRids ?? []
+  }
+
   // MARK: 接続チャネル
 
   /// シグナリングチャネル
@@ -227,6 +242,58 @@ public final class MediaChannel {
       timeout: configuration.connectionTimeout)
   }
 
+  // MARK: - RPC
+
+  /// RPC をジェネリクスで送信する。
+  public func rpc<M: RPCMethodProtocol>(
+    method: M.Type,
+    params: M.Params,
+    notification: Bool = false,
+    timeout: TimeInterval = 5.0
+  ) async throws -> RPCResponse<M.Result>? {
+    let response = try await withCheckedThrowingContinuation {
+      (continuation: CheckedContinuation<RPCResponse<Any>?, Error>) in
+      guard let rpcChannel else {
+        continuation.resume(
+          throwing: SoraError.rpcUnavailable(reason: "rpc channel is not available"))
+        return
+      }
+      _ = rpcChannel.call(
+        methodName: method.name,
+        params: params,
+        notification: notification,
+        timeout: timeout
+      ) { result in
+        continuation.resume(with: result)
+      }
+    }
+    guard let response else {
+      return nil
+    }
+    return try decodeRPCResponse(response, method: method)
+  }
+
+  private func decodeRPCResponse<M: RPCMethodProtocol>(
+    _ response: RPCResponse<Any>,
+    method: M.Type
+  ) throws -> RPCResponse<M.Result> {
+    let decoded: M.Result
+    do {
+      decoded = try decodeRPCResult(response.result, as: M.Result.self)
+    } catch {
+      throw SoraError.rpcDecodingError(reason: error.localizedDescription)
+    }
+    return RPCResponse<M.Result>(id: response.id, result: decoded)
+  }
+
+  private func decodeRPCResult<T: Decodable>(_ result: Any, as type: T.Type) throws -> T {
+    let data = try JSONSerialization.data(
+      withJSONObject: result,
+      options: [.fragmentsAllowed])
+    let decoder = JSONDecoder()
+    return try decoder.decode(T.self, from: data)
+  }
+
   // MARK: - 接続
 
   private var _handler: ((_ error: Error?) -> Void)?

Sora/PeerChannel.swift

@@ -125,6 +125,9 @@ class PeerChannel: NSObject, RTCPeerConnectionDelegate {
   var dataChannels: [String: DataChannel] = [:]
   var switchedToDataChannel: Bool = false
   var signalingOfferMessageDataChannels: [[String: Any]] = []
+  var rpcAllowedMethods: [String] = []
+  var rpcSimulcastRids: [String] = []
+  var rpcChannel: RPCChannel?
 
   weak var mediaChannel: MediaChannel?
 
@@ -938,6 +941,18 @@ class PeerChannel: NSObject, RTCPeerConnectionDelegate {
         signalingOfferMessageDataChannels = dataChannels
       }
 
+      if let rpcMethods = offer.rpcMethods {
+        rpcAllowedMethods = rpcMethods
+      } else {
+        rpcAllowedMethods = []
+      }
+
+      if let simulcastRpcRids = offer.simulcastRpcRids {
+        rpcSimulcastRids = simulcastRpcRids
+      } else {
+        rpcSimulcastRids = []
+      }
+
       // offer.simulcast が設定されている場合、WrapperVideoEncoderFactory.shared.simulcastEnabled を上書きする
       if let simulcast = offer.simulcast {
         WrapperVideoEncoderFactory.shared.simulcastEnabled = simulcast
@@ -1033,6 +1048,15 @@ class PeerChannel: NSObject, RTCPeerConnectionDelegate {
     internalHandlers.onReceiveSignaling?(signaling)
   }
 
+  /// DataChannel の rpc で受信したメッセージを処理する。
+  func handleRPCMessage(_ data: Data) {
+    guard let rpcChannel else {
+      Logger.warn(type: .peerChannel, message: "rpcChannel is unavailable")
+      return
+    }
+    rpcChannel.handleMessage(data)
+  }
+
   private func finishConnecting() {
     Logger.debug(type: .peerChannel, message: "did connect")
     Logger.debug(
@@ -1065,6 +1089,12 @@ class PeerChannel: NSObject, RTCPeerConnectionDelegate {
         message: "error: \(error.localizedDescription)")
     }
 
+    if let rpcChannel {
+      rpcChannel.invalidate(
+        reason: SoraError.rpcTransportClosed(reason: reason.description))
+      self.rpcChannel = nil
+    }
+
     sendDisconnectMessageIfNeeded(reason: reason, error: error)
 
     if configuration.isSender {
@@ -1363,6 +1393,13 @@ class PeerChannel: NSObject, RTCPeerConnectionDelegate {
       dataChannel: dataChannel, compress: compress, mediaChannel: mediaChannel,
       peerChannel: self)
     dataChannels[dataChannel.label] = dc
+
+    if label == "rpc" {
+      rpcChannel = RPCChannel(
+        dataChannel: dc,
+        rpcMethods: rpcAllowedMethods,
+        simulcastRpcRids: rpcSimulcastRids)
+    }
   }
 }