Realtime controller

.ci/doc/templates/print-result-array.tpl.java

@@ -2,6 +2,7 @@
 import io.kuzzle.sdk.Protocol.WebSocket;
 import io.kuzzle.sdk.Options.Protocol.WebSocketOptions;
 import io.kuzzle.sdk.Options.KuzzleOptions;
+import io.kuzzle.sdk.Options.SubscribeOptions;
 import io.kuzzle.sdk.CoreClasses.Responses.Response;
 
 import java.util.concurrent.ConcurrentHashMap;

.ci/doc/templates/print-result.tpl.java

@@ -2,6 +2,7 @@
 import io.kuzzle.sdk.Protocol.WebSocket;
 import io.kuzzle.sdk.Options.Protocol.WebSocketOptions;
 import io.kuzzle.sdk.Options.KuzzleOptions;
+import io.kuzzle.sdk.Options.SubscribeOptions;
 import io.kuzzle.sdk.Options.DocumentOptions;
 import io.kuzzle.sdk.CoreClasses.Responses.Response;
 

doc/3/controllers/realtime/count/index.md

@@ -0,0 +1,29 @@
+---
+code: true
+type: page
+title: count
+description: Returns the number of other connections sharing the same subscription.
+---
+
+# count
+
+Returns the number of other connections sharing the same subscription.
+
+## Arguments
+
+```java
+public CompletableFuture<Integer> count(final String roomId)
+  throws NotConnectedException, InternalException
+```
+
+| Argument  | Type              | Description          |
+|-----------|-------------------|----------------------|
+| `room_id` | <pre>String</pre> | Subscription room ID |
+
+## Return
+
+Returns the number of active connections using the same provided subscription room.
+
+## Usage
+
+<<< ./snippets/count.java

doc/3/controllers/realtime/count/snippets/count.java

@@ -0,0 +1,22 @@
+SubscribeOptions options = new SubscribeOptions();
+options.setSubscribeToSelf(true);
+
+ConcurrentHashMap<String, Object> filters = new ConcurrentHashMap<>();
+filters.put("exists", "name");
+
+ConcurrentHashMap<String, Object> document = new ConcurrentHashMap<>();
+document.put("name", "nina-vkote");
+
+final String roomId = kuzzle.getRealtimeController().subscribe(
+  "nyc-open-data",
+  "yellow-taxi",
+  filters,
+  notification -> {
+    if (notification.scope.equals(SubscribeOptions.Scope.IN.toString())) {
+      System.out.println("Document entered the scope");
+    } else {
+      System.out.println("Document left the scope");
+    }
+}).get();
+
+final Integer result = kuzzle.getRealtimeController().count(roomId).get();

doc/3/controllers/realtime/count/snippets/count.test.yml

@@ -0,0 +1,7 @@
+name: Realtime#count
+description: Counts subscribers for a subscription room.
+hooks:
+  before:
+  after:
+template: print-result
+expected: 1

doc/3/controllers/realtime/publish/index.md

@@ -0,0 +1,34 @@
+---
+code: true
+type: page
+title: publish
+description: Publishes a real-time message.
+---
+
+# publish
+
+Sends a real-time message to Kuzzle. The message will be broadcasted to all clients with subscriptions matching the index, the collection and the message content.
+
+The index and collection are indicative and serve only to distinguish the rooms. They are not required to exist in the database.
+
+**Note:** real-time messages are not persisted in the database.
+
+## Arguments
+
+```java
+public CompletableFuture<Integer> publish(
+  final String index, 
+  final String collection, 
+  final ConcurrentHashMap<String, Object> message) 
+    throws NotConnectedException, InternalException
+```
+
+| Argument     | Type               | Description                         |
+|--------------|--------------------|-------------------------------------|
+| `index`      | <pre>String</pre>  | Index name                          |
+| `collection` | <pre>String</pre>  | Collection name                     |
+| `message`    | <pre>ConcurrentHashMap<String, Object></pre> | ConcurrentHashMap representing a JSON payload |
+
+## Usage
+
+<<< ./snippets/publish.java

doc/3/controllers/realtime/publish/snippets/publish.java

@@ -0,0 +1,4 @@
+ConcurrentHashMap<String, Object> document = new ConcurrentHashMap<>();
+document.put("name", "nina-vkote");
+
+kuzzle.getRealtimeController().publish("my-index", "my-collection", document);

doc/3/controllers/realtime/publish/snippets/publish.test.yml

