graphql/yoga-server/docs/features/subscriptions

[giscus[bot]]

graphql/yoga-server/docs/features/subscriptions

GraphQL Yoga Documentation

https://www.the-guild.dev/graphql/yoga-server/docs/features/subscriptions

[jhanschoo]

Sample test using yoga.fetch a la https://www.the-guild.dev/graphql/yoga-server/v3/features/testing

export async function testSubscription<TVariables>({
  yoga,
  document,
  variables,
}: {
  yoga: Server;
  document: string;
  variables: TVariables;
}) {
  const uri = encodeURI(`http://localhost:3000/graphql?query=${document}${variables ? `&variables=${JSON.stringify(variables)}` : ''}`)
  const response = server.fetch(uri, {
    method: 'GET',
    headers: {
      accept: 'text/event-stream',
    },
  });
  return response;
}


const response = await testSubscription({ yoga, document: 'subscription Health { health }' });
if (!response.body) {
  return;
}
for await (const chunk of response.body) {
  const event = JSON.parse(chunk.toString().slice(6));
  expect(event).toHaveProperty("data.health", eventTest());
}

[jhanschoo]

Note that this comment has since been superceded by current documentation at https://the-guild.dev/graphql/yoga-server/docs/features/testing

[jhanschoo]

In the current release (3.7.3) and I suspect earlier releases, the library's PubSub has this unexpected quirk in its behavior; even after the subscribe(...) object has been returned, the subscriber doesn't listen for publish events until after its first .next() call. Following that, future publish events are buffered even before the subsequent .next() calls. That is, the following Jest tests all pass:

import { createPubSub, PubSub } from 'graphql-yoga'

describe('pubsub', () => {
  let pubsub: PubSub<{ test: [string] }>;
  beforeAll(() => {
    pubsub = createPubSub();
  });
  it('a message published before a subscriber creates its promise is lost', async () => {
    expect.assertions(1);
    const subscriber = pubsub.subscribe('test');
    pubsub.publish('test', 'message');
    let messageReceived = false;
    subscriber.next().then(() => {
      messageReceived = true;
    });
    await new Promise((resolve) => setTimeout(resolve, 1000));
    expect(messageReceived).toBe(false);
  });
  it('a subscriber obtains a message published after its promise has been created', async () => {
    expect.assertions(1);
    const subscriber = pubsub.subscribe('test');
    pubsub.publish('test', 'message 1');
    const messageP = subscriber.next();
    pubsub.publish('test', 'message 2');
    expect(await messageP).toHaveProperty('value', 'message 2');
  });
  it('subscribers\' iterates are resolved in order as long as their respective promises have been pending before being published', async () => {
    expect.assertions(4);
    const subscriber = pubsub.subscribe('test');
    pubsub.publish('test', 'message 1');
    const message2P = subscriber.next();
    const message3P = subscriber.next();
    const message4P = subscriber.next();
    const message5P = subscriber.next();
    pubsub.publish('test', 'message 2');
    pubsub.publish('test', 'message 3');
    pubsub.publish('test', 'message 4');
    pubsub.publish('test', 'message 5');
    expect(await message2P).toHaveProperty('value', 'message 2');
    expect(await message3P).toHaveProperty('value', 'message 3');
    expect(await message4P).toHaveProperty('value', 'message 4');
    expect(await message5P).toHaveProperty('value', 'message 5');
  });
  it('a message published after the first .next() call but before the next .next() call is buffered and not lost', async () => {
    expect.assertions(4);
    const subscriber = pubsub.subscribe('test');
    pubsub.publish('test', 'message 1');
    const message2P = subscriber.next();
    pubsub.publish('test', 'message 2');
    pubsub.publish('test', 'message 3');
    pubsub.publish('test', 'message 4');
    pubsub.publish('test', 'message 5');
    const message3P = subscriber.next();
    const message4P = subscriber.next();
    const message5P = subscriber.next();
    expect(await message2P).toHaveProperty('value', 'message 2');
    expect(await message3P).toHaveProperty('value', 'message 3');
    expect(await message4P).toHaveProperty('value', 'message 4');
    expect(await message5P).toHaveProperty('value', 'message 5');
  });
});

