How to use the apollo-server.PubSub function in apollo-server

import { ApolloServer, PubSub } from 'apollo-server';
import * as mongoose from 'mongoose';
import * as dotenv from 'dotenv';
import graphqlTypes from './graphqlTypes';
import resolvers from './resolvers';
import { getUser } from './auth/authMethods';
import { MongoError } from 'mongodb';

dotenv.config();

const pubsub = new PubSub();

const server = new ApolloServer({
  resolvers,
  typeDefs: graphqlTypes,
  // @ts-ignore
  context: async ({ req, connection }) => {
    if (connection) {
      return {
        ...connection.context,
        pubsub,
      };
    } else {
      // check from req
      const token = req.headers.authorization ? req.headers.authorization : '';
      const user = await getUser(token);
      return {

this.version = options.version || null;

    this.context = {
      app: this,
      appEvents,
      appVersion: this.version,
      auth: {},
      collections: this.collections,
      getFunctionsOfType: (type) => (this.functionsByType[type] || []).map(({ func }) => func),
      mutations: {},
      queries: {},
      // In a large production app, you may want to use an external pub-sub system.
      // See https://www.apollographql.com/docs/apollo-server/features/subscriptions.html#PubSub-Implementations
      // We may eventually bind this directly to Kafka.
      pubSub: new PubSub()
    };

    this.functionsByType = {};
    this.graphQL = {
      resolvers: {},
      schemas: [coreGraphQLSchema]
    };

    _.merge(this.graphQL.resolvers, coreResolvers);

    // Passing in `rootUrl` option is mostly for tests. Recommend using ROOT_URL env variable.
    const resolvedRootUrl = options.rootUrl || ROOT_URL;

    this.rootUrl = resolvedRootUrl.endsWith("/") ? resolvedRootUrl : `${resolvedRootUrl}/`;
    this.context.rootUrl = this.rootUrl;
    this.context.getAbsoluteUrl = (path) => getAbsoluteUrl(this.context.rootUrl, path);

constructor(options = {}) {
    this.options = { ...options };
    this.collections = {
      ...(options.additionalCollections || {})
    };
    this.context = {
      ...(options.context || {}),
      app: this,
      appEvents,
      collections: this.collections,
      getFunctionsOfType: (type) => (this.functionsByType[type] || []).map(({ func }) => func),
      // In a large production app, you may want to use an external pub-sub system.
      // See https://www.apollographql.com/docs/apollo-server/features/subscriptions.html#PubSub-Implementations
      // We may eventually bind this directly to Kafka.
      pubSub: new PubSub()
    };

    this.functionsByType = {};
    this.graphQL = {
      resolvers: {},
      schemas: []
    };

    this._registerFunctionsByType(options.functionsByType, "__APP__");

    if (options.graphQL) {
      if (options.graphQL.resolvers) {
        merge(this.graphQL.resolvers, options.graphQL.resolvers);
      }
      if (options.graphQL.schemas) {
        this.graphQL.schemas.push(...options.graphQL.schemas);

export const main = async () => {
  FanoutGraphqlExpressServer({
    grip: {
      url: process.env.GRIP_URL || "http://localhost:5561",
    },
    pubsub: new PubSub(),
    tables: {
      connections: MapSimpleTable(),
      notes: MapSimpleTable(),
      pubSubSubscriptions: MapSimpleTable(),
    },
  })
    .listen(process.env.PORT || 57410)
    .then(({ url, subscriptionsUrl }) => {
      console.log(`🚀 Server ready at ${url}`);
      console.log(`🚀 Subscriptions ready at ${subscriptionsUrl}`);
    });
};

const chalk = require("chalk");
const fs = require("fs");

const { visit } = require("graphql/language/visitor");
const { parse } = require("graphql/language");

require("dotenv").config();

const options = {
    quiet: false
};
process.argv.forEach(option => {
    if (option === "-q" || option === "--quiet") options.quiet = true;
});

const pubsub = new PubSub();

const CF_SCHEMA = yaml.Schema.create([
    new yaml.Type("!Ref", {
        kind: "scalar",
        construct: function(data) {
            return {
                Ref: data
            };
        }
    }),
    new yaml.Type("!Equals", {
        kind: "sequence",
        construct: function(data) {
            return {
                "Fn::Equals": data
            };

const { withFilter } = require('graphql-subscriptions');
let {PubSub} = require("apollo-server");

const MESSAGESENT = "MESSAGESENT";
const CONVERSATIONCREATED = "CONVERSATIONCREATED";
const USER_CREATED = "USER_CREATED";

const pubsub = new PubSub();

module.exports = {
  Subscription: {
    messageSent: {
      subscribe: withFilter(() => pubsub.asyncIterator(MESSAGESENT), (payload, variables) => {
        return payload.conversation === variables.conversation;
     }),
    },
    conversationCreated: {
      subscribe: () => pubsub.asyncIterator([CONVERSATIONCREATED]),
    },
    userCreated: {
      subscribe: () => pubsub.asyncIterator([USER_CREATED]),
    },
  },
  Query: {

function PrePublishedPubSub(publishes: IPubSubEnginePublish[]): PubSubEngine {
  const asyncIterator = (triggerName: string | string[]) => {
    const triggerNames = Array.isArray(triggerName)
      ? triggerName
      : [triggerName];
    const ai = createAsyncIterator(
      publishes
        .filter(p => triggerNames.includes(p.triggerName))
        .map(p => p.payload),
    );
    return ai;
  };
  return Object.assign(new PubSub(), {
    asyncIterator,
  });
}

beforeEach(async () => {
    await checkPortIsFree(8082)
    server = Server({
      apiPort: API_PORT,
      contractDirectory: 'test/bundles/nodejs',
      contractRepository: InMemoryRepository(),
      engineClients: new Map([[
        Engine.stub,
        StubEngineClient()
      ]]),
      pubSubClient: new PubSub()
    })
    await client.connect('abc')
  })

beforeEach(async () => {
    await checkPortIsFree(8082)
    const pubSubClient = new PubSub()
    const contractRepository = await populatedContractRepository()
    const contractController = ContractController({
      contractDirectory: 'test/bundles/nodejs',
      contractRepository,
      engineClients: new Map([[
        Engine.stub,
        StubEngineClient()
      ]]),
      pubSubClient
    })
    api = Api({
      contractController,
      contractRepository,
      pubSubClient
    })