@@ -0,0 +1,7 @@
+name: Realtime#publish
+description: Publishes a realtime message.
+hooks:
+  before:
+  after:
+template: default
+expected: Success

doc/3/controllers/realtime/unsubscribe/index.md

@@ -0,0 +1,25 @@
+---
+code: true
+type: page
+title: unsubscribe
+description: Removes a subscription.
+---
+
+# unsubscribe
+
+Removes a subscription.
+
+## Arguments
+
+```java
+public CompletableFuture<Void> unsubscribe(final String roomId) 
+  throws NotConnectedException, InternalException
+```
+
+| Argument  | Type               | Description          |
+|-----------|--------------------|----------------------|
+| `room_id` | <pre>String</pre>  | Subscription room ID |
+
+## Usage
+
+<<< ./snippets/unsubscribe.java

doc/3/controllers/realtime/unsubscribe/snippets/unsubscribe.java

@@ -0,0 +1,22 @@
+SubscribeOptions options = new SubscribeOptions();
+options.setSubscribeToSelf(true);
+
+ConcurrentHashMap<String, Object> filters = new ConcurrentHashMap<>();
+filters.put("exists", "name");
+
+ConcurrentHashMap<String, Object> document = new ConcurrentHashMap<>();
+document.put("name", "nina-vkote");
+
+final String roomId = kuzzle.getRealtimeController().subscribe(
+  "nyc-open-data",
+  "yellow-taxi",
+  filters,
+  notification -> {
+    if (notification.scope.equals(SubscribeOptions.Scope.IN.toString())) {
+      System.out.println("Document entered the scope");
+    } else {
+      System.out.println("Document left the scope");
+    }
+}).get();
+
+kuzzle.getRealtimeController().unsubscribe(roomId).get();

doc/3/controllers/realtime/unsubscribe/snippets/unsubscribe.test.yml

@@ -0,0 +1,7 @@
+name: Realtime#unsubscribe
+description: Removes a subscription.
+hooks:
+  before:
+  after:
+template: default
+expected: Success

src/main/java/io/kuzzle/sdk/API/Controllers/RealtimeController.java

@@ -54,6 +54,58 @@ public RealtimeController(Kuzzle kuzzle) {
     });
   }
 
