apollo-server: Datasources missing in the context when using subscriptions

Unfortunately, this is a known issue. Although we run SubscriptionServer as part of ApolloServer, the request pipeline is completely separate. That means features like persisted queries, data sources, and tracing don’t currently work with subscriptions (or queries and mutations over the WebSocket transport).

There is work underway on a refactoring of the apollo-server-core request pipeline that will make it transport independent, and that will allow us to build the WebSocket transport on top of it. Until then, I’m afraid there isn’t much we can do.

I solved this by using @PatrickStrz method with some little tweak to have each Datasource instance to have the full context, Same way apollo server initializes its Datasources

const pubSub = new PubSub();

/**
 * Generate dataSources for both transport protocol
 */
function dataSources() {
  return {
    userService: new UserService(),
    ministryService: new MinistryService(),
    mailService: new MailService(),
  };
}

/**
 * Authenticate  subscribers on initial connect
 * @param {*} connectionParams 
 */
async function onWebsocketConnect(connectionParams) {
  const authUser = await authProvider.getAuthenticatedUser({
    connectionParams,
  });
  if (authUser) {
    return { authUser, pubSub, dataSources: dataSources() };
  }
  throw new Error("Invalid Credentials");
}

/**
 * Initialize subscribtion datasources
 * @param {Context} context 
 */
function initializeSubscriptionDataSources(context) {
  const dataSources = context.dataSources;
  if (dataSources) {
    for (const instance in dataSources) {
      dataSources[instance].initialize({ context, cache: undefined });
    }
  }
}

/**
 * Constructs the context for transport (http and ws-subscription) protocols
 */
async function context({ req, connection }) {
  if (connection) {
    const subscriptionContext = connection.context;
    initializeSubscriptionDataSources(subscriptionContext);
    return subscriptionContext;
  }

  const authUser = await authProvider.getAuthenticatedUser({ req });
  return { authUser, pubSub };
}

/**
 * Merges other files schema and resolvers to a whole
 */
const schema = makeExecutableSchema({
  typeDefs: [rootTypeDefs, userTypeDefs, ministryTypeDefs, mailTypeDefs],
  resolvers: [rootResolvers, userResolvers, ministryResolvers, mailResolvers],
});

/**
 * GraphQL server config
 */
const graphQLServer = new ApolloServer({
  schema,
  context,
  dataSources,
  subscriptions: {
    onConnect: onWebsocketConnect,
  },
});

Hope this helps 😃

@jbaxleyiii so why is this ticket closed? Is 3.0 out?

Just worth noting that the subscriptions integration in Apollo Server has always been incredibly superficial, and we are planning to remove it in Apollo Server 3 (replacing it with instructions on how to use a more maintained subscriptions package alongside Apollo Server). We do hope to have a more deeply integrated subscriptions implementation at some point but the current version promises more than it delivers, as this and many other issues indicate.

I solved this by using @BrockReece’s method but initializing DataSource classes manually:

const constructDataSourcesForSubscriptions = (context) => {
  const initializeDataSource = (dataSourceClass) => {
    const instance = new dataSourceClass()
    instance.initialize({ context, cache: undefined })
    return instance 
  }

  const LanguagesAPI = initializeDataSource(LanguagesAPI)

  return {
    LanguagesAPI,
  }
}

