Observe in real time

Observe model mutations in real time

You can subscribe to changes on your Models. This reacts dynamically to updates of data to the underlying Storage Engine, which could be the result of GraphQL Subscriptions as well as Queries or Mutations that run against the backing AppSync API if you are synchronizing with the cloud.

NOTE: AWS AppSync has an adjustable limit of 100 subscriptions per connection. DataStore automatically subscribes to create, update, and delete mutations for all models.

This means that GraphQL APIs with DataStore enabled are limited to 33 models and DynamoDB tables.

(3 subscriptions * 33 models = 99 subscriptions per connection).

However, You can request a service limit increase from AWS AppSync to meet the real-time requirements of your application.

1Amplify.DataStore.observe(Post.class,
2    cancelable -> Log.i("MyAmplifyApp", "Observation began."),
3    postChanged -> {
4        Post post = postChanged.item();
5        Log.i("MyAmplifyApp", "Post: " + post);
6    },
7    failure -> Log.e("MyAmplifyApp", "Observation failed.", failure),
8    () -> Log.i("MyAmplifyApp", "Observation complete.")
9);

1Amplify.DataStore.observe(Post::class.java,
2    { Log.i("MyAmplifyApp", "Observation began") },
3    {
4        val post = it.item()
5        Log.i("MyAmplifyApp", "Post: $post")
6    },
7    { Log.e("MyAmplifyApp", "Observation failed", it) },
8    { Log.i("MyAmplifyApp", "Observation complete") }
9)

1Amplify.DataStore.observe(Post::class)
2    .onStart { Log.i("MyAmplifyApp", "Observation began") }
3    .catch { Log.e("MyAmplifyApp", "Observation failed", it) }
4    .onCompletion { Log.i("MyAmplifyApp", "Observation complete") }
5    .collect { Log.i("MyAmplifyApp", "Post: ${it.item()}") }

1RxAmplify.DataStore.observe(Post.class)
2    .subscribe(
3        post -> Log.i("MyAmplifyApp", "Post: " + post),
4        failure -> Log.e("MyAmplifyApp", "Observation failed.", failure),
5        () -> Log.i("MyAmplifyApp", "Observation complete.")
6    );

Observe query results in real-time

observeQuery(...) returns an initial data set, similar to query(...), and also automatically subscribes to subsequent changes to the query.

The first snapshot returned from observeQuery will contain the same results as calling query directly on your Model - a collection of all the locally available records. Subsequent snapshots will be emitted as updates to the dataset and received in the local store.

While data is syncing from the cloud, snapshots will contain all of the items synced so far and an isSynced status of false. When the sync process is complete, a snapshot will be emitted with all the records in the local store and an isSynced status of true.

In addition to typical real-time use cases, observeQuery can be used on app launch to show your customers an initial data set from the local store while new data is being synced from cloud.

observeQuery also accepts the same predicates and sorting options as query.

1String tag = "ObserveQuery";
2Consumer<DataStoreQuerySnapshot<Post>> onQuerySnapshot = value ->{
3        Log.d(tag, "success on snapshot");
4        Log.d(tag, "number of records: " + value.getItems().size());
5        Log.d(tag, "sync status: " + value.getIsSynced());
6};
7Consumer<Cancelable> observationStarted = value ->{
8        Log.d(tag, "success on cancelable");
9};
10Consumer<DataStoreException> onObservationError = value ->{
11        Log.d(tag, "error on snapshot$value");
12};
13Action onObservationComplete = () ->{
14        Log.d(tag, "complete");
15};
16QueryPredicate predicate =
17        Post.TITLE.beginsWith("post").and(Post.RATING.gt(10));
18QuerySortBy querySortBy = new QuerySortBy("post", "rating", QuerySortOrder.ASCENDING);
19List<QuerySortBy> sortByList = new ArrayList<QuerySortBy>();
20sortByList.add(querySortBy);
21ObserveQueryOptions options = new ObserveQueryOptions(predicate, sortByList);
22Amplify.DataStore.<Post>observeQuery(
23        Post.class,
24        options,
25        observationStarted,
26        onQuerySnapshot,
27        onObservationError,
28        onObservationComplete
29);

1val tag = "ObserveQuery"
2val onQuerySnapshot: Consumer<DataStoreQuerySnapshot<Post>> =
3    Consumer<DataStoreQuerySnapshot<Post>> { value: DataStoreQuerySnapshot<Post> ->
4        Log.d(tag, "success on snapshot")
5        Log.d(tag, "number of records: " + value.items.size)
6        Log.d(tag, "sync status: " + value.isSynced)
7    }
8
9val observationStarted =
10    Consumer { _: Cancelable ->
11        Log.d(tag, "success on cancelable")
12    }
13val onObservationError =
14    Consumer { value: DataStoreException ->
15        Log.d(tag, "error on snapshot $value")
16    }
17val onObservationComplete = Action {
18    Log.d(tag, "complete")
19}
20val predicate: QueryPredicate =
21    Post.TITLE.beginsWith("post").and(Post.RATING.gt(10))
22val querySortBy = QuerySortBy("post", "rating", QuerySortOrder.ASCENDING)
23
24val options = ObserveQueryOptions(predicate, listOf(querySortBy))
25Amplify.DataStore.observeQuery(
26    Post::class.java,
27    options,
28    observationStarted,
29    onQuerySnapshot,
30    onObservationError,
31    onObservationComplete
32)

1val querySortBy = QuerySortBy("post", "rating", QuerySortOrder.ASCENDING)
2Amplify.DataStore.observeQuery(
3    Post::class,
4    ObserveQueryOptions(Post.TITLE.beginsWith("post").and(Post.RATING.gt(10)),
5        listOf(querySortBy))
6    ).onStart { Log.i("MyAmplifyApp", "Observation began") }
7        .catch { Log.e("MyAmplifyApp", "Observation failed", it) }
8        .onCompletion { Log.i("MyAmplifyApp", "Observation complete") }
9        .collect { value -> Log.i("MyAmplifyApp", "Blog: ${value.items.size}") }

1QueryPredicate predicate =
2    Post.TITLE.beginsWith("post")
3        .and(Post.RATING.gt(10));
4QuerySortBy querySortBy = new QuerySortBy("post", "rating", QuerySortOrder.ASCENDING);
5List<QuerySortBy> sortByList = new ArrayList<QuerySortBy>();
6sortByList.add(querySortBy);
7ObserveQueryOptions options = new ObserveQueryOptions(predicate, sortByList);
8rxDataStore.observeQuery(Model.class, options).subscribe(
9    modelDataStoreQuerySnapshot -> {
10        Log.d("mySampleApp", "success on snapshot");
11        Log.d("mySampleApp", "number of records: " + modelDataStoreQuerySnapshot.getItems().size());
12        Log.d("mySampleApp", "sync status: " + modelDataStoreQuerySnapshot.getIsSynced());
13    });

Conflict resolution

DataStore events