Building Chatty — Part 6: GraphQL Subscriptions
A WhatsApp clone with React Native and Apollo
This is the sixth blog in a multipart series where we will be building Chatty, a WhatsApp clone, using React Native and Apollo. You can view the code for this part of the series here.
In this tutorial, we’ll focus on adding GraphQL Subscriptions, which will give our app real-time instant messaging capabilities!
Here’s what we will accomplish in this tutorial:
- Introduce Event-based Subscriptions
- Build server-side infrastructure to handle GraphQL Subscriptions via WebSockets
- Design GraphQL Subscriptions and add them to our GraphQL Schemas and Resolvers
- Build client-side infrastructure to handle GraphQL Subscriptions via WebSockets
- Subscribe to GraphQL Subscriptions on our React Native client and handle real-time updates
Event-based Subscriptions
Real-time capable apps need a way to be pushed data from the server. In some real-time architectures, all data is considered live data, and anytime data changes on the server, it’s pushed through a WebSocket or long-polling and updated on the client. While this sort of architecture means we can expect data to update on the client without writing extra code, it starts to get tricky and non-performant as apps scale. For one thing, you don’t need to keep track of every last bit of data if it’s not relevant to the user. Moreover, it’s not obvious what changes to data should trigger an event, what that event should look like, and how our clients should react.
With an event-based subscription model in GraphQL — much like with queries and mutations — a client can tell the server exactly what data it wants to be pushed and what that data should look like. This leads to fewer events tracked on the server and pushed to the client, and precise event handling on both ends!
GraphQL Subscriptions on the Server
It’s probably easiest to think about our event based subscriptions setup from the client’s perspective. All queries and mutations will still get executed with standard HTTP requests. This will keep request execution more reliable and the WebSocket connection unclogged. We will only use WebSockets for subscriptions, and the client will only subscribe to events it cares about — the ones that affect stuff for the current user.
Designing GraphQL Subscriptions
Let’s focus on the most important event that ever happens within a messaging app — getting a new message.
When someone creates a new message, we want all group members to be notified that a new message was created. We don’t want our users to know about every new message being created by everybody on Chatty, so we’ll create a system where users subscribe to new message events just for their own groups. We can build this subscription model right into our GraphQL Schema!
Let’s modify our GraphQL Schema in server/data/schema.js to include a GraphQL Subscription for when new messages are added to a group we care about:
Step 6.1: Add Subscription to Schema
Changed server/data/schema.js
... leaveGroup(id: Int!, userId: Int!): Group # let user leave group
updateGroup(id: Int!, name: String): Group
} type Subscription {
# Subscription fires on every message added
# for any of the groups with one of these groupIds
messageAdded(userId: Int, groupIds: [Int]): Message
}
schema {
query: Query
mutation: Mutation
subscription: Subscription
}
`;
That’s it!
GraphQL Subscription Infrastructure
Our Schema uses GraphQL Subscriptions, but our server has no way to handle them yet.
Fortunately, apollo-server comes prepacked with excellent tools to handle subscriptions right out the gate!
First, we’ll use apollo-server to create a PubSub manager. PubSub is basically just event emitters wrapped with a function that filters messages. It can easily be replaced later with something more scalable like graphql-redis-subscriptions.
Let’s create a new file server/subscriptions.js where we’ll keep any subscription infrastructure:
Step 6.2: Create subscriptions.js
Added server/subscriptions.js
...import { PubSub } from 'apollo-server';export const pubsub = new PubSub();export default pubsub;
Now that we’ve created a PubSub, we can use this class to publish and subscribe to events as they occur in our Resolvers.
We can modify server/data/resolvers.js as follows:
Step 6.3: Add Subscription to Resolvers
Changed server/data/resolvers.js
...import GraphQLDate from 'graphql-date';
import { Group, Message, User } from './connectors';
import { pubsub } from '../subscriptions';const MESSAGE_ADDED_TOPIC = 'messageAdded';export const resolvers = {
Date: GraphQLDate,... userId,
text,
groupId,
}).then((message) => {
// publish subscription notification with the whole message
pubsub.publish(MESSAGE_ADDED_TOPIC, { [MESSAGE_ADDED_TOPIC]: message });
return message;
});
},
createGroup(_, { name, userIds, userId }) {... .then(group => group.update({ name }));
},
},
Subscription: {
messageAdded: {
// the subscription payload is the message.
subscribe: () => pubsub.asyncIterator(MESSAGE_ADDED_TOPIC),
},
},
Group: {
users(group) {
return group.getUsers();
Whenever a user creates a message, we trigger pubsub to publish the messageAdded event along with the newly created message. pubsubwill emit an event to any clients subscribed to messageAdded and pass them the new message.
But we only want to emit this event to clients who care about the message because it was sent to one of their user’s groups! We can modify our implementation to filter who gets the event emission:
Step 6.4: Add withFilter to messageAdded
Changed server/data/resolvers.js
...import GraphQLDate from 'graphql-date';
import { withFilter } from 'apollo-server';import { Group, Message, User } from './connectors';
import { pubsub } from '../subscriptions';... },
Subscription: {
messageAdded: {
subscribe: withFilter(
() => pubsub.asyncIterator(MESSAGE_ADDED_TOPIC),
(payload, args) => {
return Boolean(
args.groupIds &&
~args.groupIds.indexOf(payload.messageAdded.groupId) &&
args.userId !== payload.messageAdded.userId, // don't send to user creating message
);
},
),
},
},
Group: {
Using withFilter, we create a filter which returns true when the groupId of a new message matches one of the groupIds passed into our messageAdded subscription. This filter will be applied whenever pubsub.publish(MESSAGE_ADDED_TOPIC, { [MESSAGE_ADDED_TOPIC]: message }) is triggered, and only clients whose subscriptions pass the filter will receive the message.
Our Resolvers are all set up.
Putting it all together
Our apollo-server will automatically attempt to serve subscriptions via WebSockets as soon as we define subscriptions in our Schema -- how cool is that?! Whenever the pubsub iterators in our subscription resolvers receive events, it will filter which subscribed clients will receive the data, and then the server will push GraphQL responses over WebSockets to those clients. There's a whole lot of magic happening here. Let's verify it's all working:
A GraphQL Subscription is written on the client much like a query or mutation. For example, in GraphQL Playground, we could write the following GraphQL Subscription for messageAdded:
subscription messageAdded($groupIds: [Int]){
messageAdded(groupIds: $groupIds) {
id
to {
name
}
from {
username
}
text
}
}Let’s check out GraphQL Playground and see if everything works:
New Subscription Workflow
We’ve successfully set up GraphQL Subscriptions on our server.
Since we have the infrastructure in place, let’s add one more subscription for some extra practice. We can use the same methodology we used for subscribing to new Messages and apply it to new Groups. After all, it’s important that our users know right away that they’ve been added to a new group.
The steps are as follows:
- Add the subscription to our Schema:
Step 6.5: Add groupAdded to Schema
Changed server/data/schema.js
... # Subscription fires on every message added
# for any of the groups with one of these groupIds
messageAdded(userId: Int, groupIds: [Int]): Message
groupAdded(userId: Int): Group
}
schema {
- Publish to the subscription when a new
Groupis created and resolve the subscription in the Resolvers:
Step 6.6: Add groupAdded to Resolvers
Changed server/data/resolvers.js
...import { pubsub } from '../subscriptions';const MESSAGE_ADDED_TOPIC = 'messageAdded';
const GROUP_ADDED_TOPIC = 'groupAdded';export const resolvers = {
Date: GraphQLDate,... users: [user, ...friends],
})
.then(group => group.addUsers([user, ...friends])
.then((res) => {
// append the user list to the group object
// to pass to pubsub so we can check members
group.users = [user, ...friends];
pubsub.publish(GROUP_ADDED_TOPIC, { [GROUP_ADDED_TOPIC]: group });
return group;
})),
),
);
},... },
),
},
groupAdded: {
subscribe: () => pubsub.asyncIterator(GROUP_ADDED_TOPIC),
},
},
Group: {
users(group) {
- Filter the recipients of the emitted new group with
withFilter:
Step 6.7: Add withFilter to groupAdded
Changed server/data/resolvers.js
...import GraphQLDate from 'graphql-date';
import { withFilter } from 'apollo-server';
import { map } from 'lodash';import { Group, Message, User } from './connectors';
import { pubsub } from '../subscriptions';... ),
},
groupAdded: {
subscribe: withFilter(
() => pubsub.asyncIterator(GROUP_ADDED_TOPIC),
(payload, args) => {
return Boolean(
args.userId &&
~map(payload.groupAdded.users, 'id').indexOf(args.userId) &&
args.userId !== payload.groupAdded.users[0].id, // don't send to user creating group
);
},
),
},
},
Group: {
All set!
GraphQL Subscriptions on the Client
Time to add subscriptions inside our React Native client. We’ll start by adding a few packages to our client:
# make sure you're adding the package in the client!!!
cd client
npm i apollo-link-ws apollo-utilities subscriptions-transport-wsWe’ll use subscription-transport-ws on the client to create a WebSocket client connected to our WebSocket endpoint. apollo-serverserves this endpoint at /graphql by default.
We then add this WebSocket client into our Apollo workflow via apollo-link-ws. We need to split the various operations that flow through Apollo into subscription operations and non-subscription (query and mutation) and handle them with the appropriate links:
Step 6.8: Add wsClient to networkInterface
Changed client/package.json
... "apollo-link-error": "^1.0.7",
"apollo-link-http": "^1.3.3",
"apollo-link-redux": "^0.2.1",
"apollo-link-ws": "^1.0.5",
"apollo-utilities": "^1.0.6",
"buffer": "^5.0.8",
"graphql": "^0.12.3",
"graphql-tag": "^2.4.2",... "react-navigation-redux-helpers": "^1.1.2",
"react-redux": "^5.0.5",
"redux": "^3.7.2",
"redux-devtools-extension": "^2.13.2",
"subscriptions-transport-ws": "^0.9.5"
},
"devDependencies": {
"babel-jest": "23.4.0",
Changed client/src/app.js
...import { ReduxCache, apolloReducer } from 'apollo-cache-redux';
import ReduxLink from 'apollo-link-redux';
import { onError } from 'apollo-link-error';
import { WebSocketLink } from 'apollo-link-ws';
import { getMainDefinition } from 'apollo-utilities';
import { SubscriptionClient } from 'subscriptions-transport-ws';import AppWithNavigationState, {
navigationReducer,...
const httpLink = createHttpLink({ uri: `http://${URL}` });// Create WebSocket client
export const wsClient = new SubscriptionClient(`ws://${URL}/graphql`, {
reconnect: true,
connectionParams: {
// Pass any arguments you want for initialization
},
});const webSocketLink = new WebSocketLink(wsClient);const requestLink = ({ queryOrMutationLink, subscriptionLink }) =>
ApolloLink.split(
({ query }) => {
const { kind, operation } = getMainDefinition(query);
return kind === 'OperationDefinition' && operation === 'subscription';
},
subscriptionLink,
queryOrMutationLink,
);const link = ApolloLink.from([
reduxLink,
errorLink,
requestLink({
queryOrMutationLink: httpLink,
subscriptionLink: webSocketLink,
}),
]);export const client = new ApolloClient({
That’s it — we’re ready to start adding subscriptions!
Designing GraphQL Subscriptions
Our GraphQL Subscriptions are going to be ridiculously easy to write now that we’ve had practice with queries and mutations. We’ll first write our messageAdded subscription in a new file client/src/graphql/message-added.subscription.js:
Step 6.9: Create MESSAGE_ADDED_SUBSCRIPTION
Added client/src/graphql/message-added.subscription.js
...import gql from 'graphql-tag';import MESSAGE_FRAGMENT from './message.fragment';const MESSAGE_ADDED_SUBSCRIPTION = gql`
subscription onMessageAdded($userId: Int, $groupIds: [Int]){
messageAdded(userId: $userId, groupIds: $groupIds){
... MessageFragment
}
}
${MESSAGE_FRAGMENT}
`;export default MESSAGE_ADDED_SUBSCRIPTION;
I’ve retitled the subscription onMessageAdded to distinguish the name from the subscription itself.
The groupAdded component will look extremely similar:
Step 6.10: Create GROUP_ADDED_SUBSCRIPTION
Added client/src/graphql/group-added.subscription.js
...import gql from 'graphql-tag';import MESSAGE_FRAGMENT from './message.fragment';const GROUP_ADDED_SUBSCRIPTION = gql`
subscription onGroupAdded($userId: Int){
groupAdded(userId: $userId){
id
name
messages(first: 1) {
edges {
cursor
node {
... MessageFragment
}
}
}
}
}
${MESSAGE_FRAGMENT}
`;export default GROUP_ADDED_SUBSCRIPTION;
Our subscriptions are fired up and ready to go. We just need to add them to our UI/UX and we’re finished.
Connecting Subscriptions to Components
Our final step is to connect our new subscriptions to our React Native components.
Let’s first apply messageAdded to the Messages component. When a user is looking at messages within a group thread, we want new messages to pop onto the thread as they’re created.
The graphql module in react-apollo exposes a prop function named subscribeToMore that can attach subscriptions to a component. Inside the subscribeToMore function, we pass the subscription, variables, and tell the component how to modify query data state with updateQuery.
Take a look at the updated code in our Messages component in client/src/screens/messages.screen.js:
Step 6.11: Apply subscribeToMore to Messages
Changed client/src/screens/messages.screen.js
...import GROUP_QUERY from '../graphql/group.query';
import CREATE_MESSAGE_MUTATION from '../graphql/create-message.mutation';
import USER_QUERY from '../graphql/user.query';
import MESSAGE_ADDED_SUBSCRIPTION from '../graphql/message-added.subscription';const styles = StyleSheet.create({
container: {... });
} // we don't resubscribe on changed props
// because it never happens in our app
if (!this.subscription) {
this.subscription = nextProps.subscribeToMore({
document: MESSAGE_ADDED_SUBSCRIPTION,
variables: {
userId: 1, // fake the user for now
groupIds: [nextProps.navigation.state.params.groupId],
},
updateQuery: (previousResult, { subscriptionData }) => {
const newMessage = subscriptionData.data.messageAdded; return update(previousResult, {
group: {
messages: {
edges: {
$unshift: [{
__typename: 'MessageEdge',
node: newMessage,
cursor: Buffer.from(newMessage.id.toString()).toString('base64'),
}],
},
},
},
});
},
});
} this.setState({
usernameColors,
});... }),
loading: PropTypes.bool,
loadMoreEntries: PropTypes.func,
subscribeToMore: PropTypes.func,
};const ITEMS_PER_PAGE = 10;... first: ITEMS_PER_PAGE,
},
}),
props: ({ data: { fetchMore, loading, group, subscribeToMore } }) => ({
loading,
group,
subscribeToMore,
loadMoreEntries() {
return fetchMore({
// query: ... (you can specify a different query.
After we connect subscribeToMore to the component’s props, we attach a subscription property on the component (so there’s only one) which initializes subscribeToMore with the required parameters. Inside updateQuery, when we receive a new message, we make sure its not a duplicate, and then unshift the message onto our collection of messages.
Does it work?!
We need to subscribe to new Groups and Messages so our Groups component will update in real time. The Groups component needs to subscribe to groupAdded and messageAdded because in addition to new groups popping up when they’re created, the latest messages should also show up in each group’s preview.
However, instead of using subscribeToMore in our Groups screen, we should actually consider applying these subscriptions to a higher order component (HOC) for our application. If we navigate away from the Groups screen at any point, we will unsubscribe and won't receive real-time updates while we're away from the screen. We'd need to refetch queries from the network when returning to the Groups screen to guarantee that our data is up to date.
If we attach our subscription to a higher order component, like AppWithNavigationState, we can stay subscribed to the subscriptions no matter where the user navigates and always keep our state up to date in real time!
Let’s apply the USER_QUERY to AppWithNavigationState in client/src/navigation.js and include two subscriptions using subscribeToMore for new Messages and Groups:
Step 6.12: Apply subscribeToMore to AppWithNavigationState
Changed client/src/navigation.js
...} from 'react-navigation-redux-helpers';
import { Text, View, StyleSheet } from 'react-native';
import { connect } from 'react-redux';
import { graphql, compose } from 'react-apollo';
import update from 'immutability-helper';
import { map } from 'lodash';
import { Buffer } from 'buffer';import Groups from './screens/groups.screen';
import Messages from './screens/messages.screen';...import GroupDetails from './screens/group-details.screen';
import NewGroup from './screens/new-group.screen';import { USER_QUERY } from './graphql/user.query';
import MESSAGE_ADDED_SUBSCRIPTION from './graphql/message-added.subscription';
import GROUP_ADDED_SUBSCRIPTION from './graphql/group-added.subscription';const styles = StyleSheet.create({
container: {
flex: 1,...const addListener = createReduxBoundAddListener("root");class AppWithNavigationState extends Component {
componentWillReceiveProps(nextProps) {
if (!nextProps.user) {
if (this.groupSubscription) {
this.groupSubscription();
} if (this.messagesSubscription) {
this.messagesSubscription();
}
} if (nextProps.user &&
(!this.props.user || nextProps.user.groups.length !== this.props.user.groups.length)) {
// unsubscribe from old if (typeof this.messagesSubscription === 'function') {
this.messagesSubscription();
}
// subscribe to new
if (nextProps.user.groups.length) {
this.messagesSubscription = nextProps.subscribeToMessages();
}
} if (!this.groupSubscription && nextProps.user) {
this.groupSubscription = nextProps.subscribeToGroups();
}
} render() {
return (
<AppNavigator navigation={addNavigationHelpers({... }
}AppWithNavigationState.propTypes = {
dispatch: PropTypes.func.isRequired,
nav: PropTypes.object.isRequired,
subscribeToGroups: PropTypes.func,
subscribeToMessages: PropTypes.func,
user: PropTypes.shape({
id: PropTypes.number.isRequired,
email: PropTypes.string.isRequired,
groups: PropTypes.arrayOf(
PropTypes.shape({
id: PropTypes.number.isRequired,
name: PropTypes.string.isRequired,
}),
),
}),
};const mapStateToProps = state => ({
nav: state.nav,
});const userQuery = graphql(USER_QUERY, {
options: () => ({ variables: { id: 1 } }), // fake the user for now
props: ({ data: { loading, user, subscribeToMore } }) => ({
loading,
user,
subscribeToMessages() {
return subscribeToMore({
document: MESSAGE_ADDED_SUBSCRIPTION,
variables: {
userId: 1, // fake the user for now
groupIds: map(user.groups, 'id'),
},
updateQuery: (previousResult, { subscriptionData }) => {
const previousGroups = previousResult.user.groups;
const newMessage = subscriptionData.data.messageAdded; const groupIndex = map(previousGroups, 'id').indexOf(newMessage.to.id); return update(previousResult, {
user: {
groups: {
[groupIndex]: {
messages: {
edges: {
$set: [{
__typename: 'MessageEdge',
node: newMessage,
cursor: Buffer.from(newMessage.id.toString()).toString('base64'),
}],
},
},
},
},
},
});
},
});
},
subscribeToGroups() {
return subscribeToMore({
document: GROUP_ADDED_SUBSCRIPTION,
variables: { userId: user.id },
updateQuery: (previousResult, { subscriptionData }) => {
const newGroup = subscriptionData.data.groupAdded; return update(previousResult, {
user: {
groups: { $push: [newGroup] },
},
});
},
});
},
}),
});export default compose(
connect(mapStateToProps),
userQuery,
)(AppWithNavigationState);
We have to do a little extra work to guarantee that our messageSubscription updates when we add or remove new groups. Otherwise, if a new group is created and someone sends a message, the user won’t be subscribed to receive that new message. When we need to update the subscription, we unsubscribe by calling the subscription as a function messageSubscription() and then reset messageSubscriptionto reflect the latest nextProps.subscribeToMessages.
One of the cooler things about Apollo is it caches all the queries and data that we’ve fetched and reuses data for the same query in the future instead of requesting it from the network (unless we specify otherwise). USER_QUERY will make a request to the network and then data will be reused for subsequent executions. Our app setup tracks any data changes with subscriptions, so we only end up requesting the data we need from the server once!
Handling broken connections
We need to do one more step to make sure our app stays updated in real-time. Sometimes users will lose internet connectivity or the WebSocket might disconnect temporarily. During these blackout periods, our client won’t receive any subscription events, so our app won’t receive new messages or groups. subscriptions-transport-ws has a built-in reconnecting mechanism, but it won't track any missed subscription events.
The simplest way to handle this issue is to refetch all relevant queries when our app reconnects. wsClient exposes an onReconnectedfunction that will call the supplied callback function when the WebSocket reconnects. We can simply call refetch on our queries in the callback.
Step 6.13: Apply onReconnected to refetch missed subscription events
Changed client/src/navigation.js
...import MESSAGE_ADDED_SUBSCRIPTION from './graphql/message-added.subscription';
import GROUP_ADDED_SUBSCRIPTION from './graphql/group-added.subscription';import { wsClient } from './app';const styles = StyleSheet.create({
container: {
flex: 1,... if (this.messagesSubscription) {
this.messagesSubscription();
} // clear the event subscription
if (this.reconnected) {
this.reconnected();
}
} else if (!this.reconnected) {
this.reconnected = wsClient.onReconnected(() => {
this.props.refetch(); // check for any data lost during disconnect
}, this);
} if (nextProps.user &&...AppWithNavigationState.propTypes = {
dispatch: PropTypes.func.isRequired,
nav: PropTypes.object.isRequired,
refetch: PropTypes.func,
subscribeToGroups: PropTypes.func,
subscribeToMessages: PropTypes.func,
user: PropTypes.shape({...
const userQuery = graphql(USER_QUERY, {
options: () => ({ variables: { id: 1 } }), // fake the user for now
props: ({ data: { loading, user, refetch, subscribeToMore } }) => ({
loading,
user,
refetch,
subscribeToMessages() {
return subscribeToMore({
document: MESSAGE_ADDED_SUBSCRIPTION,
Changed client/src/screens/messages.screen.js
...import _ from 'lodash';
import moment from 'moment';import { wsClient } from '../app';import Message from '../components/message.component';
import MessageInput from '../components/message-input.component';
import GROUP_QUERY from '../graphql/group.query';... });
} if (!this.reconnected) {
this.reconnected = wsClient.onReconnected(() => {
this.props.refetch(); // check for any data lost during disconnect
}, this);
} this.setState({
usernameColors,
});
} else if (this.reconnected) {
// remove event subscription
this.reconnected();
}
}... }),
loading: PropTypes.bool,
loadMoreEntries: PropTypes.func,
refetch: PropTypes.func,
subscribeToMore: PropTypes.func,
};... first: ITEMS_PER_PAGE,
},
}),
props: ({ data: { fetchMore, loading, group, refetch, subscribeToMore } }) => ({
loading,
group,
refetch,
subscribeToMore,
loadMoreEntries() {
return fetchMore({
Final product:
Almost Done!
That’s it for Part 6. In Part 7, final core post in this series, we will add authentication to our application. With authentication added, we will have a full fledged, real-time chat app!
As always, please share your thoughts, questions, struggles, and breakthroughs below!
