GraphQL Subscriptions: Real-Time Data Made Simple

const { ApolloServer, gql, PubSub } = require('apollo-server');

const pubsub = new PubSub();
const MESSAGE_ADDED = 'MESSAGE_ADDED';

// Schema
const typeDefs = gql`
  type Subscription {
    messageAdded(chatId: ID!): Message!
  }

  type Message {
    id: ID!
    text: String!
    chatId: ID!
    author: User!
    createdAt: String!
  }

  type Mutation {
    sendMessage(chatId: ID!, text: String!): Message!
  }
`;

// Resolvers
const resolvers = {
  Mutation: {
    sendMessage: (parent, { chatId, text }, context) => {
      const message = {
        id: String(Date.now()),
        text,
        chatId,
        author: context.user,
        createdAt: new Date().toISOString()
      };

      // Publish the message to subscribers
      pubsub.publish(MESSAGE_ADDED, { 
        messageAdded: message 
      });

      return message;
    }
  },

  Subscription: {
    messageAdded: {
      subscribe: (parent, { chatId }) => {
        // Return an async iterator that filters by chatId
        return pubsub.asyncIterator([MESSAGE_ADDED]);
      },
      resolve: (payload, { chatId }) => {
        // Only send if message is for this chat
        if (payload.messageAdded.chatId === chatId) {
          return payload.messageAdded;
        }
        return null;
      }
    }
  }
};

const server = new ApolloServer({
  typeDefs,
  resolvers,
  subscriptions: {
    path: '/subscriptions',
  }
});

server.listen().then(({ url, subscriptionsUrl }) => {
  console.log(`Server ready at ${url}`);
  console.log(`Subscriptions ready at ${subscriptionsUrl}`);
});

import { ApolloClient, InMemoryCache, split, HttpLink } from '@apollo/client';
import { GraphQLWsLink } from '@apollo/client/link/subscriptions';
import { getMainDefinition } from '@apollo/client/utilities';
import { createClient } from 'graphql-ws';

// HTTP connection for queries and mutations
const httpLink = new HttpLink({
  uri: 'http://localhost:4000/graphql'
});

// WebSocket connection for subscriptions
const wsLink = new GraphQLWsLink(createClient({
  url: 'ws://localhost:4000/subscriptions'
}));

// Split traffic based on operation type
const splitLink = split(
  ({ query }) => {
    const definition = getMainDefinition(query);
    return (
      definition.kind === 'OperationDefinition' &&
      definition.operation === 'subscription'
    );
  },
  wsLink,      // Use WebSocket for subscriptions
  httpLink,    // Use HTTP for queries and mutations
);

const client = new ApolloClient({
  link: splitLink,
  cache: new InMemoryCache()
});

import { useSubscription, gql } from '@apollo/client';

const MESSAGE_SUBSCRIPTION = gql`
  subscription OnMessageAdded($chatId: ID!) {
    messageAdded(chatId: $chatId) {
      id
      text
      author {
        name
        avatar
      }
      createdAt
    }
  }
`;

function ChatRoom({ chatId }) {
  const [messages, setMessages] = useState([]);

  const { data, loading, error } = useSubscription(
    MESSAGE_SUBSCRIPTION,
    { variables: { chatId } }
  );

  useEffect(() => {
    if (data) {
      // Add new message to the list
      setMessages(prev => [...prev, data.messageAdded]);
    }
  }, [data]);

  if (loading) return <p>Connecting...</p>;
  if (error) return <p>Error: {error.message}</p>;

  return (
    <div>
      {messages.map(message => (
        <div key={message.id}>
          <strong>{message.author.name}:</strong> {message.text}
        </div>
      ))}
    </div>
  );
}

const { RedisPubSub } = require('graphql-redis-subscriptions');
const Redis = require('ioredis');

const options = {
  host: process.env.REDIS_HOST,
  port: process.env.REDIS_PORT,
  retryStrategy: times => Math.min(times * 50, 2000)
};

const pubsub = new RedisPubSub({
  publisher: new Redis(options),
  subscriber: new Redis(options)
});

// Now all server instances share the same PubSub