+  /**
+   * Returns the number of other connections sharing the same subscription.
+   *
+   * @param roomId
+   * @return
+   * @throws NotConnectedException
+   * @throws InternalException
+   */
+  public CompletableFuture<Integer> count(final String roomId) throws NotConnectedException, InternalException {
+    return kuzzle
+        .query(new KuzzleMap()
+            .put("controller", "realtime")
+            .put("action", "count")
+            .put("body", new KuzzleMap().put("room_id", roomId)))
+        .thenApplyAsync((response) -> ((KuzzleMap) response.result).getNumber("count").intValue());
+  }
+
+  /**
+   * Sends a real-time message to Kuzzle. The message will be dispatched to
+   * all clients with subscriptions matching the index, the collection and
+   * the message content.
+   *
+   * @param index
+   * @param collection
+   * @param message
+   * @return
+   * @throws NotConnectedException
+   * @throws InternalException
+   */
+  public CompletableFuture<Void> publish(final String index, final String collection, final ConcurrentHashMap<String, Object> message) throws NotConnectedException, InternalException {
+    return kuzzle
+        .query(new KuzzleMap()
+            .put("controller", "realtime")
+            .put("action", "publish")
+            .put("index", index)
+            .put("collection", collection)
+            .put("body", new KuzzleMap().put("message", message)))
+        .thenApplyAsync((response) -> null);
+  }
+
+  /**
+   * Subscribe to a collection.
+   *
+   * @param index
+   * @param collection
+   * @param filters
+   * @param handler
+   * @param options
+   * @return
+   * @throws NotConnectedException
+   * @throws InternalException
+   */
   public CompletableFuture<String> subscribe(final String index, final String collection, final ConcurrentHashMap<String, Object> filters, final NotificationHandler handler, final SubscribeOptions options) throws NotConnectedException, InternalException {
     ConcurrentHashMap<String, Object> queryOptions = new ConcurrentHashMap<>();
     boolean subscribeToSelf = true;
@@ -98,4 +150,27 @@ public CompletableFuture<String> subscribe(final String index, final String coll
   public CompletableFuture<String> subscribe(final String index, final String collection, final ConcurrentHashMap<String, Object> filters, final NotificationHandler handler) throws NotConnectedException, InternalException {
     return this.subscribe(index, collection, filters, handler, null);
   }
+
+  /**
+   * Removes a subscription
+   *
+   * @param roomId
+   * @return
+   * @throws NotConnectedException
+   * @throws InternalException
+   */
+  public CompletableFuture<Void> unsubscribe(final String roomId) throws NotConnectedException, InternalException {
+    return kuzzle
+        .query(new KuzzleMap()
+            .put("controller", "realtime")
+            .put("action", "unsubscribe")
+            .put("body", new KuzzleMap().put("roomId", roomId)))
+        .thenApplyAsync((response) -> {
+          ArrayList<Subscription> subs = subscriptions.get(roomId);
+          if (subs != null) {
+            subscriptions.get(roomId).clear();
+          }
+          return null;
+        });
+  }
 }

src/test/java/io/kuzzle/test/API/Controllers/RealtimeControllerTest.java

@@ -23,6 +23,41 @@
 public class RealtimeControllerTest {
   private AbstractProtocol networkProtocol = Mockito.mock(WebSocket.class);
 
+  @Test
+  public void countTest() throws NotConnectedException, InternalException {
+    Kuzzle kuzzleSpy = spy(new Kuzzle(networkProtocol));
+    NotificationHandler notificationHandler = mock(NotificationHandler.class);
+
+    ArgumentCaptor arg = ArgumentCaptor.forClass(KuzzleMap.class);
+
+    kuzzleSpy.getRealtimeController().count("roomId");
+    verify(kuzzleSpy).query((KuzzleMap) arg.capture());
+
+    assertEquals("realtime", ((KuzzleMap) arg.getValue()).getString("controller"));
+    assertEquals("count", ((KuzzleMap) arg.getValue()).getString("action"));
+    assertEquals("roomId", ((KuzzleMap)((KuzzleMap) arg.getValue()).get("body")).getString("room_id"));
+  }
+
+  @Test
+  public void publishTest() throws NotConnectedException, InternalException {
+    Kuzzle kuzzleSpy = spy(new Kuzzle(networkProtocol));
+    NotificationHandler notificationHandler = mock(NotificationHandler.class);
+
+    ArgumentCaptor arg = ArgumentCaptor.forClass(KuzzleMap.class);
+
+    ConcurrentHashMap<String, Object> message = new ConcurrentHashMap<>();
+    message.put("foo", "bar");
+
+    kuzzleSpy.getRealtimeController().publish("index", "collection", message);
+    verify(kuzzleSpy).query((KuzzleMap) arg.capture());
+
+    assertEquals("realtime", ((KuzzleMap) arg.getValue()).getString("controller"));
+    assertEquals("publish", ((KuzzleMap) arg.getValue()).getString("action"));
+    assertEquals("index", ((KuzzleMap) arg.getValue()).getString("index"));
+    assertEquals("collection", ((KuzzleMap) arg.getValue()).getString("collection"));
+    assertEquals("bar", ((ConcurrentHashMap<String, Object>)((KuzzleMap)((KuzzleMap) arg.getValue()).get("body")).get("message")).get("foo").toString());
+  }
+
   @Test
   public void subscribeTest() throws NotConnectedException, InternalException {
     Kuzzle kuzzleSpy = spy(new Kuzzle(networkProtocol));
@@ -83,4 +118,19 @@ public void notificationHandlerWithoutSubscribeToSelfTest() throws NotConnectedE
     kuzzleSpy.trigger(Event.unhandledResponse, response);
     verify(notificationHandler, never()).run(any(Response.class));
   }
+
+  @Test
+  public void unsubscribeTest() throws NotConnectedException, InternalException {
+    Kuzzle kuzzleSpy = spy(new Kuzzle(networkProtocol));
+    NotificationHandler notificationHandler = mock(NotificationHandler.class);
+
+    ArgumentCaptor arg = ArgumentCaptor.forClass(KuzzleMap.class);
+
+    kuzzleSpy.getRealtimeController().unsubscribe("roomId");
+    verify(kuzzleSpy).query((KuzzleMap) arg.capture());
+
+    assertEquals("realtime", ((KuzzleMap) arg.getValue()).getString("controller"));
+    assertEquals("unsubscribe", ((KuzzleMap) arg.getValue()).getString("action"));
+    assertEquals("roomId", ((KuzzleMap)((KuzzleMap) arg.getValue()).get("body")).getString("roomId"));
+  }
 }