<commit_before>import Dispatch
import Runtime
import RxSwift
import NIO

/**
 * Implements the "Subscribe" algorithm described in the GraphQL specification.
 *
 * Returns a Promise which resolves to either an AsyncIterator (if successful)
 * or an ExecutionResult (error). The promise will be rejected if the schema or
 * other arguments to this function are invalid, or if the resolved event stream
 * is not an async iterable.
 *
 * If the client-provided arguments to this function do not result in a
 * compliant subscription, a GraphQL Response (ExecutionResult) with
 * descriptive errors and no data will be returned.
 *
 * If the source stream could not be created due to faulty subscription
 * resolver logic or underlying systems, the promise will resolve to a single
 * ExecutionResult containing `errors` and no `data`.
 *
 * If the operation succeeded, the promise resolves to an AsyncIterator, which
 * yields a stream of ExecutionResults representing the response stream.
 *
 * Accepts either an object with named arguments, or individual arguments.
 */
func subscribe(
    queryStrategy: QueryFieldExecutionStrategy,
    mutationStrategy: MutationFieldExecutionStrategy,
    subscriptionStrategy: SubscriptionFieldExecutionStrategy,
    instrumentation: Instrumentation,
    schema: GraphQLSchema,
    documentAST: Document,
    rootValue: Any,
    context: Any,
    eventLoopGroup: EventLoopGroup,
    variableValues: [String: Map] = [:],
    operationName: String? = nil,
    fieldResolver: GraphQLFieldResolve? = nil,
    subscribeFieldResolver: GraphQLFieldResolve? = nil
) -> EventLoopFuture<SubscriptionResult> {
    
    
    let sourceFuture = createSourceEventStream(
        queryStrategy: queryStrategy,
        mutationStrategy: mutationStrategy,
        subscriptionStrategy: subscriptionStrategy,
        instrumentation: instrumentation,
        schema: schema,
        documentAST: documentAST,
        rootValue: rootValue,
        context: context,
        eventLoopGroup: eventLoopGroup,
        variableValues: variableValues,
        operationName: operationName,
        subscribeFieldResolver: subscribeFieldResolver
    )

    // For each payload yielded from a subscription, map it over the normal
    // GraphQL `execute` function, with `payload` as the rootValue.
    // This implements the "MapSourceToResponseEvent" algorithm described in
    // the GraphQL specification. The `execute` function provides the
    // "ExecuteSubscriptionEvent" algorithm, as it is nearly identical to the
    // "ExecuteQuery" algorithm, for which `execute` is also used.
    func mapSourceToResponse(payload: Any) -> EventLoopFuture<GraphQLResult> {
        return execute(
            queryStrategy: queryStrategy,
            mutationStrategy: mutationStrategy,
            subscriptionStrategy: subscriptionStrategy,
            instrumentation: instrumentation,
            schema: schema,
            documentAST: documentAST,
            rootValue: payload, // Make payload the root value
            context: context,
            eventLoopGroup: eventLoopGroup,
            variableValues: variableValues,
            operationName: operationName
        )
    }
    return sourceFuture.flatMap{ subscriptionResult -> EventLoopFuture<SubscriptionResult> in
        do {
            let subscriptionObserver = try subscriptionResult.get()
            let eventObserver = subscriptionObserver.map { eventPayload -> GraphQLResult in
                return try! mapSourceToResponse(payload: eventPayload).wait() // TODO Remove this wait
            }
            // TODO Making a future here feels it indicates a mistake...
            return eventLoopGroup.next().makeSucceededFuture(SubscriptionResult.success(eventObserver))
        } catch let graphQLError as GraphQLError {
            return eventLoopGroup.next().makeSucceededFuture(SubscriptionResult.failure(graphQLError))
        }
    }
}

