chore(datastore): add subscribe() impl

packages/amplify_datastore/ios/Classes/FlutterApiPlugin.swift

@@ -9,8 +9,19 @@ import Combine
 public class FlutterApiPlugin: APICategoryPlugin
 {
     public var key: PluginKey = "awsAPIPlugin"
-    init() {
-
+    private let apiAuthFactory: APIAuthProviderFactory
+    private let nativeApiPlugin: NativeApiPlugin
+    private let nativeSubscriptionEvents: PassthroughSubject<NativeGraphQLSubscriptionResponse, Never>
+    private var cancellables = Set<AnyCancellable>()
+
+    init(
+        apiAuthProviderFactory: APIAuthProviderFactory,
+        nativeApiPlugin: NativeApiPlugin,
+        subscriptionEventBus: PassthroughSubject<NativeGraphQLSubscriptionResponse, Never>
+    ) {
+        self.apiAuthFactory = apiAuthProviderFactory
+        self.nativeApiPlugin = nativeApiPlugin
+        self.nativeSubscriptionEvents = subscriptionEventBus
     }
 
     // TODO: Implment in Async Swift v2
@@ -23,15 +34,81 @@ public class FlutterApiPlugin: APICategoryPlugin
         preconditionFailure("method not supported")
     }
 
-    // TODO: Implment in Async Swift v2
-    public func subscribe<R>(request: GraphQLRequest<R>) -> AmplifyAsyncThrowingSequence<GraphQLSubscriptionEvent<R>> where R : Decodable {
-        preconditionFailure("method not supported")
+    public func subscribe<R: Decodable>(
+        request: GraphQLRequest<R>
+    ) -> AmplifyAsyncThrowingSequence<GraphQLSubscriptionEvent<R>> where R : Decodable {
+        var subscriptionId: String? = ""
+
+        // TODO: write a e2e test to ensure we don't go over 100 AppSync connections
+        func unsubscribe(subscriptionId: String?){
+            if let subscriptionId {
+                DispatchQueue.main.async {
+                    self.nativeApiPlugin.unsubscribe(subscriptionId: subscriptionId) {}
+                }
+            }
+        }
+
+        // TODO: shouldn't there be a timeout if there is no start_ack returned in a certain period of time
+        let (sequence, cancellable) = nativeSubscriptionEvents
+            .filter { $0.subscriptionId == subscriptionId }
+            .handleEvents(receiveCompletion: {_ in
+                unsubscribe(subscriptionId: subscriptionId)
+            }, receiveCancel: {
+                unsubscribe(subscriptionId: subscriptionId)
+            })
+            .compactMap { [weak self] event -> GraphQLSubscriptionEvent<R>? in
+                switch event.type {
+                    case "connecting":
+                        return .connection(.connecting)
+                    case "start_ack":
+                        return .connection(.connected)
+                    case "complete":
+                        return .connection(.disconnected)
+                    case "data":
+                        if let responseDecoded: GraphQLResponse<R> =
+                            try? self?.decodePayloadJson(request: request, payload: event.payloadJson)
+                        {
+                            return .data(responseDecoded)
+                        }
+                        return nil
+                    case "error":
+                        // TODO: (5d) error parsing
+                        print("received error:  \(String(describing: event.payloadJson))")
+                        return nil
+                    default:
+                        print("ERROR unsupported subscription event type! \(String(describing: event.type))")
+                        return nil
+                }
+            }
+            .toAmplifyAsyncThrowingSequence()
+        cancellables.insert(cancellable) // the subscription is bind with class instance lifecycle, it should be released when stream is finished or unsubscribed
+        sequence.send(.connection(.connecting))
+        DispatchQueue.main.async {
+            self.nativeApiPlugin.subscribe(request: request.toNativeGraphQLRequest()) { response in
+                subscriptionId = response.subscriptionId
+            }
+        }
+        return sequence
+    }
+
+    private func decodePayloadJson<R>(
+        request: GraphQLRequest<R>,
+        payload: String?
+    ) throws -> GraphQLResponse<R> {
+        guard let payload else {
+            throw DataStoreError.decodingError("Request payload could not be empty", "")
+        }
+
+        return GraphQLResponse<R>.fromAppSyncResponse(
+            string: payload,
+            decodePath: request.decodePath
+        )
     }
 
     public func configure(using configuration: Any?) throws { }
 
     public func apiAuthProviderFactory() -> APIAuthProviderFactory {
-        preconditionFailure("method not supported")
+        return self.apiAuthFactory
     }
 
     public func add(interceptor: any URLRequestInterceptor, for apiName: String) throws {
@@ -67,7 +144,7 @@ public class FlutterApiPlugin: APICategoryPlugin
     }
 
     public func reachabilityPublisher() throws -> AnyPublisher<ReachabilityUpdate, Never>? {
-        preconditionFailure("method not supported")
+        return nil
     }
 
 

packages/amplify_datastore/ios/Classes/SwiftAmplifyDataStorePlugin.swift

@@ -14,6 +14,7 @@ public class SwiftAmplifyDataStorePlugin: NSObject, FlutterPlugin, NativeAmplify
     private let customTypeSchemaRegistry: FlutterSchemaRegistry
     private let dataStoreObserveEventStreamHandler: DataStoreObserveEventStreamHandler?
     private let dataStoreHubEventStreamHandler: DataStoreHubEventStreamHandler?
+    private let nativeSubscriptionEventBus = PassthroughSubject<NativeGraphQLSubscriptionResponse, Never>()
     private var channel: FlutterMethodChannel?
     private var observeSubscription: AnyCancellable?
     private let nativeAuthPlugin: NativeAuthPlugin
@@ -88,7 +89,14 @@ public class SwiftAmplifyDataStorePlugin: NSObject, FlutterPlugin, NativeAmplify
                 AWSAuthorizationType(rawValue: $0)
             }
             try Amplify.add(
-                plugin: FlutterApiPlugin()
+                plugin: FlutterApiPlugin(
+                    apiAuthProviderFactory: FlutterAuthProviders(
+                        authProviders: authProviders,
+                        nativeApiPlugin: nativeApiPlugin
+                    ), 
+                    nativeApiPlugin: nativeApiPlugin,
+                    subscriptionEventBus: nativeSubscriptionEventBus
+                )
             )
             return completion(.success(()))
         } catch let apiError as APIError {
@@ -608,7 +616,7 @@ public class SwiftAmplifyDataStorePlugin: NSObject, FlutterPlugin, NativeAmplify
     }
 
     func sendSubscriptionEvent(event: NativeGraphQLSubscriptionResponse, completion: @escaping (Result<Void, any Error>) -> Void) {
-        fatalError("not implemented")
+        nativeSubscriptionEventBus.send(event)
     }
 
     private func checkArguments(args: Any) throws -> [String: Any] {

packages/amplify_datastore/ios/Classes/api/GraphQLRequest+Extension.swift

@@ -0,0 +1,26 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+
+import Foundation
+import Amplify
+
+extension GraphQLRequest {
+    func toNativeGraphQLRequest() -> NativeGraphQLRequest {
+        let variablesJson = self.variables
+            .flatMap { try? JSONSerialization.data(withJSONObject: $0, options: []) }
+            .flatMap { String(data: $0, encoding: .utf8) }
+
+        return NativeGraphQLRequest(
+            document: self.document,
+            apiName: self.apiName,
+            variablesJson: variablesJson ?? "{}",
+            responseType: String(describing: self.responseType),
+            decodePath: self.decodePath
+        )
+    }
+}

packages/amplify_datastore/ios/Classes/api/GraphQLResponse+Decode.swift

@@ -0,0 +1,168 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+
+import Foundation
+import Amplify
+import AWSPluginsCore
+
+extension GraphQLResponse {
+    static var jsonDecoder: JSONDecoder {
+        let decoder = JSONDecoder(dateDecodingStrategy: ModelDateFormatting.decodingStrategy)
+        return decoder
+    }
+
+    static var jsonEncoder: JSONEncoder {
+        let encoder = JSONEncoder(dateEncodingStrategy: ModelDateFormatting.encodingStrategy)
+        return encoder
+    }
+
+    public static func fromAppSyncResponse<R: Decodable>(
+        string: String,
+        decodePath: String?,
+        modelName: String? = nil
+    ) -> GraphQLResponse<R> {
+        guard let data = string.data(using: .utf8) else {
+            return .failure(.transformationError(
+                string,
+                .operationError("Unable to deserialize json data", "Check the event structure.", nil)
+            ))
+        }
+        return fromAppSyncResponse(data: data, decodePath: decodePath, modelName: modelName)
+    }
+
+    public static func fromAppSyncResponse<R: Decodable>(
+        data: Data,
+        decodePath: String?,
+        modelName: String? = nil
+    ) -> GraphQLResponse<R> {
+        toJson(data: data)
+            .flatMap { fromAppSyncResponse(json: $0, decodePath: decodePath, modelName: modelName) }
+            .mapError {
+                if let response = String(data: data, encoding: .utf8) {
+                    return .transformationError(response, $0)
+                } else {
+                    return .transformationError("Response is not string encodable", $0)
+                }
+            }
+    }
+
+    static func fromAppSyncResponse<R: Decodable>(
+        json: JSONValue,
+        decodePath: String?,
+        modelName: String?
+    ) -> Result<R, APIError> {
+        if let decodePath {
+            if let payload = json.value(at: decodePath) {
+                return decodeDataPayload(payload, modelName: modelName)
+            } else {
+                return .failure(.operationError("Empty data on decode path \(decodePath)", "", nil))
+            }
+        } else {
+            return decodeDataPayload(json, modelName: modelName)
+        }
+    }
+
+    static func decodeDataPayload<R: Decodable>(
+        _ dataPayload: JSONValue,
+        modelName: String?
+    ) -> Result<R, APIError> {
+        if R.Type.self == String.self {
+            return decodeDataPayloadToString(dataPayload).map { $0 as! R }
+        }
+
+        let dataPayloadWithTypeName = modelName.flatMap {
+            dataPayload.asObject?.merging(
+                ["__typename": .string($0)]
+            ) { a, _ in a }
+        }.map { JSONValue.object($0) } ?? dataPayload
+
+        if R.Type.self == AnyModel.self {
+            return decodeDataPayloadToAnyModel(dataPayloadWithTypeName).map { $0 as! R }
+        }
+
+        return fromJson(dataPayloadWithTypeName)
+            .flatMap { data in
+                Result<R, Error> { try jsonDecoder.decode(R.self, from: data) }
+                    .mapError { APIError.operationError("Could not decode json to type \(R.self)", "", $0)}
+            }
+    }
+
+    static func decodeDataPayloadToAnyModel(
+        _ dataPaylod: JSONValue
+    ) -> Result<AnyModel, APIError> {
+        guard let typeName = dataPaylod.__typeName?.stringValue else {
+            return .failure(.operationError(
+                "Could not retrieve __typeName from object",
+                """
+                Could not retrieve the `__typename` attribute from the return value. Be sure to include __typename in \
+                the selection set of the GraphQL operation. GraphQL:
+                \(dataPaylod)
+                """
+            ))
+        }
+
+        return decodeDataPayloadToString(dataPaylod).flatMap { underlyingModelString in
+            do {
+                return .success(.init(try ModelRegistry.decode(
+                    modelName: typeName,
+                    from: underlyingModelString,
+                    jsonDecoder: jsonDecoder
+                )))
+            } catch {
+                return .failure(.operationError(
+                    "Could not decode to \(typeName) with \(underlyingModelString)",
+                    ""
+                ))
+            }
+        }
+    }
+
+    static func decodeDataPayloadToString(
+        _ dataPayload: JSONValue
+    ) -> Result<String, APIError> {
+        do {
+            let data = try jsonEncoder.encode(dataPayload)
+            guard let string = String(data: data, encoding: .utf8) else {
+                return .failure(
+                    .operationError("Could not get String from Data", "", nil)
+                )
+            }
+            return .success(string)
+        } catch {
+            return .failure(.operationError(
+                "Could not get the String representation of the GraphQL response",
+                ""
+            ))
+        }
+    }
+
+    static func toJson(data: Data) -> Result<JSONValue, APIError> {
+        do {
+            return .success(try jsonDecoder.decode(JSONValue.self, from: data))
+        } catch {
+            return .failure(.operationError(
+                "Could not decode to JSONValue from GraphQL Response",
+                "Service issue",
+                error
+            ))
+        }
+    }
+
+    static func fromJson(_ json: JSONValue) -> Result<Data, APIError> {
+        do {
+            return .success(try jsonEncoder.encode(json))
+        } catch {
+            return .failure(.operationError(
+                "Could not encode JSONValue to Data",
+                "",
+                error
+            ))
+        }
+    }
+
+}

packages/amplify_datastore/ios/Classes/api/Publisher+Extension.swift

@@ -0,0 +1,29 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+
+import Foundation
+import Combine
+import Amplify
+
+extension Publisher {
+    func toAmplifyAsyncThrowingSequence() -> (AmplifyAsyncThrowingSequence<Output>, AnyCancellable) {
+        let sequence = AmplifyAsyncThrowingSequence<Output>()
+        let cancellable = self.sink { completion in
+            switch completion {
+            case .finished:
+                sequence.finish()
+            case .failure(let error):
+                sequence.fail(error)
+            }
+        } receiveValue: { data in
+            sequence.send(data)
+        }
+
+        return (sequence, cancellable)
+    }
+}

packages/amplify_datastore/ios/Classes/types/hub/FlutterHubElement.swift

@@ -52,11 +52,11 @@ public struct FlutterHubElement {
             self.version = hubElement.version
             self.deleted = self.model["_deleted"] as? Bool ?? false
             if let value = self.model["_lastChangedAt"] as? Double {
-                self.lastChangedAt = Int(value)
+                self.lastChangedAt = Int64(value)
             } else if let value = self.model["_lastChangedAt"] as? String {
-                self.lastChangedAt = Int(value)
+                self.lastChangedAt = Int64(value)
             } else if let value = self.model["_lastChangedAt"] as? Int {
-                self.lastChangedAt = value
+                self.lastChangedAt = Int64(value)
             }
         } catch {
             throw FlutterDataStoreError.hubEventCast