Subscribe to real-time events

In this guide, we will outline the benefits of enabling real-time data integrations and how to set up and filter these subscriptions. We will also cover how to unsubscribe from subscriptions.

Before you begin, you will need:

• An application connected to the API
• Data already created to modify

Set up a real-time subscription

Subscriptions is a GraphQL feature that allows the server to send data to its clients when a specific event happens. For example, you can subscribe to an event when a new record is created, updated, or deleted through the API. You can enable real-time data integration in your app with a subscription.

1import { Amplify, API } from 'aws-amplify';
2import { GraphQLSubscription } from '@aws-amplify/api';
3import * as subscriptions from './graphql/subscriptions';
4import {
5  OnCreateTodoSubscription,
6  OnUpdateTodoSubscription,
7  OnDeleteTodoSubscription
8} from './API';
9
10// Subscribe to creation of Todo
11const createSub = API.graphql<GraphQLSubscription<OnCreateTodoSubscription>>({
12  query: subscriptions.onCreateTodo
13}).subscribe({
14  next: ({ provider, value }) => console.log({ provider, value }),
15  error: (error) => console.warn(error)
16});
17
18// Subscribe to update of Todo
19const updateSub = API.graphql<GraphQLSubscription<OnUpdateTodoSubscription>>({
20  query: subscriptions.onUpdateTodo
21}).subscribe({
22  next: ({ provider, value }) => console.log({ provider, value }),
23  error: (error) => console.warn(error)
24});
25
26// Subscribe to deletion of Todo
27const deleteSub = API.graphql<GraphQLSubscription<OnDeleteTodoSubscription>>({
28  query: subscriptions.onDeleteTodo
29}).subscribe({
30  next: ({ provider, value }) => console.log({ provider, value }),
31  error: (error) => console.warn(error)
32});
33// Stop receiving data updates from the subscription
34createSub.unsubscribe();
35updateSub.unsubscribe();
36deleteSub.unsubscribe();

1import { Amplify, API } from 'aws-amplify';
2import * as subscriptions from './graphql/subscriptions';
3
4// Subscribe to creation of Todo
5const createSub = API.graphql({ query: subscriptions.onCreateTodo }).subscribe({
6  next: ({ provider, value }) => console.log({ provider, value }),
7  error: (error) => console.warn(error)
8});
9
10// Subscribe to update of Todo
11const updateSub = API.graphql({ query: subscriptions.onUpdateTodo }).subscribe({
12  next: ({ provider, value }) => console.log({ provider, value }),
13  error: (error) => console.warn(error)
14});
15
16// Subscribe to deletion of Todo
17const deleteSub = API.graphql({ query: subscriptions.onDeleteTodo }).subscribe({
18  next: ({ provider, value }) => console.log({ provider, value }),
19  error: (error) => console.warn(error)
20});
21
22// Stop receiving data updates from the subscription
23createSub.unsubscribe();
24updateSub.unsubscribe();
25deleteSub.unsubscribe();

Set up server-side subscription filters

Subscriptions take an optional filter argument to define service-side subscription filters:

1// ...
2import { GraphQLSubscription } from '@aws-amplify/api';
3import {
4  OnCreateTodoSubscriptionVariables,
5  OnCreateTodoSubscription
6} from './API';
7
8const variables: OnCreateTodoSubscriptionVariables = {
9  filter: {
10    // Only receive Todo messages where the "type" field is "Personal"
11    type: { eq: 'Personal' }
12  }
13};
14
15const sub = API.graphql<GraphQLSubscription<OnCreateTodoSubscription>>({
16  query: subscriptions.onCreateTodo,
17  variables
18}).subscribe({
19  next: ({ provider, value }) => console.log({ provider, value }),
20  error: (error) => console.warn(error)
21});

1// ...
2
3const variables = {
4  filter: {
5    // Only receive Todo messages where the "type" field is "Personal"
6    type: { eq: 'Personal' }
7  }
8};
9
10const sub = API.graphql({
11  query: subscriptions.onCreateTodo,
12  variables
13}).subscribe({
14  next: ({ provider, value }) => console.log({ provider, value }),
15  error: (error) => console.warn(error)
16});

If you want to get all subscription events, don't specify any filter parameters.

Subscription connection status updates

Now that your application is set up and using subscriptions, you may want to know when the subscription is finally established, or reflect to your users when the subscription isn't healthy. You can monitor the connection state for changes through the Hub local eventing system.

1import { CONNECTION_STATE_CHANGE, ConnectionState } from '@aws-amplify/pubsub';
2import { Hub } from 'aws-amplify';
3
4Hub.listen('api', (data: any) => {
5  const { payload } = data;
6  if (payload.event === CONNECTION_STATE_CHANGE) {
7    const connectionState = payload.data.connectionState as ConnectionState;
8    console.log(connectionState);
9  }
10});

Subscription connection states

