Merge pull request #31 from nova-wallet/feature/websocket-batch

Websocket batch support

SubstrateSdk/Classes/Network/JSONRPCEngine.swift

@@ -12,6 +12,13 @@ public protocol JSONRPCResponseHandling {
     func handle(error: Error)
 }
 
+public typealias JSONRPCBatchId = String
+
+public struct JSONRPCBatchRequestItem {
+    public let requestId: UInt16
+    public let data: Data
+}
+
 public struct JSONRPCRequest: Equatable {
     public let requestId: UInt16
     public let data: Data
@@ -41,6 +48,39 @@ struct JSONRPCResponseHandler<T: Decodable>: JSONRPCResponseHandling {
     }
 }
 
+struct JSONRPCBatchHandler: JSONRPCResponseHandling {
+    let itemsCount: Int
+
+    public let completionClosure: ([Result<JSON, Error>]) -> Void
+
+    public func handle(data: Data) {
+        do {
+            let decoder = JSONDecoder()
+            let responses = try decoder.decode([JSONRPCData<JSON?>].self, from: data)
+            let results: [Result<JSON, Error>] = responses.map { rpcData in
+                if let value = rpcData.result {
+                    return .success(value)
+                } else if let error = rpcData.error {
+                    return .failure(error)
+                } else {
+                    return .failure(JSONRPCEngineError.emptyResult)
+                }
+            }
+
+            completionClosure(results)
+
+        } catch {
+            let errorList: [Result<JSON, Error>] = (0..<itemsCount).map { _ in .failure(error) }
+            completionClosure(errorList)
+        }
+    }
+
+    public func handle(error: Error) {
+        let errorList: [Result<JSON, Error>] = (0..<itemsCount).map { _ in .failure(error) }
+        completionClosure(errorList)
+    }
+}
+
 public struct JSONRPCOptions {
     public let resendOnReconnect: Bool
 
@@ -54,6 +94,7 @@ public protocol JSONRPCSubscribing: AnyObject {
     var requestData: Data { get }
     var requestOptions: JSONRPCOptions { get }
     var remoteId: String? { get set }
+    var unsubscribeMethod: String { get }
 
     func handle(data: Data) throws
     func handle(error: Error, unsubscribed: Bool)
@@ -64,6 +105,7 @@ public final class JSONRPCSubscription<T: Decodable>: JSONRPCSubscribing {
     public let requestData: Data
     public let requestOptions: JSONRPCOptions
     public var remoteId: String?
+    public let unsubscribeMethod: String
 
     private lazy var jsonDecoder = JSONDecoder()
 
@@ -74,12 +116,14 @@ public final class JSONRPCSubscription<T: Decodable>: JSONRPCSubscribing {
         requestId: UInt16,
         requestData: Data,
         requestOptions: JSONRPCOptions,
+        unsubscribeMethod: String,
         updateClosure: @escaping (T) -> Void,
         failureClosure: @escaping (Error, Bool) -> Void
     ) {
         self.requestId = requestId
         self.requestData = requestData
         self.requestOptions = requestOptions
+        self.unsubscribeMethod = unsubscribeMethod
         self.updateClosure = updateClosure
         self.failureClosure = failureClosure
     }
@@ -105,15 +149,37 @@ public protocol JSONRPCEngine: AnyObject {
     func subscribe<P: Encodable, T: Decodable>(
         _ method: String,
         params: P?,
+        unsubscribeMethod: String,
         updateClosure: @escaping (T) -> Void,
         failureClosure: @escaping (Error, Bool) -> Void
     )
         throws -> UInt16
 
     func cancelForIdentifier(_ identifier: UInt16)
+
+    func addBatchCallMethod<P: Encodable>(
+        _ method: String,
+        params: P?,
+        batchId: JSONRPCBatchId
+    ) throws
+
+    func submitBatch(
+        for batchId: JSONRPCBatchId,
+        options: JSONRPCOptions,
+        completion closure: (([Result<JSON, Error>]) -> Void)?
+    ) throws -> UInt16
+
+    func clearBatch(for batchId: JSONRPCBatchId)
 }
 
 public extension JSONRPCEngine {
+    func submitBatch(
+        for batchId: JSONRPCBatchId,
+        completion closure: (([Result<JSON, Error>]) -> Void)?
+    ) throws -> UInt16 {
+        try submitBatch(for: batchId, options: JSONRPCOptions(), completion: closure)
+    }
+
     func callMethod<P: Encodable, T: Decodable>(
         _ method: String,
         params: P?,
@@ -126,4 +192,19 @@ public extension JSONRPCEngine {
             completion: closure
         )
     }
+
+    func subscribe<P: Encodable, T: Decodable>(
+        _ method: String,
+        params: P?,
+        updateClosure: @escaping (T) -> Void,
+        failureClosure: @escaping (Error, Bool) -> Void
+    ) throws -> UInt16 {
+        try subscribe(
+            method,
+            params: params,
+            unsubscribeMethod: RPCMethod.storageUnsubscribe,
+            updateClosure: updateClosure,
+            failureClosure: failureClosure
+        )
+    }
 }

SubstrateSdk/Classes/Network/WebSocketEngine+Protocol.swift

@@ -1,6 +1,60 @@
 import Foundation
 
 extension WebSocketEngine: JSONRPCEngine {
+    public func addBatchCallMethod<P: Encodable>(
+        _ method: String,
+        params: P?,
+        batchId: JSONRPCBatchId
+    ) throws {
+        mutex.lock()
+
+        defer {
+            mutex.unlock()
+        }
+
+        let batchItem = try prepareBatchRequestItem(method: method, params: params)
+        storeBatchItem(batchItem, for: batchId)
+    }
+
+    public func submitBatch(
+        for batchId: JSONRPCBatchId,
+        options: JSONRPCOptions,
+        completion closure: (([Result<JSON, Error>]) -> Void)?
+    ) throws -> UInt16 {
+        mutex.lock()
+
+        defer {
+            mutex.unlock()
+        }
+
+        guard let batchItems = partialBatches[batchId], let requestId = batchItems.first?.requestId else {
+            throw JSONRPCEngineError.emptyResult
+        }
+
+        clearPartialBatchStorage(for: batchId)
+
+        let request = try prepareBatchRequest(
+            requestId: requestId,
+            from: batchItems,
+            options: options,
+            completion: closure
+        )
+
+        updateConnectionForRequest(request)
+
+        return requestId
+    }
+
+    public func clearBatch(for batchId: JSONRPCBatchId) {
+        mutex.lock()
+
+        defer {
+            mutex.unlock()
+        }
+
+        clearPartialBatchStorage(for: batchId)
+    }
+
     public func callMethod<P: Encodable, T: Decodable>(
         _ method: String,
         params: P?,
@@ -28,6 +82,7 @@ extension WebSocketEngine: JSONRPCEngine {
     public func subscribe<P: Encodable, T: Decodable>(
         _ method: String,
         params: P?,
+        unsubscribeMethod: String,
         updateClosure: @escaping (T) -> Void,
         failureClosure: @escaping (Error, Bool) -> Void
     ) throws -> UInt16 {
@@ -50,6 +105,7 @@ extension WebSocketEngine: JSONRPCEngine {
             requestId: request.requestId,
             requestData: request.data,
             requestOptions: request.options,
+            unsubscribeMethod: unsubscribeMethod,
             updateClosure: updateClosure,
             failureClosure: failureClosure
         )