[n1ru4l]

That is expected behavior. The PubSub subscriptions are lazy to prevent memory leaks. See https://repeater.js.org/

[jhanschoo]

Indeed; though considering that the implementation is meant to be opaque, I thought that I should make a note of it

[kudresov]

      Subscription: {
        globalCounter: {
          // Merge initial value with source stream of new values
          subscribe: () =>
            pipe(
              Repeater.merge([
                // cause an initial event so the
                // globalCounter is streamed to the client
                // upon initiating the subscription
                undefined,
                // event stream for future updates
                pubSub.subscribe('globalCounter:change')
              ]),
              // map all stream values to the latest globalCounter
              map(() => globalCounter)
            ),
          resolve: payload => payload
        }
      },

I have tried this example and it does seem to work, I had to modify this to something along those line.

      Subscription: {
        globalCounter: {
          // Merge initial value with source stream of new values
          subscribe: () =>
            pipe(
              Repeater.merge([
                // cause an initial event so the
                // globalCounter is streamed to the client
                // upon initiating the subscription
                globalCounter,
                // event stream for future updates
                pubSub.subscribe('globalCounter:change')
              ]),
              // map all stream values to the latest globalCounter
              map((counter) => counter)
            ),
          resolve: payload => payload
        }
      },

[kudresov]

Sorry posted my comment as reply accidentally. I am trying an example and one part isn't working:

     Subscription: {
        globalCounter: {
          // Merge initial value with source stream of new values
          subscribe: () =>
            pipe(
              Repeater.merge([
                // cause an initial event so the
                // globalCounter is streamed to the client
                // upon initiating the subscription
                undefined,
                // event stream for future updates
                pubSub.subscribe('globalCounter:change')
              ]),
              // map all stream values to the latest globalCounter
              map(() => globalCounter)
            ),
          resolve: payload => payload
        }
      },

I have tried this example and it does seem to work, I had to modify this to something along those line.

      Subscription: {
        globalCounter: {
          // Merge initial value with source stream of new values
          subscribe: () =>
            pipe(
              Repeater.merge([
                // cause an initial event so the
                // globalCounter is streamed to the client
                // upon initiating the subscription
                globalCounter,
                // event stream for future updates
                pubSub.subscribe('globalCounter:change')
              ]),
              // map all stream values to the latest globalCounter
              map((counter) => counter)
            ),
          resolve: payload => payload
        }
      },

Does that sounds right?

[petarbiocic-pontis]

Hi.

Is there a way to combine express with websockets, for yoga? I can see the example on this page but it uses plain nodejs.

[ariefsn]

You can try this.

const createWebSocketServer = (server: Server<typeof IncomingMessage, typeof ServerResponse>) => {
  const webSocketServer = new WebSocketServer({
    server,
    path: yoga.graphqlEndpoint
  })

  useServer({
    execute(args) {
      return (args.rootValue as any).execute(args)
    },
    subscribe(args) {
      return (args.rootValue as any).subscribe(args)
    },
    onSubscribe: async (ctx, _id, params) => {
      const { schema, execute, subscribe, contextFactory, parse, validate } = yoga.getEnveloped({
        ...ctx,
        req: ctx.extra.request,
        socket: ctx.extra.socket,
        params
      })

      const args = {
        schema,
        operationName: params.operationName,
        document: parse(params.query),
        variableValues: params.variables,
        contextValue: await contextFactory(),
        rootValue: {
          execute,
          subscribe
        }
      }

      const errors = validate(args.schema, args.document)
      if (errors.length) return errors
      return args
    }
  }, webSocketServer)
}

and in your express server:

const server = app.listen(port, () => {
  createWebSocketServer(server)

  console.log(`[server]: Server is running at ${port}`);
});