Update README.md

Author: toliaqat

README.md

@@ -1,11 +1,14 @@
 # Continuous Queries
 
-Continuous Queries are great way to get update notifications on your filters.
+Query tasks have a special option to create Continuous queries.
+Continuous queries are great way to get update notifications on your filters.
 Continuous query is started by creating a query task service with CONTINUOUS
-option. And then setting up a notification on that service. A continuous query
+option, and setting up a notification on that query task instance. A continuous query
 task does two things:
- 1. It will give the historical, existing data as the first notification
- 2. It will give you any further updates
+ 1. It will self patch, causing the initial notification, once the initial query results are available, providing the documents currently in the index.
+ 2. On any additional index updates that match the query associated with the
+task, it will once again self patch, with the result being the document that
+changed or was added. Once again this will cause notifications to be sent.
 
 In the remainder of the document we expect you have basic understanding of [[QueryTaskService]].
 
@@ -28,8 +31,8 @@ Here are the basic steps required to efficiently use the continuous query tasks.
 
 We can avoid setting up the subscription with query task here, and instead do
 the polling on this continues query task service for updates. But that would
-NOT be efficient. Instead we recommend using subscription model here and get
-the results whenever they are available from this friend on the other side.
+not be efficient. Instead we recommend using subscription model here and get
+the results whenever they are available.
 
 In rest of the document we will go over the steps mentioned above.
 
@@ -39,14 +42,17 @@ Following is the simple example of QueryTask in which CONTINUOUS query option is
 This query task is filtering the results based on Kind field clause.
 
 ```java
-QueryTask.Query query = QueryTask.Query.Builder.create()
-        .addKindFieldClause(Employee.class)
-        .build();
-
-QueryTask queryTask = QueryTask.Builder.create()
-        .addOption(QueryOption.EXPAND_CONTENT)
-        .addOption(QueryOption.CONTINUOUS)
-        .setQuery(query).build();
+public QueryTask createContinuousQuery() {
+    QueryTask.Query query = QueryTask.Query.Builder.create()
+            .addKindFieldClause(EmployeeService.Employee.class)
+            .build();
+
+    QueryTask queryTask = QueryTask.Builder.create()
+            .addOption(QueryOption.EXPAND_CONTENT)
+            .addOption(QueryOption.CONTINUOUS)
+            .setQuery(query).build();
+    return queryTask;
+}
 ```
 
 By default, query tasks services expire after few (10 at the time of writing)
@@ -93,96 +99,191 @@ Json payload of above query.
 }
 ```
 
-## Send the request
+## Sending the request and subscribing to notifications
 
-After sending the query we need to capture the returned query task service link
-and subscribe to it for any updates.  *NOTE:* One important thing to note here
-is that continuous query task service being a long running service that queries the
-index regularly for updates. Hence this should be used with care and should be
-called by a single host. If following code is running on a replicated service
-on multiple nodes, then we would be triggering this continues query task
-multiple times on our targetHost, which would be a over kill and can be big
-performance hit on your poor hosts.
+After creating the query task, we POST it to the local-query-tasks factory service
+with specific documentSelfLink (QUERY_TASK_LINK) that will create a query task with
+that name, helping us to point to this query task service easily in future.
+
+On creation of this continuous query task service we call `subscribeToContinuousQuery`
+to create the subscription to get notifications from query task service with updates.
+
+**Note:** We always use a *local* query task (CORE_LOCAL_QUERY_TASKS) for continuous queries!
+
+*See also* [[DeferredResult-Tutorial]]
 
 ```java
