graphql-ws-subscription

client.js

import { Client } from 'subscriptions-transport-ws';

const client = new Client(`ws://localhost:4001/`);

const query = `subscription events {
  eventsUpdated{
    Name
  }
}`;

client.subscribe({query: query}, onMessage);

function onMessage(error, result) {
  console.log(error, result);
}

schema.js

import {
  GraphQLObjectType,
  GraphQLSchema,
  GraphQLList,
  GraphQLInt,
  GraphQLString
} from 'graphql';

const eventType = new GraphQLObjectType({
  name: 'Event',
  fields: {
    Id: {
      type: GraphQLInt
    },
    Name: {
      type: GraphQLString
    }
  }
});

const schema = new GraphQLSchema({
  query: new GraphQLObjectType({
    name: 'Query',
    fields: {
      events: { type: new GraphQLList(eventType) }
    },
  }),
  subscription: new GraphQLObjectType({
    name: 'Subscription',
    fields: {
      eventsAdded: {
        type: new GraphQLList(eventType),
        args: {
          name: { type: GraphQLString },
        },
        resolve(events, {name}) {
          return !name ? events : events.filter(event => event.Name.toLowerCase().indexOf(name.toLowerCase()) > -1);
        }
      },
      eventsUpdated: {
        type: new GraphQLList(eventType),
        args: {
          name: { type: GraphQLString },
        },
        resolve(events, {name}) {
          console.log('get');
          return !name ? events : events.filter(event => event.Name.toLowerCase().indexOf(name.toLowerCase()) > -1);
        }
      },
      eventsRemoved: {
        type: new GraphQLList(GraphQLString),
        args: {
          id: { type: GraphQLString },
        },
        resolve(events, {id}) {
          return events.filter(event => event.Id === id);
        }
      }
    }
  })
});

export default schema;

server.js

import { createServer } from 'http';
import { SubscriptionServer } from 'subscriptions-transport-ws';

import InnerFeed from '../inner-feed/inner-feed-wrapper/src/InnerFeed';

import { pubsub, subscriptionManager } from './subscriptions';

const WS_PORT = '4001';

const httpServer = createServer((request, response) => {
  response.writeHead(404);
  response.end();
});

const subscriptionsServer = new SubscriptionServer({ subscriptionManager }, httpServer);

const feed = new InnerFeed({
  config: require('./innerFeedConfig.json'),
  logger: console
});

feed
  .connect()
  .then(() => {
    feed.on(({object: {Added, Updated, Removed}}) => {
      if (Added && Added.length) {
        console.log('eventsAdded', Added.length);
        pubsub.publish('eventsAdded', Added);
      }
      if (Updated && Updated.length) {
        console.log('eventsUpdated', Updated.length);
        pubsub.publish('eventsUpdated', Updated);
      }
      if (Removed && Removed.length) {
        console.log('eventsRemoved', Removed);
        pubsub.publish('eventsRemoved', Removed);
      }
    });
  })
  .catch( console.error );

httpServer.listen(WS_PORT, () => {
  return console.log(`Websocket Server is now running on http://localhost:${WS_PORT}`);
});

subscriptions.js

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

import schema from './schema';

// the default PubSub is based on EventEmitters. It can easily
// be replaced with one different one, e.g. Redis
const pubsub = new PubSub();

const subscriptionManager = new SubscriptionManager({
  schema,
  pubsub,
  setupFunctions: {
    eventsAdded: (options, {name}) => ({
      eventsAdded: events => events.some(e => e.Name.toLowerCase().indexOf(name.toLowerCase()) > -1)
    }),
    eventsUpdated: (options, {name}) => ({
      eventsUpdated: events => {
        return events.some(e => e.Name.toLowerCase().indexOf(name.toLowerCase()) > -1);
      }
    }),
    eventsRemoved: (options, {id}) => ({
      eventsRemoved: events => events.some(e => e.Id === id)
    }),
  },
});

export { pubsub, subscriptionManager };