• Connected - Connected and working with no issues.
• ConnectedPendingDisconnect - The connection has no active subscriptions and is disconnecting.
• ConnectedPendingKeepAlive - The connection is open, but has missed expected keep-alive messages.
• ConnectedPendingNetwork - The connection is open, but the network connection has been disrupted. When the network recovers, the connection will continue serving traffic.
• Connecting - Attempting to connect.
• ConnectionDisrupted - The connection is disrupted and the network is available.
• ConnectionDisruptedPendingNetwork - The connection is disrupted and the network connection is unavailable.
• Disconnected - Connection has no active subscriptions and is disconnecting.

1// ...
2import { GraphQLQuery, GraphQLSubscription } from '@aws-amplify/api';
3import {
4  ListTodosQuery
5  OnCreateTodoSubscription,
6  OnUpdateTodoSubscription,
7  OnDeleteTodoSubscription
8} from './API';
9
10const fetchRecentData = () => {
11  // Retrieve some/all data from AppSync
12  const allTodos = await API.graphql<GraphQLQuery<ListTodosQuery>>({
13    query: queries.listTodos
14  });
15}
16
17let priorConnectionState: ConnectionState;
18
19Hub.listen("api", (data: any) => {
20  const { payload } = data;
21  if (
22    payload.event === CONNECTION_STATE_CHANGE
23  ) {
24
25    if (priorConnectionState === ConnectionState.Connecting && payload.data.connectionState === ConnectionState.Connected) {
26      fetchRecentData();
27    }
28    priorConnectionState = payload.data.connectionState;
29  }
30});
31
32const createSub = API.graphql<GraphQLSubscription<OnCreateTodoSubscription>>(
33  { query: subscriptions.onCreateTodo }
34).subscribe({
35  next: data => // Process incoming messages
36});
37
38const updateSub = API.graphql<GraphQLSubscription<OnUpdateTodoSubscription>>(
39  { query: subscriptions.onUpdateTodo }
40).subscribe({
41  next: data => // Process incoming messages
42});
43
44const deleteSub = API.graphql<GraphQLSubscription<OnDeleteTodoSubscription>>(
45  { query: subscriptions.onDeleteTodo }
46).subscribe({
47  next: data => // Process incoming messages
48});
49
50const cleanupSubscriptions = () => {
51  createSub.unsubscribe();
52  updateSub.unsubscribe();
53  deleteSub.unsubscribe();
54}

1// ...
2
3const fetchRecentData = () => {
4  // Retrieve some/all data from AppSync
5  const allTodos = await API.graphql({ query: queries.listTodos });
6}
7
8let priorConnectionState: ConnectionState;
9
10Hub.listen("api", (data: any) => {
11  const { payload } = data;
12  if (
13    payload.event === CONNECTION_STATE_CHANGE
14  ) {
15
16    if (priorConnectionState === ConnectionState.Connecting && payload.data.connectionState === ConnectionState.Connected) {
17      fetchRecentData();
18    }
19    priorConnectionState = payload.data.connectionState;
20  }
21});
22
23const createSub = API.graphql(
24  { query: subscriptions.onCreateTodo }
25).subscribe({
26  next: data => // Process incoming messages
27});
28
29const updateSub = API.graphql(
30  { query: subscriptions.onUpdateTodo }
31).subscribe({
32  next: data => // Process incoming messages
33});
34
35const deleteSub = API.graphql(
36  { query: subscriptions.onDeleteTodo }
37).subscribe({
38  next: data => // Process incoming messages
39});
40
41const cleanupSubscriptions = () => {
42  createSub.unsubscribe();
43  updateSub.unsubscribe();
44  deleteSub.unsubscribe();
45}

1type Post @model @auth(rules: [{ allow: public }]) {
2  id: ID!
3  title: String!
4  content: String
5  comments: [Comment] @hasMany
6}
7
8type Comment @model @auth(rules: [{ allow: public }]) {
9  id: ID!
10  content: String
11}

1# Post type
2onCreatePost
3onUpdatePost
4onDeletePost
5
6# Comment type
7onCreateComment
8onUpdateComment
9onDeleteComment

1type Post @model @auth(rules: [{ allow: public }]) {
2  id: ID!
3  title: String!
4  content: String
5  comments: [Comment] @hasMany
6}
7
8type Comment @model @auth(rules: [{ allow: public }]) {
9  id: ID!
10  content: String
11  postCommentsId: ID!
12}
13
14type Subscription {
15  onCommentByPostId(postCommentsId: ID!): Comment
16    @aws_subscribe(mutations: ["createComment"])
17}

1import { API } from 'aws-amplify';
2import { onCommentByPostId } from './graphql/subscriptions';
3
4API.graphql({
5  query: onCommentByPostId,
6  variables: {
7    postCommentsId: '12345'
8  }
9}).subscribe({
10  next: (data) => {
11    console.log('data: ', data);
12  }
13});

Unsubscribe from a subscription

You can also unsubscribe from events by using subscriptions by implementing the following:

1// Stop receiving data updates from the subscription
2sub.unsubscribe();

Conclusion

Congratulations! You have finished the Subscribe to real-time events guide. In this guide, you set up subscriptions for real-time events and learned how to filter and cancel these subscriptions when needed.

Next steps

Our recommended next steps include continuing to build out and customize your information architecture for your data. Some resources that will help with this work include:

• Authorization rules
• Data modeling
• Custom business logic