-Operation post = Operation.createPost(targetHost, ServiceUriPaths.CORE_LOCAL_QUERY_TASKS)
-     .setBody(queryTask)
-     .setReferer(this.clientHost.getUri());
-
-post.setCompletion((o, e) -> {
-     if (e != null) {
-          System.out.printf("Query failed %s", e.toString());
-          return;
-     }
-     QueryTask queryResponse = o.getBody(QueryTask.class);
-     subscribeToContinuousQueryTask(targetHost, queryResponse.documentSelfLink);
-);
-
-this.clientHost.sendRequest(post);
+public void createAndSubscribeToContinuousQuery(Operation op) {
+    QueryTask queryTask = createContinuousQuery();
+    queryTask.documentSelfLink = QUERY_TASK_LINK;
+    Operation post = Operation.createPost(getHost(), ServiceUriPaths.CORE_LOCAL_QUERY_TASKS)
+            .setBody(queryTask)
+            .setReferer(getHost().getUri());
+
+    getHost().sendWithDeferredResult(post)
+            .thenAccept((state) -> subscribeToContinuousQuery())
+            .whenCompleteNotify(op);
+}
 ```
 
-## Subscribe to the results
+Now let us see the code for `subscribeToContinuousQuery` which will create start
+subscription service to listen for any notifications from continuous query task service
+we created earlier. We are using `startSubscriptionService` API of the ServiceHost
+to setup the subscription service. This API is creating a subscription service with a
+callback URI, and registring that URI with the continuous query task service.
 
-Notice above, in the completion handler we are calling
-`subscribeToContinuousQueryTask` (shown below) method with the self-Link of the
-query task service.
+Please read the section "Details!" for mode details on creating a robust solution.
 
 ```java
-public void subscribeToContinuousQueryTask(ServiceHost host, String serviceLink) {
-     Consumer<Operation> target = this::processResults;
+public void subscribeToContinuousQuery() {
+    Operation post = Operation
+            .createPost(UriUtils.buildUri(getHost(), QUERY_TASK_LINK))
+            .setReferer(getHost().getUri());
 
-     Operation subPost = Operation
-          .createPost(UriUtils.buildUri(host, serviceLink))
-          .setReferer(host.getUri());
-
-     host.startSubscriptionService(subPost, target);
+    URI subscriptionUri = getHost().startSubscriptionService(post, this::processResults);
+    updateSubscriptionLink(subscriptionUri);
 }
-
 ```
 
-`subscribeToContinuousQueryTask` is just calling `startSubscriptionService`
-method on the local host that will listen for any notifications from the the
-target service and call our target method(`processResults`).
+
+*See also section on Subscriptions in [[Programming-Model]]*
 
 ## Process the results
 
 Following is the basic implementation of `processResults` that would be called
 whenever the query task service on the target host has any new data for us to
 process. We check for presence of the results and then loop over all the result
-documents to process.
+documents to process. During the processing we can check `state.documentUpdateAction`
+to see what was the last action (created, deleted) on that service.
 
 ```java
 public void processResults(Operation op) {
-     QueryTask body = op.getBody(QueryTask.class);
+    QueryTask body = op.getBody(QueryTask.class);
+
+    if (body.results == null || body.results.documentLinks.isEmpty()) {
+        return;
+    }
+
+    for (Object doc : body.results.documents.values()) {
+        EmployeeService.Employee state = Utils.fromJson(doc, EmployeeService.Employee.class);
+        getHost().log(Level.INFO, "Employee Name: %s, Action: %s", state.name, state.documentUpdateAction);
+    }
+}
+```
+
+## Details!
+So far we have went over the basics of continuous query tasks. But there are
+few things, we need to be aware of, when we are writing code for production.
+
+In a multi-node environment, one important thing to note
+is that continuous query task service, being a long running service, that hits the
+index regularly for updates. Hence this should be used with care and should be
+called by a single host. If above code is running on a replicated service
+on multiple nodes, then we would be triggering this continuous query task
+multiple times on our host, which would be a over kill and can be big
+performance hit on your hosts.
+
+That means when we are creating continuous queries, we need to make
+sure, that the caller with subscription to notifications is fault-tolarant and
+load-balanced. We do that by making a stateful service which is replicated,
+and owner-selected. Lets call this service a watch service.
+
+Because it will be a replicated service, it will give us
+fault-tolerance in case one of the replica goes down. And beause this is
+owner-selected we can make sure that only one node is doing the heavy lifting
+at anyone time, and other nodes are not dublicating the work. That watch
+service will create the continuous query task and watch for notifications from
+it. But not just that, we would want it to do this only on one of the
+replicated nodes, not on all nodes. How do we make sure that stateful service
+is only creating the query task in one of the replcas? We can do that by
+checking for the ownership of its state in handleNodeGroupMaintenance method.
+
+What is `handleNodeGroupMaintenance`? This is overridable method in stateful
+services that is invoked by the Xenon runtime when there is change in the
+ownership of the state of the service.
+
+We can override this method in our watch service and check if we are owner or
+not using `hasOption(ServiceOption.DOCUMENT_OWNER)`, and based on that create
+continuous query task if we are owner, and delete running continuous query task
+service and subscription if we are not the owner anymore.
+Following is the example code that implements `handleNodeGroupMaintenance` 
+
+```
+@Override
+public void handleNodeGroupMaintenance(Operation op) {
+    // Create continuous queries and subscriptions in case of change in node group topology.
+    if (hasOption(ServiceOption.DOCUMENT_OWNER)) {
+        createAndSubscribeToContinuousQuery(op);
+    } else {
+        deleteSubscriptionAndContinuousQuery(op);
+    }
+}
+```
 
-     if (body.results == null || body.results.documentLinks.isEmpty()) {
-          return;
-     }
+We have already seen the basic implementation of
+`createAndSubscribeToContinuousQueryTask` earlier. Now lets see what
+`deleteSubscriptionAndContinuousQuery` might look like in this watch service.
 
-     for (Object doc : body.results.documents.values()) {
-          Employee state = Utils.fromJson(doc, Employee.class);
-          System.out.printf("Name %s \n", state.name);
-     }
+```
+private void deleteSubscriptionAndContinuousQuery(Operation op) {
+    Operation unsubscribeOperation = Operation.createPost(UriUtils.buildUri(getHost(), QUERY_TASK_LINK))
+            .setReferer(getUri())
+            .setCompletion((o, e) -> {
+                updateSubscriptionLink(null);
+                deleteContinuousQuery();
+            });
+
+    getStateAndApply(state -> getHost().stopSubscriptionService(unsubscribeOperation,
+            UriUtils.buildUri(state.subscriptionLink)));
 }
 ```
+In above method we are first creating a `Operation` to unsubscribe the
+subscription service being created by the runtime. `stopSubscriptionService` deletes
+both the subscriber entry in the query task service and also deletes the subscription callback
+endpoint.
+On the completion of these deletions we are deleting the continuous query service by
+calling `deleteContinuousQuery`.
+
+Following is the implementation of `deleteContinuousQuery`.
+```
+private void deleteContinuousQuery() {
+    getHost().sendRequest(Operation
+            .createDelete(UriUtils.buildUri(getHost(), QUERY_TASK_LINK))
+            .setReferer(getUri()));
+}
+```
+
+Following is the implementation of `getStateAndApply`.
+```
+private void getStateAndApply(Consumer<? super State> action) {
+    Operation get = Operation
+            .createGet(this, this.getSelfLink())
+            .setReferer(getUri());
+
+    getHost().sendWithDeferredResult(get, State.class)
+            .thenAccept(action)
+            .whenCompleteNotify(get);
+} 
+```
+
 ## FAQ
 
 #### Can I do continuous query on a stateless service?
 No, because queries work on stateful and persisted services.
 
-#### How can I see list of all created continuous query services?
-You can do curl on `http://host/core/query-tasks` and `http://host/core/local-query-tasks` to see list of all query task services.
+#### How can I see the list of all created continuous query services?
+You can do curl on `http://[host]/core/local-query-tasks` to see list of all local query task services.
 
-#### My continuous query task is not there after few minutes. Why it got disappeared?
+#### My continuous query task is terminated after few minutes?
 It got expired after 10 minutes. Set it to never expire using `querytask.documentExpirationTimeMicros = Long.MAX_VALUE;`
 
 #### Why I am not getting any results in subscription of my continues query tasks service?
 Subscription handler will be called when there are any updates. Make sure your
 updates are being reflected on the index.
 
 #### Let’s say I’ve got a query for “all example services”, and one of the example services is deleted: do I learn that it’s no longer in the query result?
-You will get a PATCH, with the body being the last version of the example
-service when it was deleted. The documentUpdateAction will be DELETE.
+You will get a `PATCH`, with the body being the last version of the example
+service when it was deleted. The `documentUpdateAction` will be `DELETE`.
 
 #### You said, any update that satisfies the query filter will cause the results to be updated and a self PATCH to be sent on the service. Does that mean I’ll see a diff of new things that match that query?
 
@@ -192,21 +293,11 @@ with the specific update to a service that matched the query if your query no
 longer match anything, you get no notifications. You can cancel it, have it
 expire, etc.
 
-#### I want to get notification on every single create, update, delete on all services on my host. Will a single continuous query task do that?
-Please don't try this at your home(production). Yes, you can do that, but it
-will be a big performance hit on your host for the duration of your query task.
-Make sure you set expire to be short.
-
 #### When does a continuous query ends?
 It never ends, until it gets expired. You will keep getting notification as
-long as there are updates that fulfil continuous query task's filter.
+long as there are updates that fulfil continuous query task's filter or until it gets expired.
 
 #### How do you calculate a total sum using a continuous query when you don't know when your are done getting update notifications?
-Well a total sum implies you know the full set that you want to compute it
+Well, a total sum implies you know the full set that you want to compute it
 over, which means that you can use a normal query. If you want to keep
 counting, and do a running sum, then you need to use a continuous query.
-
-#### Ok, I am still confused. What does a continuous query actually do?
-A continuous query does two things:
- 1. It will give the historical, existing data as the first notification
- 2. It will give you any further updates