fun subscribe()

in packages/amplify_api/android/src/main/kotlin/com/amazonaws/amplify/amplify_api/FlutterGraphQLApi.kt [196:304]
102 lines of code
5 McCabe index (conditional complexity)

    fun subscribe(
        flutterResult: MethodChannel.Result,
        request: Map<String, Any>,
        graphqlSubscriptionStreamHandler: GraphQLSubscriptionStreamHandler
    ) {
        val apiName: String?
        val document: String
        val variables: Map<String, Any>
        val id: String
        var established = false

        try {
            apiName = FlutterApiRequest.getGraphQlApiName(request)
            document = FlutterApiRequest.getGraphQLDocument(request)
            variables = FlutterApiRequest.getVariables(request)
            id = FlutterApiRequest.getCancelToken(request)
        } catch (e: Exception) {
            handler.post {
                ExceptionUtil.postExceptionToFlutterChannel(
                    flutterResult, "ApiException",
                    ExceptionUtil.createSerializedUnrecognizedError(e)
                )
            }
            return
        }

        val connectionCallback = Consumer<String> {
            established = true
            LOG.debug("Subscription established: $id")
            handler.post { flutterResult.success(null) }
        }

        val responseCallback = Consumer<GraphQLResponse<String>> { response ->
            val payload: Map<String, Any> = mapOf(
                "data" to response.data,
                "errors" to response.errors.map { it.toMap() }
            )
            LOG.debug("GraphQL subscription event received: $payload")
            graphqlSubscriptionStreamHandler.sendEvent(
                payload,
                id,
                GraphQLSubscriptionEventTypes.DATA
            )
        }

        val errorCallback = Consumer<ApiException> {
            OperationsManager.removeOperation(id)
            if (established) {
                graphqlSubscriptionStreamHandler.sendError(
                    "ApiException",
                    id,
                    ExceptionUtil.createSerializedError(it)
                )
            } else {
                handler.post {
                    ExceptionUtil.postExceptionToFlutterChannel(
                        flutterResult, "ApiException",
                        ExceptionUtil.createSerializedError(it)
                    )
                }
            }
        }

        val disconnectionCallback = Action {
            OperationsManager.removeOperation(id)
            LOG.debug("Subscription has been closed successfully")
            graphqlSubscriptionStreamHandler.sendEvent(
                null,
                id,
                GraphQLSubscriptionEventTypes.DONE
            )
        }

        scope.launch(dispatcher) {
            val operation: GraphQLOperation<String?>?

            if (apiName != null) {
                operation = Amplify.API.subscribe(
                    apiName,
                    SimpleGraphQLRequest<String>(
                        document,
                        variables,
                        String::class.java,
                        GsonVariablesSerializer()
                    ),
                    connectionCallback,
                    responseCallback,
                    errorCallback,
                    disconnectionCallback
                )
            } else {
                operation = Amplify.API.subscribe(
                    SimpleGraphQLRequest<String>(
                        document,
                        variables,
                        String::class.java,
                        GsonVariablesSerializer()
                    ),
                    connectionCallback,
                    responseCallback,
                    errorCallback,
                    disconnectionCallback
                )
            }
            if (operation != null) {
                OperationsManager.addOperation(id, operation)
            }
        }
    }