const server = new ApolloServer({
  ...
  dataSources: () => {
    return: {
      LanguagesAPI: new LanguagesAPI(),
    }
  },
  context: ({req, connection}) => {
    if (connection) {
      return {
        dataSources: constructDataSourcesForSubscriptions(connection.context)
    }
    return {
      // req context stuff
    }
  }
}

Hope this helps 😃

Any update on this?

any update on this?

It’s 2020 now… Any update on this?

Hi, did anyone come up with a solution to this?

I am currently adding my dataSources to both the dataSource and context methods, but this feels kind of icky…

const LanguagesAPI = require('./dataSources/LanguagesAPI')

const server = new ApolloServer({
  ...
  dataSources: () => {
    return: {
      LanguagesAPI: new LanguagesAPI(),
    }
  },
  context: ({req, connection}) => {
    if (connection) {
      return {
        dataSources: {
          LanguagesAPI: new LanguagesAPI(),
       }
    }
    return {
      // req context stuff
    }
  }
}

any update on this?

Any update? Can’t use data sources as newing one up leaves it uninitialised…

For future visitors, here’s an example implementation using Apollo Server 3 (apollo-server-express 3.9), WebSocketServer from ‘ws’, and useServer from ‘graphql-ws’, including authorization.

import { InMemoryLRUCache } from 'apollo-server-caching';
import { ApolloServer } from 'apollo-server-express';
import { ApolloServerPluginDrainHttpServer } from 'apollo-server-core';
import express from 'express';
import http from 'http';
import { WebSocketServer } from 'ws';
import { useServer } from 'graphql-ws/lib/use/ws';
import { getUser, connectToDatabase } from './utils';
import schema from './setup';

const makeContext = async ({ req, database }) => {
    return {
      database, // used to set up dataSources, required to be in context
      user: getUser(req.headers, database), // checks for Authorization on req.headers and fetches user from database
    };
};

const makeWScontext = async ({ connectionParams, database, dataSources: dataSourceFn }) => {
  const cache = new InMemoryLRUCache();
  const dataSources = dataSourceFn(); // pass in as function, then call to receive objects
  for (const dataSource in dataSources)
    dataSources[dataSource].initialize({ context, cache });
  const user = getUser(context.connectionParams); // this uses the connectionParams sent from frontend to look for Authorization Bearer token
    return {
      database,
      user,
      dataSources, // unlike query context, we need to manually add dataSources to WebSocket context
    };
};

const makeServer = async () => {
  const database = await connectToDatabase(); // for example mongo instance
  const dataSources = () => ({
    posts: new Posts(...), // ... could refer to database.collection('posts') if using mongo
    users: new Users(...),
  });

  const app = express();
  const httpServer = http.createServer(app);

  const wsPath = '/graphql';
  const wsServer = new WebSocketServer({
    server: httpServer,
    path: wsPath,
  });

  const serverCleanup = useServer(
    {
      schema,
      context: (context) => {
        return makeWScontext({
          context,
          database,
          dataSources, // pass dataSources function to custom WebSocket context function, which must call dataSource.initialize({ context, cache }) manually. supply cache here if you want it to be shared across sockets
        });
      },
    },
    wsServer
  );

  const server = new ApolloServer({
    csrfPrevention: true,
    schema,
    dataSources, // Pass dataSources function to new ApolloServer, and it will call dataSource.initialize({ context, cache }) on all, and add to context automatically
    context: async ({ req }) =>
      await makeContext({
        database,
        req,
      }),
    plugins: [
      ApolloServerPluginDrainHttpServer({ httpServer }),
      {
        async serverWillStart() {
          return {
            async drainServer() {
              await serverCleanup.dispose();
            },
          };
        },
      },
    ],
  });

  await server.start();

  server.applyMiddleware({
    app,
    cors: {
      origin: '*', // <- allow request from all domains
      methods: 'POST,GET,OPTIONS',
      allowedHeaders: 'Content-Type, Authorization',
    },
  });

  httpServer.listen({ port: PORT }, () => {
    const hostUrl = process.env.API_HOST || 'http://localhost:4000';
    const wsUrl = hostUrl.replace('http', 'ws');
    console.log(
      `🚀 Queries and mutations ready at ${hostUrl}${server.graphqlPath}`
    );
    console.log(`🚀 Subscriptions ready at ${wsUrl}${wsPath}`);
  });
  
  makeServer()

Just want to note/summarize that

• @BrockReece’s example wouldn’t ever call initialize on the DataSource instances, which might be problematic for some classes of DataSource.
• @PatrickStrz’s and @josedache’s examples do call initialize, but with an undefined cache property, which is problematic (more following)
• @MikaStark’s example doesn’t contain the code for their createDataSources function, so unsure whether it calls initialize methods with a cache.

My impression is that the cache property on DataSourceConfig is the way multiple DataSource instances are intended to share state. For example, the one Apollo-supported DataSource, RESTDataSource, uses the cache as the backing store for its HTTP request/response cache (it namespaces its entries with the prefix httpcache:). When initialized with cache: undefined, RESTDataSource instances will all use instance-specific InMemoryLRUCaches. Since you typically instantiate new instances for every GraphQL operation, that means multiple operations don’t share a cache. That might not be what you want.

I suspect most people will instead want to explicitly pass a cache to the Apollo Server config (which the framework then initializes query + mutation datasources with), and also pass that same object to the manual initialize DataSource calls done for subscriptions. Then every operation the server handles, including subscriptions, uses the same cache.

For reference, the default is (src) a new InMemoryLRUCache() from apollo-server-caching.

@josedache Awesome, works like a charm! Thank you!

We can follow the 3.0 Roadmap, which includes “Unify diverging request pipelines”:

https://github.com/apollographql/apollo-server/issues/2360