How to use graphql-subscriptions - 10 common examples

it("should create resolver with tags & filter", async () => {
			svc.pubsub.asyncIterator.mockClear();
			broker.call.mockClear();
			withFilter.mockImplementation((fn1, fn2) => [fn1, fn2]);

			const res = svc.createAsyncIteratorResolver("posts.find", ["a", "b"], "posts.filter");

			expect(res).toEqual({
				subscribe: [expect.any(Function), expect.any(Function)],
				resolve: expect.any(Function),
			});

			// Test first function
			expect(res.subscribe[0]()).toBe("iterator-result");

			expect(svc.pubsub.asyncIterator).toBeCalledTimes(1);
			expect(svc.pubsub.asyncIterator).toBeCalledWith(["a", "b"]);

			// Test second function without payload
			expect(await res.subscribe[1]()).toBe(false);

});

  const generateConfig = await config!.getDefault().extension('generate');

  // connect to db
  const db = await connect(generateConfig.db.dbConfig);

  const schema = makeExecutableSchema({
    typeDefs,
    resolvers,
    resolverValidationOptions: {
      requireResolversForResolveType: false
    }
  });

  const pubSub = new PubSub();
  const context = createKnexRuntimeContext(db, pubSub);
  const apolloConfig = {
    schema,
    context
  }

  const apolloServer = new ApolloServer(apolloConfig)

  apolloServer.applyMiddleware({ app })

  const httpServer = http.createServer(app)
  apolloServer.installSubscriptionHandlers(httpServer)

  httpServer.listen({ port: 4000 }, () => {
    // tslint:disable-next-line: no-console
    console.log(`🚀  Server ready at http://localhost:4000/graphql`)

export const TeamsSubscriptions = {
  teamsUpdate: {
    resolve(rootValue, { simulatorId, type, cleared }) {
      // Get the simulator
      let returnVal = rootValue;
      if (type) {
        returnVal = returnVal.filter(t => t.type === type);
      }
      if (simulatorId) {
        returnVal = returnVal.filter(t => t.simulatorId === simulatorId);
      }
      if (cleared) return returnVal;
      return returnVal.filter(c => !c.cleared);
    },
    subscribe: withFilter(
      () => pubsub.asyncIterator("teamsUpdate"),
      (rootValue, { simulatorId, type }) => true
    )
  }
};

export const TeamsTypes = {
  Team: {
    location(team) {
      const deck = App.decks.find(d => d.id === team.location);
      if (deck) {
        return deck;
      }
      return App.rooms.find(r => r.id === team.location);
    },
    officers(team) {

import { withFilter } from 'graphql-subscriptions';
import { getThread } from '../models/thread';
import { userCanViewChannel, userCanViewDirectMessageThread } from './utils';
const { listenToNewMessages } = require('../models/message');
import asyncify from '../utils/asyncify';
import { trackUserThreadLastSeenQueue } from 'shared/bull/queues.js';
import type { Message } from '../models/message';

/**
 * Define the message subscription resolvers
 */
module.exports = {
  Subscription: {
    messageAdded: {
      resolve: (message: any) => message,
      subscribe: withFilter(
        asyncify(listenToNewMessages, err => {
          throw new Error(err);
        }),
        async (message, { thread }, { user }) => {
          // Message was sent in different thread, early return
          if (message.threadId !== thread) return false;
          if (message.threadType === 'directMessageThread') {
            if (!user) return false;
            return userCanViewDirectMessageThread(message.threadId, user.id);
          }

          const threadData = await getThread(message.threadId);
          if (!threadData) return false;

          return await userCanViewChannel(
            threadData.channelId,

}
})


export default {
  // auth stuffs
  newComment: {
    type: CommentType,
    args: {
      input: {
        type: NewCommentSubscriptionInput
      }
    },
    // resolve have to be there... dont know why yet
    resolve: data => data,
    subscribe: withFilter(
      () => pubsub.asyncIterator(SubsTypes.NewComment), (payload, variables = {}, context) => {
        // TODO: can't add context in playground and test subscription without frontend
        // disable auth for dev env?
        if (isNilOrEmpty(context.user)) {
          return false
        }
        const globalReportId = variables.input && variables.input.reportId
        const reportId = Number(fromGlobalId(globalReportId).id)
        return payload.report_id === reportId
      }
    ),
  }
}

END_POINT,
	FE_URL,
	GRAPHQL_DEPTH_LIMIT,
	ACCESS_TOKEN
} from '@environments'

// const gateway = new ApolloGateway({
// 	serviceList: [
// 		{ name: 'accounts', url: 'http://localhost:11041/graphql' },
// 		{ name: 'reviews', url: 'http://localhost:11042/graphql' },
// 		{ name: 'products', url: 'http://localhost:11043/graphql' },
// 		{ name: 'inventory', url: 'http://localhost:11044/graphql' }
// 	]
// })

const pubsub = new PubSub()
class MyErrorTrackingExtension extends GraphQLExtension {
	willSendResponse(o) {
		const { context, graphqlResponse } = o

		context.trackErrors(graphqlResponse.errors)

		return o
	}
	// Other lifecycle methods include
	// requestDidStart
	// parsingDidStart
	// validationDidStart
	// executionDidStart
	// willSendResponse
}

//@flow

/**
 * Create the subscription manager to be used by the subscription server
 */
const { SubscriptionManager } = require('graphql-subscriptions');
const schema = require('../schema');
const pubsub = require('./listeners/pubsub');
const channels = require('./listeners/channels');

module.exports = new SubscriptionManager({
  schema,
  pubsub,
  // setupFunctions map a Subscription type as defined in the
  // schema (e.g. messageAdded) to certain channels (e.g. MESSAGE_ADDED)
  // Subscriptions can also listen to multiple channels
  setupFunctions: {
    messageAdded: (_, { thread }) => ({
      [channels.MESSAGE_ADDED]: {
        filter: message => message.threadId === thread,
      },
    }),
    notificationAdded: (_, __, { user }) => ({
      [channels.NOTIFICATION_ADDED]: {
        filter: notification => notification,
      },
    }),

it('should create a question when authenticated', async () => {
  const user = await createRows.createUser()
  const pubSub = new PubSub()
  const context = getContext({ user, pubSub })
  const variables = {
    content: 'What does the fox say?',
  }

  const result = await graphql(schema, query, rootValue, context, variables)
  expect(result).toMatchSnapshot()
})

import { PubSub, SubscriptionManager } from 'graphql-subscriptions';
import schema from '../schema';

const pubsub = new PubSub();
const subscriptionManager = new SubscriptionManager({
  schema,
  pubsub,
  setupFunctions: {
  },
});

export { subscriptionManager, pubsub };

links: () => db.links
  },
  Mutation: {
    createLink: (_, { url, description }) => {
      const link = { id: shortid.generate(), description, url };
      db.links.push(link);
      pubsub.publish(CHANNEL.LINK, { [SUBSCRIPTION.LINK]: link });
      return link;
    }
  },
  Subscription: {
    [SUBSCRIPTION.LINK]: {
      resolve: (payload, args, context, info) => {
        return payload.link;
      },
      subscribe: withFilter(
        (rootValue, args, context, info) => {
          return pubsub.asyncIterator(CHANNEL.LINK);
        },
        (payload, variables, context, info) => {
          console.log('payload: ', payload);
          console.log('variables: ', variables);
          return true;
        }
      )
    }
  }
};

const schema = makeExecutableSchema({
  typeDefs,
  resolvers