How to use this with Apollo Server?

[wmertens]

Title says it all, how to integrate? Put the middleware first and ignore Apollo altogether? But then it can't use features like persisted queries.

[enisdenjo]

Sadly, I don't have much experience with Apollo's tooling. So I cannot really say whether graphql-sse can be integrated in such a manner where all other Apollo middleware is reused.

Help or suggestions is more than welcome!

[wmertens]

We wrote our own graphql SSE thing, before we found this project, and ours is an apollo plugin on the server side. Unfortunately it doesn't detect closed channels :-(

Here's what the server side looks like:

import {GraphQLError, subscribe} from 'graphql'
import type {
	ApolloServerPlugin,
	GraphQLRequestContextResponseForOperation,
	GraphQLRequestListener,
	GraphQLResponse,
} from '@apollo/server'
import {ApolloServerErrorCode} from '@apollo/server/errors'

const isSubscription = ({
	operation,
}: GraphQLRequestContextResponseForOperation<GqlContext>) =>
	operation.operation === 'subscription'

const isAsyncIterator = (iter): iter is AsyncIterator<any> =>
	typeof iter[Symbol.asyncIterator] === 'function'

const handleSubscription = async (
	arg: GraphQLRequestContextResponseForOperation<GqlContext>
): Promise<GraphQLResponse | null> => {
	if (!isSubscription(arg)) return null
	const url = arg.request.http?.search || ''
	const match = /^\??.*s=(\d+)&t=(.+)(&|$)/.exec(url)
	if (!match) throw new Error('s and t query parameters missing on POST')
	const [, subId, token] = match
	const channel = state.byToken.get(token)
	if (!channel) throw new Error('unknown channel')
	const iter = await subscribe(arg)
	// if not asynciterator, it's an error
	if (!isAsyncIterator(iter))
		throw new GraphQLError('Subscription failed', {
			extensions: {
				code: ApolloServerErrorCode.INTERNAL_SERVER_ERROR,
				errors: iter.errors,
			},
		})

	// hook up iterator to channel & end request
	const handler = async () => {
		// TODO close iter on channel close, how to signal?
		try {
			for await (const value of iter) channel.send([subId, value])
			channel.send([subId], 'end')
		} catch (err) {
			channel.send([subId, err.message], 'error')
		}
	}
	handler()
	// Provide empty response so Apollo finishes processing
	return null
}

const pluginStatic = {responseForOperation: handleSubscription}
export const ssePlugin: ApolloServerPlugin = {
	requestDidStart:
		async (): Promise<void | GraphQLRequestListener<GqlContext>> =>
			pluginStatic,
}

[enisdenjo]

As I said, I have no experience with Apollo's tooling - I cannot help you with debugging your code, sorry.

[wmertens]

I'd rather join forces and make your code also work as an Apollo Plugin :-)

Basically, the plugin gets a query, checks if it's a subscription, extracts the information to find the return channel, and runs the subscription.

Because it's a plugin, it has persistent query support and anything else that's set up

[enisdenjo]

graphql-sse is rather simple, it takes a request and returns a response. You can hook into its execution model through the onSubscribe hook where you can execute the rest of middlewares/plugins of Apollo (I assume).

Also read about the rest of the hooks in the handler.