/**
 * Implements the "CreateSourceEventStream" algorithm described in the
 * GraphQL specification, resolving the subscription source event stream.
 *
 * Returns a Promise which resolves to either an AsyncIterable (if successful)
 * or an ExecutionResult (error). The promise will be rejected if the schema or
 * other arguments to this function are invalid, or if the resolved event stream
 * is not an async iterable.
 *
 * If the client-provided arguments to this function do not result in a
 * compliant subscription, a GraphQL Response (ExecutionResult) with
 * descriptive errors and no data will be returned.
 *
 * If the the source stream could not be created due to faulty subscription
 * resolver logic or underlying systems, the promise will resolve to a single
 * ExecutionResult containing `errors` and no `data`.
 *
 * If the operation succeeded, the promise resolves to the AsyncIterable for the
 * event stream returned by the resolver.
 *
 * A Source Event Stream represents a sequence of events, each of which triggers
 * a GraphQL execution for that event.
 *
 * This may be useful when hosting the stateful subscription service in a
 * different process or machine than the stateless GraphQL execution engine,
 * or otherwise separating these two steps. For more on this, see the
 * "Supporting Subscriptions at Scale" information in the GraphQL specification.
 */
func createSourceEventStream(
    queryStrategy: QueryFieldExecutionStrategy,
    mutationStrategy: MutationFieldExecutionStrategy,
    subscriptionStrategy: SubscriptionFieldExecutionStrategy,
    instrumentation: Instrumentation,
    schema: GraphQLSchema,
    documentAST: Document,
    rootValue: Any,
    context: Any,
    eventLoopGroup: EventLoopGroup,
    variableValues: [String: Map] = [:],
    operationName: String? = nil,
    subscribeFieldResolver: GraphQLFieldResolve? = nil
) -> EventLoopFuture<SourceEventStreamResult> {

    let executeStarted = instrumentation.now
    
    do {
        // If a valid context cannot be created due to incorrect arguments,
        // this will throw an error.
        let exeContext = try buildExecutionContext(
            queryStrategy: queryStrategy,
            mutationStrategy: mutationStrategy,
            subscriptionStrategy: subscriptionStrategy,
            instrumentation: instrumentation,
            schema: schema,
            documentAST: documentAST,
            rootValue: rootValue,
            context: context,
            eventLoopGroup: eventLoopGroup,
            rawVariableValues: variableValues,
            operationName: operationName
            // TODO shouldn't we be including the subscribeFieldResolver??
        )
        return try executeSubscription(context: exeContext, eventLoopGroup: eventLoopGroup)<commit_msg>Minor refactor of event resolver for clarity<commit_after>import Dispatch
import Runtime
import RxSwift
import NIO

/**
 * Implements the "Subscribe" algorithm described in the GraphQL specification.
 *
 * Returns a Promise which resolves to either an AsyncIterator (if successful)
 * or an ExecutionResult (error). The promise will be rejected if the schema or
 * other arguments to this function are invalid, or if the resolved event stream
 * is not an async iterable.
 *
 * If the client-provided arguments to this function do not result in a
 * compliant subscription, a GraphQL Response (ExecutionResult) with
 * descriptive errors and no data will be returned.
 *
 * If the source stream could not be created due to faulty subscription
 * resolver logic or underlying systems, the promise will resolve to a single
 * ExecutionResult containing `errors` and no `data`.
 *
 * If the operation succeeded, the promise resolves to an AsyncIterator, which
 * yields a stream of ExecutionResults representing the response stream.
 *
 * Accepts either an object with named arguments, or individual arguments.
 */
func subscribe(
    queryStrategy: QueryFieldExecutionStrategy,
    mutationStrategy: MutationFieldExecutionStrategy,
    subscriptionStrategy: SubscriptionFieldExecutionStrategy,
    instrumentation: Instrumentation,
    schema: GraphQLSchema,
    documentAST: Document,
    rootValue: Any,
    context: Any,
    eventLoopGroup: EventLoopGroup,
    variableValues: [String: Map] = [:],
    operationName: String? = nil,
    fieldResolver: GraphQLFieldResolve? = nil,
    subscribeFieldResolver: GraphQLFieldResolve? = nil
) -> EventLoopFuture<SubscriptionResult> {
    
    
    let sourceFuture = createSourceEventStream(
        queryStrategy: queryStrategy,
        mutationStrategy: mutationStrategy,
        subscriptionStrategy: subscriptionStrategy,
        instrumentation: instrumentation,
        schema: schema,
        documentAST: documentAST,
        rootValue: rootValue,
        context: context,
        eventLoopGroup: eventLoopGroup,
        variableValues: variableValues,
        operationName: operationName,
        subscribeFieldResolver: subscribeFieldResolver
    )
    
    return sourceFuture.flatMap{ subscriptionResult -> EventLoopFuture<SubscriptionResult> in
        do {
            let subscriptionObserver = try subscriptionResult.get()
            let eventObserver = subscriptionObserver.map { eventPayload -> GraphQLResult in
                
                // For each payload yielded from a subscription, map it over the normal
                // GraphQL `execute` function, with `payload` as the rootValue.
                // This implements the "MapSourceToResponseEvent" algorithm described in
                // the GraphQL specification. The `execute` function provides the
                // "ExecuteSubscriptionEvent" algorithm, as it is nearly identical to the
                // "ExecuteQuery" algorithm, for which `execute` is also used.
                let eventResolved = try execute(
                    queryStrategy: queryStrategy,
                    mutationStrategy: mutationStrategy,
                    subscriptionStrategy: subscriptionStrategy,
                    instrumentation: instrumentation,
                    schema: schema,
                    documentAST: documentAST,
                    rootValue: eventPayload,
                    context: context,
                    eventLoopGroup: eventLoopGroup,
                    variableValues: variableValues,
                    operationName: operationName
                ).wait() // TODO remove this wait
                
                return eventResolved
            }
            // TODO Making a future here feels it indicates a mistake...
            return eventLoopGroup.next().makeSucceededFuture(SubscriptionResult.success(eventObserver))
        } catch let graphQLError as GraphQLError {
            return eventLoopGroup.next().makeSucceededFuture(SubscriptionResult.failure(graphQLError))
        }
    }
}

/**
 * Implements the "CreateSourceEventStream" algorithm described in the
 * GraphQL specification, resolving the subscription source event stream.
 *
 * Returns a Promise which resolves to either an AsyncIterable (if successful)
 * or an ExecutionResult (error). The promise will be rejected if the schema or
 * other arguments to this function are invalid, or if the resolved event stream
 * is not an async iterable.
 *
 * If the client-provided arguments to this function do not result in a
 * compliant subscription, a GraphQL Response (ExecutionResult) with
 * descriptive errors and no data will be returned.
 *
 * If the the source stream could not be created due to faulty subscription
 * resolver logic or underlying systems, the promise will resolve to a single
 * ExecutionResult containing `errors` and no `data`.
 *
 * If the operation succeeded, the promise resolves to the AsyncIterable for the
 * event stream returned by the resolver.
 *
 * A Source Event Stream represents a sequence of events, each of which triggers
 * a GraphQL execution for that event.
 *
 * This may be useful when hosting the stateful subscription service in a
 * different process or machine than the stateless GraphQL execution engine,
 * or otherwise separating these two steps. For more on this, see the
 * "Supporting Subscriptions at Scale" information in the GraphQL specification.
 */
func createSourceEventStream(
    queryStrategy: QueryFieldExecutionStrategy,
    mutationStrategy: MutationFieldExecutionStrategy,
    subscriptionStrategy: SubscriptionFieldExecutionStrategy,
    instrumentation: Instrumentation,
    schema: GraphQLSchema,
    documentAST: Document,
    rootValue: Any,
    context: Any,
    eventLoopGroup: EventLoopGroup,
    variableValues: [String: Map] = [:],
    operationName: String? = nil,
    subscribeFieldResolver: GraphQLFieldResolve? = nil
) -> EventLoopFuture<SourceEventStreamResult> {

    let executeStarted = instrumentation.now
    
    do {
        // If a valid context cannot be created due to incorrect arguments,
        // this will throw an error.
        let exeContext = try buildExecutionContext(
            queryStrategy: queryStrategy,
            mutationStrategy: mutationStrategy,
            subscriptionStrategy: subscriptionStrategy,
            instrumentation: instrumentation,
            schema: schema,
            documentAST: documentAST,
            rootValue: rootValue,
            context: context,
            eventLoopGroup: eventLoopGroup,
            rawVariableValues: variableValues,
            operationName: operationName
        )
        return try executeSubscription(context: exeContext, eventLoopGroup: eventLoopGroup)
{'fraction_non_alphanumeric': 0.056717081850533806, 'fraction_numerical': 0.0, 'mean_word_length': 3.1276009791921666, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6709706', 'n_tokens_mistral': 3098, 'n_tokens_neox': 3026, 'n_words': 1228}