Merge pull request ably#604 from ably/publishBatch-params

Rest.publishBatch: support overloaded method that takes params

Author: QuintinWillison

lib/src/main/java/io/ably/lib/rest/AblyBase.java

@@ -239,15 +239,25 @@ public void requestAsync(String method, String path, Param[] params, HttpCore.Re
      */
     @Experimental
     public PublishResponse[] publishBatch(Message.Batch[] pubSpecs, ChannelOptions channelOptions) throws AblyException {
-        return publishBatchImpl(pubSpecs, channelOptions).sync();
+        return publishBatchImpl(pubSpecs, channelOptions, null).sync();
+    }
+
+    @Experimental
+    public PublishResponse[] publishBatch(Message.Batch[] pubSpecs, ChannelOptions channelOptions, Param[] params) throws AblyException {
+        return publishBatchImpl(pubSpecs, channelOptions, params).sync();
     }
 
     @Experimental
     public void publishBatchAsync(Message.Batch[] pubSpecs, ChannelOptions channelOptions, final Callback<PublishResponse[]> callback) throws AblyException {
-        publishBatchImpl(pubSpecs, channelOptions).async(callback);
+        publishBatchImpl(pubSpecs, channelOptions, null).async(callback);
+    }
+
+    @Experimental
+    public void publishBatchAsync(Message.Batch[] pubSpecs, ChannelOptions channelOptions, Param[] params, final Callback<PublishResponse[]> callback) throws AblyException {
+        publishBatchImpl(pubSpecs, channelOptions, params).async(callback);
     }
 
-    private Http.Request<PublishResponse[]> publishBatchImpl(final Message.Batch[] pubSpecs, ChannelOptions channelOptions) throws AblyException {
+    private Http.Request<PublishResponse[]> publishBatchImpl(final Message.Batch[] pubSpecs, ChannelOptions channelOptions, final Param[] params) throws AblyException {
         boolean hasClientSuppliedId = false;
         for(Message.Batch spec : pubSpecs) {
             for(Message message : spec.messages) {
@@ -270,7 +280,7 @@ private Http.Request<PublishResponse[]> publishBatchImpl(final Message.Batch[] p
             @Override
             public void execute(HttpScheduler http, final Callback<PublishResponse[]> callback) throws AblyException {
                 HttpCore.RequestBody requestBody = options.useBinaryProtocol ? MessageSerializer.asMsgpackRequest(pubSpecs) : MessageSerializer.asJSONRequest(pubSpecs);
-                http.post("/messages", HttpUtils.defaultAcceptHeaders(options.useBinaryProtocol), null, requestBody, new HttpCore.ResponseHandler<PublishResponse[]>() {
+                http.post("/messages", HttpUtils.defaultAcceptHeaders(options.useBinaryProtocol), params, requestBody, new HttpCore.ResponseHandler<PublishResponse[]>() {
                     @Override
                     public PublishResponse[] handleResponse(HttpCore.Response response, ErrorInfo error) throws AblyException {
                         if(error != null && error.code != 40020) {

lib/src/test/java/io/ably/lib/test/rest/RestChannelBulkPublishTest.java

@@ -5,9 +5,12 @@
 import java.util.Random;
 
 import io.ably.lib.test.common.ParameterizedTest;
+import io.ably.lib.test.common.Helpers.ChannelWaiter;
+import io.ably.lib.test.common.Helpers.MessageWaiter;
 import io.ably.lib.types.*;
 
 import io.ably.lib.rest.AblyRest;
+import io.ably.lib.realtime.*;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
@@ -18,15 +21,15 @@ public class RestChannelBulkPublishTest extends ParameterizedTest  {
 
     /**
      * Publish a single message on multiple channels
-     * 
+     *
      * The payload constructed has the form
      * [
      *   {
      *     channel: [ <channel 0>, <channel 1>, ... ],
      *     message: [{ data: <message text> }]
      *   }
      * ]
-     * 
+     *
      * It publishes the given message on all of the given channels.
      */
     @Test
@@ -69,9 +72,66 @@ public void bulk_publish_multiple_channels_simple() {
         }
     }
 
+    /**
+     * As above but with the param method
+     */
+    @Test
+    public void bulk_publish_multiple_channels_param() {
+        AblyRealtime rxAbly = null;
+        try {
+            /* setup library instance */
+            ClientOptions opts = createOptions(testVars.keys[0].keyStr);
+            AblyRest ably = new AblyRest(opts);
+            rxAbly = new AblyRealtime(opts);
+
+            /* first, publish some messages */
+            int channelCount = 5;
+            ArrayList<String> channelIds = new ArrayList<String>();
+            ArrayList<MessageWaiter> rxWaiters = new ArrayList<MessageWaiter>();
+            for (int i = 0; i < channelCount; i++) {
+                String channelId = "persisted:" + randomString();
+                channelIds.add(channelId);
+                Channel rxChannel = rxAbly.channels.get(channelId);
+                MessageWaiter messageWaiter = new MessageWaiter(rxChannel);
+                rxWaiters.add(messageWaiter);
+                new ChannelWaiter(rxChannel).waitFor(ChannelState.attached);
+            }
+
+            Message message = new Message(null, "bulk_publish_multiple_channels_param");
+            String messageId = message.id = randomString();
+            Message.Batch payload = new Message.Batch(channelIds, Collections.singleton(message));
+
+            Param[] params = new Param[]{new Param("quickAck", "true")};
+
+            PublishResponse[] result = ably.publishBatch(new Message.Batch[]{payload}, null, params);
+            for (PublishResponse response : result) {
+                assertEquals("Verify expected response id", response.messageId, messageId);
+                assertTrue("Verify expected channel name", channelIds.contains(response.channelId));
+                assertNull("Verify no publish error", response.error);
+            }
+
+            /* Wait to get a message on each channel -- since we're using
+             * quickAck, getting an ack is no longer a guarantee that the
+             * message has been processed, so getting history immediately after
+             * the publish returns may fail */
+            for (MessageWaiter messageWaiter : rxWaiters) {
+                messageWaiter.waitFor(1);
+            }
+        } catch (AblyException e) {
+            e.printStackTrace();
+            fail("bulk_publish_multiple_channels_param: Unexpected exception");
+            return;
+        } finally {
+            if(rxAbly != null) {
+                rxAbly.close();
+            }
+        }
+
+    }
+
     /**
      * Publish a multiple messages on multiple channels
-     * 
+     *
      * The payload constructed has the form
      * [
      *   {
@@ -94,7 +154,7 @@ public void bulk_publish_multiple_channels_simple() {
      *   },
      *   ...
      * ]
-     * 
+     *
      * It publishes the given messages on the associated channels.
      */
     @Test