diff --git a/android/src/main/java/io/ably/lib/push/PushChannel.java b/android/src/main/java/io/ably/lib/push/PushChannel.java index 8ac796368..55b3a96d8 100644 --- a/android/src/main/java/io/ably/lib/push/PushChannel.java +++ b/android/src/main/java/io/ably/lib/push/PushChannel.java @@ -49,7 +49,7 @@ public void subscribeClient() throws AblyException { * @throws AblyException */ public void subscribeClientAsync(CompletionListener listener) { - subscribeClientImpl().async(new CompletionListener.ToCallback(listener)); + subscribeClientImpl().async(new CompletionListener.ToCallback<>(listener)); } protected Http.Request subscribeClientImpl() { @@ -83,7 +83,7 @@ public void subscribeDevice() throws AblyException { * @throws AblyException */ public void subscribeDeviceAsync(CompletionListener listener) { - subscribeDeviceImpl().async(new CompletionListener.ToCallback(listener)); + subscribeDeviceImpl().async(new CompletionListener.ToCallback<>(listener)); } protected Http.Request subscribeDeviceImpl() { @@ -131,7 +131,7 @@ public void unsubscribeClient() throws AblyException { * @throws AblyException */ public void unsubscribeClientAsync(CompletionListener listener) { - unsubscribeClientImpl().async(new CompletionListener.ToCallback(listener)); + unsubscribeClientImpl().async(new CompletionListener.ToCallback<>(listener)); } protected Http.Request unsubscribeClientImpl() { @@ -163,7 +163,7 @@ public void unsubscribeDevice() throws AblyException { * @throws AblyException */ public void unsubscribeDeviceAsync(CompletionListener listener) { - unsubscribeDeviceImpl().async(new CompletionListener.ToCallback(listener)); + unsubscribeDeviceImpl().async(new CompletionListener.ToCallback<>(listener)); } protected Http.Request unsubscribeDeviceImpl() { diff --git a/lib/src/main/java/io/ably/lib/push/PushBase.java b/lib/src/main/java/io/ably/lib/push/PushBase.java index dbfe31e10..d175e0290 100644 --- a/lib/src/main/java/io/ably/lib/push/PushBase.java +++ b/lib/src/main/java/io/ably/lib/push/PushBase.java @@ -80,7 +80,7 @@ public void publish(Param[] recipient, JsonObject payload) throws AblyException * @throws AblyException */ public void publishAsync(Param[] recipient, JsonObject payload, final CompletionListener listener) { - publishImpl(recipient, payload).async(new CompletionListener.ToCallback(listener)); + publishImpl(recipient, payload).async(new CompletionListener.ToCallback<>(listener)); } private Http.Request publishImpl(final Param[] recipient, final JsonObject payload) { @@ -275,7 +275,7 @@ public void remove(String deviceId) throws AblyException { * @param listener A listener to be notified of success or failure. */ public void removeAsync(String deviceId, CompletionListener listener) { - removeImpl(deviceId).async(new CompletionListener.ToCallback(listener)); + removeImpl(deviceId).async(new CompletionListener.ToCallback<>(listener)); } protected Http.Request removeImpl(final String deviceId) { @@ -310,7 +310,7 @@ public void removeWhere(Param[] params) throws AblyException { * @param listener A listener to be notified of success or failure. */ public void removeWhereAsync(Param[] params, CompletionListener listener) { - removeWhereImpl(params).async(new CompletionListener.ToCallback(listener)); + removeWhereImpl(params).async(new CompletionListener.ToCallback<>(listener)); } protected Http.Request removeWhereImpl(Param[] params) { @@ -435,7 +435,7 @@ public void remove(ChannelSubscription subscription) throws AblyException { * @throws AblyException */ public void removeAsync(ChannelSubscription subscription, CompletionListener listener) { - removeImpl(subscription).async(new CompletionListener.ToCallback(listener)); + removeImpl(subscription).async(new CompletionListener.ToCallback<>(listener)); } protected Http.Request removeImpl(ChannelSubscription subscription) { @@ -476,7 +476,7 @@ public void removeWhere(Param[] params) throws AblyException { * @throws AblyException */ public void removeWhereAsync(Param[] params, CompletionListener listener) { - removeWhereImpl(params).async(new CompletionListener.ToCallback(listener)); + removeWhereImpl(params).async(new CompletionListener.ToCallback<>(listener)); } protected Http.Request removeWhereImpl(Param[] params) { diff --git a/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java b/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java index 3e556e508..3ea1d0d6a 100644 --- a/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java +++ b/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java @@ -43,6 +43,7 @@ import io.ably.lib.types.ProtocolMessage; import io.ably.lib.types.ProtocolMessage.Action; import io.ably.lib.types.ProtocolMessage.Flag; +import io.ably.lib.types.PublishResult; import io.ably.lib.types.Summary; import io.ably.lib.types.UpdateDeleteResult; import io.ably.lib.util.CollectionUtils; @@ -443,6 +444,16 @@ private static void callCompletionListenerError(CompletionListener listener, Err } } + private static void callCompletionListenerError(Callback listener, ErrorInfo err) { + if(listener != null) { + try { + listener.onError(err); + } catch(Throwable t) { + Log.e(TAG, "Unexpected exception calling CompletionListener", t); + } + } + } + private void setAttached(ProtocolMessage message) { clearAttachTimers(); properties.attachSerial = message.channelSerial; @@ -1025,7 +1036,7 @@ private void unsubscribeImpl(String name, MessageListener listener) { * @throws AblyException */ public void publish(String name, Object data) throws AblyException { - publish(name, data, null); + publish(name, data, (Callback) null); } /** @@ -1037,7 +1048,7 @@ public void publish(String name, Object data) throws AblyException { * @throws AblyException */ public void publish(Message message) throws AblyException { - publish(message, null); + publish(message, (Callback) null); } /** @@ -1049,7 +1060,7 @@ public void publish(Message message) throws AblyException { * @throws AblyException */ public void publish(Message[] messages) throws AblyException { - publish(messages, null); + publish(messages, (Callback) null); } /** @@ -1065,12 +1076,34 @@ public void publish(Message[] messages) throws AblyException { *

* This listener is invoked on a background thread. * @throws AblyException + * @deprecated Use {@link #publish(String, Object, Callback)} instead. */ + @Deprecated public void publish(String name, Object data, CompletionListener listener) throws AblyException { Log.v(TAG, "publish(String, Object); channel = " + this.name + "; event = " + name); publish(new Message[] {new Message(name, data)}, listener); } + /** + * Publishes a single message to the channel with the given event name and payload. + * When publish is called with this client library, it won't attempt to implicitly attach to the channel, + * so long as transient publishing is available in the library. + * Otherwise, the client will implicitly attach. + *

+ * Spec: RTL6i + * @param name the event name + * @param data the message payload + * @param callback A callback may optionally be passed in to this call to be notified of success or failure of the operation, + * receiving a {@link PublishResult} with message serial(s) on success. + *

+ * This callback is invoked on a background thread. + * @throws AblyException + */ + public void publish(String name, Object data, Callback callback) throws AblyException { + Log.v(TAG, "publish(String, Object); channel = " + this.name + "; event = " + name); + publish(new Message[] {new Message(name, data)}, callback); + } + /** * Publishes a message to the channel. * When publish is called with this client library, it won't attempt to implicitly attach to the channel. @@ -1081,12 +1114,31 @@ public void publish(String name, Object data, CompletionListener listener) throw *

* This listener is invoked on a background thread. * @throws AblyException + * @deprecated Use {@link #publish(Message, Callback)} instead. */ + @Deprecated public void publish(Message message, CompletionListener listener) throws AblyException { Log.v(TAG, "publish(Message); channel = " + this.name + "; event = " + message.name); publish(new Message[] {message}, listener); } + /** + * Publishes a message to the channel. + * When publish is called with this client library, it won't attempt to implicitly attach to the channel. + *

+ * Spec: RTL6i + * @param message A {@link Message} object. + * @param callback A callback may optionally be passed in to this call to be notified of success or failure of the operation, + * receiving a {@link PublishResult} with message serial(s) on success. + *

+ * This callback is invoked on a background thread. + * @throws AblyException + */ + public void publish(Message message, Callback callback) throws AblyException { + Log.v(TAG, "publish(Message); channel = " + this.name + "; event = " + message.name); + publish(new Message[] {message}, callback); + } + /** * Publishes an array of messages to the channel. * When publish is called with this client library, it won't attempt to implicitly attach to the channel. @@ -1098,7 +1150,12 @@ public void publish(Message message, CompletionListener listener) throws AblyExc * This listener is invoked on a background thread. * @throws AblyException */ + @Deprecated public synchronized void publish(Message[] messages, CompletionListener listener) throws AblyException { + publish(messages, Listeners.fromCompletionListener(listener)); + } + + public synchronized void publish(Message[] messages, Callback listener) throws AblyException { Log.v(TAG, "publish(Message[]); channel = " + this.name); ConnectionManager connectionManager = ably.connection.connectionManager; ConnectionManager.State connectionState = connectionManager.getConnectionState(); @@ -1125,7 +1182,7 @@ public synchronized void publish(Message[] messages, CompletionListener listener case suspended: throw AblyException.fromErrorInfo(new ErrorInfo("Unable to publish in failed or suspended state", 400, 40000)); default: - connectionManager.send(msg, queueMessages, Listeners.fromCompletionListener(listener)); + connectionManager.send(msg, queueMessages, listener); } } diff --git a/lib/src/main/java/io/ably/lib/realtime/CompletionListener.java b/lib/src/main/java/io/ably/lib/realtime/CompletionListener.java index 7a7205e2f..c2c9c8eb5 100644 --- a/lib/src/main/java/io/ably/lib/realtime/CompletionListener.java +++ b/lib/src/main/java/io/ably/lib/realtime/CompletionListener.java @@ -43,14 +43,14 @@ public void onError(ErrorInfo reason) { } } - class ToCallback implements Callback { + class ToCallback implements Callback { private CompletionListener listener; public ToCallback(CompletionListener listener) { this.listener = listener; } @Override - public void onSuccess(Void v) { + public void onSuccess(T v) { listener.onSuccess(); } diff --git a/lib/src/main/java/io/ably/lib/rest/ChannelBase.java b/lib/src/main/java/io/ably/lib/rest/ChannelBase.java index 11e46751f..b6dd93655 100644 --- a/lib/src/main/java/io/ably/lib/rest/ChannelBase.java +++ b/lib/src/main/java/io/ably/lib/rest/ChannelBase.java @@ -17,6 +17,7 @@ import io.ably.lib.types.Param; import io.ably.lib.types.PresenceMessage; import io.ably.lib.types.PresenceSerializer; +import io.ably.lib.types.PublishResult; import io.ably.lib.types.UpdateDeleteResult; import io.ably.lib.util.Crypto; @@ -66,6 +67,23 @@ void publish(Http http, String name, Object data) throws AblyException { publishImpl(http, name, data).sync(); } + /** + * Publish a message on this channel using the REST API and return the result. + * Since the REST API is stateless, this request is made independently + * of any other request on this or any other channel. + * @param name the event name + * @param data the message payload; + * @return A {@link PublishResult} containing the message serial(s) + * @throws AblyException + */ + public PublishResult publishWithResult(String name, Object data) throws AblyException { + return publishWithResult(ably.http, name, data); + } + + PublishResult publishWithResult(Http http, String name, Object data) throws AblyException { + return publishImpl(http, name, data).sync(); + } + /** * Publish a message on this channel using the REST API. * Since the REST API is stateless, this request is made independently @@ -76,16 +94,37 @@ void publish(Http http, String name, Object data) throws AblyException { * @param listener a listener to be notified of the outcome of this message. *

* This listener is invoked on a background thread. + * @deprecated Use {@link #publishAsync(String, Object, Callback)} instead. */ + @Deprecated public void publishAsync(String name, Object data, CompletionListener listener) { publishAsync(ably.http, name, data, listener); } void publishAsync(Http http, String name, Object data, CompletionListener listener) { - publishImpl(http, name, data).async(new CompletionListener.ToCallback(listener)); + publishImpl(http, name, data).async(new CompletionListener.ToCallback<>(listener)); } - private Http.Request publishImpl(Http http, String name, Object data) { + /** + * Asynchronously publish a message on this channel using the REST API. + * Since the REST API is stateless, this request is made independently + * of any other request on this or any other channel. + * + * @param name the event name + * @param data the message payload; + * @param callback a callback to be notified of the outcome of this message with the {@link PublishResult}. + *

+ * This callback is invoked on a background thread. + */ + public void publishAsync(String name, Object data, Callback callback) { + publishAsync(ably.http, name, data, callback); + } + + void publishAsync(Http http, String name, Object data, Callback callback) { + publishImpl(http, name, data).async(callback); + } + + private Http.Request publishImpl(Http http, String name, Object data) { return publishImpl(http, new Message[] {new Message(name, data)}); } @@ -105,26 +144,52 @@ void publish(Http http, final Message[] messages) throws AblyException { publishImpl(http, messages).sync(); } + /** + * Publish an array of messages on this channel. When there are + * multiple messages to be sent, it is more efficient to use this + * method to publish them in a single request, as compared with + * publishing via multiple independent requests. + * @param messages array of messages to publish. + * @throws AblyException + */ + public PublishResult publishWithResult(final Message[] messages) throws AblyException { + return publishImpl(ably.http, messages).sync(); + } + /** * Asynchronously publish an array of messages on this channel * * @param messages the message * @param listener a listener to be notified of the outcome of this message. + * @deprecated Use {@link #publishAsync(Message[], Callback)} instead. *

* This listener is invoked on a background thread. */ + @Deprecated public void publishAsync(final Message[] messages, final CompletionListener listener) { publishAsync(ably.http, messages, listener); } void publishAsync(Http http, final Message[] messages, final CompletionListener listener) { - publishImpl(http, messages).async(new CompletionListener.ToCallback(listener)); + publishImpl(http, messages).async(new CompletionListener.ToCallback<>(listener)); } - private Http.Request publishImpl(Http http, final Message[] messages) { - return http.request(new Http.Execute() { + /** + * Asynchronously publish an array of messages on this channel + * + * @param messages the message + * @param listener a listener to be notified of the outcome of this message. + *

+ * This listener is invoked on a background thread. + */ + public void publishAsync(final Message[] messages, final Callback listener) { + publishImpl(ably.http, messages).async(listener); + } + + private Http.Request publishImpl(Http http, final Message[] messages) { + return http.request(new Http.Execute() { @Override - public void execute(HttpScheduler http, final Callback callback) throws AblyException { + public void execute(HttpScheduler http, final Callback callback) throws AblyException { /* handle message ids */ boolean hasClientSuppliedId = false; for(Message message : messages) { @@ -145,7 +210,15 @@ public void execute(HttpScheduler http, final Callback callback) throws Ab HttpCore.RequestBody requestBody = ably.options.useBinaryProtocol ? MessageSerializer.asMsgpackRequest(messages) : MessageSerializer.asJsonRequest(messages); final Param[] params = ably.options.addRequestIds ? Param.array(Crypto.generateRandomRequestId()) : null; // RSC7c - http.post(basePath + "/messages", HttpUtils.defaultAcceptHeaders(ably.options.useBinaryProtocol), params, requestBody, null, true, callback); + // Create ResponseHandler from BodyHandler + HttpCore.BodyHandler bodyHandler = PublishResult.getBodyHandler(); + HttpCore.ResponseHandler responseHandler = (response, error) -> { + if (error != null) throw AblyException.fromErrorInfo(error); + String[] serials = bodyHandler.handleResponseBody(response.contentType, response.body); + return new PublishResult(serials); + }; + + http.post(basePath + "/messages", HttpUtils.defaultAcceptHeaders(ably.options.useBinaryProtocol), params, requestBody, responseHandler, true, callback); } }); } diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelMessageEditTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelMessageEditTest.java index 342a051db..9d330365f 100644 --- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelMessageEditTest.java +++ b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelMessageEditTest.java @@ -12,6 +12,7 @@ import io.ably.lib.types.MessageOperation; import io.ably.lib.types.PaginatedResult; import io.ably.lib.types.Param; +import io.ably.lib.types.PublishResult; import io.ably.lib.types.UpdateDeleteResult; import io.ably.lib.util.Crypto; import org.junit.After; @@ -58,25 +59,23 @@ public void getMessage_retrieveBySerial() throws Exception { String channelName = "mutable:get_message_" + UUID.randomUUID() + "_" + testParams.name; Channel channel = ably.channels.get(channelName); - // Publish a message - channel.publish("test_event", "Test message data"); + Helpers.AsyncWaiter publishResultAsyncWaiter = new Helpers.AsyncWaiter<>(); - // Get the message from history to obtain its serial - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); + // Publish a message + channel.publish("test_event", "Test message data", publishResultAsyncWaiter); - Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); + publishResultAsyncWaiter.waitFor(); + PublishResult publishResult = publishResultAsyncWaiter.result; + assertEquals("Expected to have one serial", 1, publishResult.serials.length); // Retrieve the message by serial - Message retrievedMessage = waitForUpdatedMessageAppear(channel, publishedMessage.serial); + Message retrievedMessage = waitForUpdatedMessageAppear(channel, publishResult.serials[0]); // Verify the retrieved message assertNotNull("Expected non-null retrieved message", retrievedMessage); - assertEquals("Expected same message name", publishedMessage.name, retrievedMessage.name); - assertEquals("Expected same message data", publishedMessage.data, retrievedMessage.data); - assertEquals("Expected same serial", publishedMessage.serial, retrievedMessage.serial); + assertEquals("Expected same message name", "test_event", retrievedMessage.name); + assertEquals("Expected same message data", "Test message data", retrievedMessage.data); + assertEquals("Expected same serial", publishResult.serials[0], retrievedMessage.serial); } /** @@ -87,20 +86,19 @@ public void updateMessage_updateData() throws Exception { String channelName = "mutable:update_message_" + UUID.randomUUID() + "_" + testParams.name; Channel channel = ably.channels.get(channelName); + Helpers.AsyncWaiter publishResultAsyncWaiter = new Helpers.AsyncWaiter<>(); + // Publish a message - channel.publish("test_event", "Original message data"); + channel.publish("test_event", "Original message data", publishResultAsyncWaiter); // Get the message from history to obtain its serial - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); - - Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); + publishResultAsyncWaiter.waitFor(); + PublishResult publishResult = publishResultAsyncWaiter.result; + assertEquals("Expected to have one serial", 1, publishResult.serials.length); // Update the message Message updateMessage = new Message(); - updateMessage.serial = publishedMessage.serial; + updateMessage.serial = publishResult.serials[0]; updateMessage.data = "Updated message data"; updateMessage.name = "updated_event"; @@ -111,7 +109,7 @@ public void updateMessage_updateData() throws Exception { waiter.waitFor(); // Retrieve the updated message - Message updatedMessage = waitForUpdatedMessageAppear(channel, publishedMessage.serial); + Message updatedMessage = waitForUpdatedMessageAppear(channel, publishResult.serials[0]); // Verify the message was updated assertNotNull("Expected non-null updated message", updatedMessage); @@ -130,27 +128,26 @@ public void updateMessage_updateEncodedData() throws Exception { ChannelOptions channelOptions = ChannelOptions.withCipherKey(Crypto.generateRandomKey()); Channel channel = ably.channels.get(channelName, channelOptions); + Helpers.AsyncWaiter publishResultAsyncWaiter = new Helpers.AsyncWaiter<>(); + // Publish a message channel.publish("test_event", "Original message data"); // Get the message from history to obtain its serial - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); - - Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); + publishResultAsyncWaiter.waitFor(); + PublishResult publishResult = publishResultAsyncWaiter.result; + assertEquals("Expected to have one serial", 1, publishResult.serials.length); // Update the message Message updateMessage = new Message(); - updateMessage.serial = publishedMessage.serial; + updateMessage.serial = publishResult.serials[0]; updateMessage.data = "Updated message data"; updateMessage.name = "updated_event"; channel.updateMessage(updateMessage); // Retrieve the updated message - Message updatedMessage = waitForUpdatedMessageAppear(channel, publishedMessage.serial); + Message updatedMessage = waitForUpdatedMessageAppear(channel, publishResult.serials[0]); // Verify the message was updated assertNotNull("Expected non-null updated message", updatedMessage); @@ -167,26 +164,25 @@ public void deleteMessage_softDelete() throws Exception { String channelName = "mutable:delete_message_" + UUID.randomUUID() + "_" + testParams.name; Channel channel = ably.channels.get(channelName); + Helpers.AsyncWaiter publishResultAsyncWaiter = new Helpers.AsyncWaiter<>(); + // Publish a message - channel.publish("test_event", "Message to be deleted"); + channel.publish("test_event", "Message to be deleted", publishResultAsyncWaiter); // Get the message from history - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); - - Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); + publishResultAsyncWaiter.waitFor(); + PublishResult publishResult = publishResultAsyncWaiter.result; + assertEquals("Expected to have one serial", 1, publishResult.serials.length); // Delete the message Message deleteMessage = new Message(); - deleteMessage.serial = publishedMessage.serial; + deleteMessage.serial = publishResult.serials[0]; deleteMessage.data = "Message deleted"; channel.deleteMessage(deleteMessage); // Retrieve the deleted message - Message deletedMessage = waitForDeletedMessageAppear(channel, publishedMessage.serial); + Message deletedMessage = waitForDeletedMessageAppear(channel, publishResult.serials[0]); // Verify the message was soft deleted assertNotNull("Expected non-null deleted message", deletedMessage); @@ -201,25 +197,24 @@ public void getMessageVersions_retrieveHistory() throws Exception { String channelName = "mutable:message_versions_" + UUID.randomUUID() + "_" + testParams.name; Channel channel = ably.channels.get(channelName); + Helpers.AsyncWaiter publishResultAsyncWaiter = new Helpers.AsyncWaiter<>(); + // Publish a message - channel.publish("test_event", "Original data"); + channel.publish("test_event", "Original data", publishResultAsyncWaiter); // Get the message from history - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); - - Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); + publishResultAsyncWaiter.waitFor(); + PublishResult publishResult = publishResultAsyncWaiter.result; + assertEquals("Expected to have one serial", 1, publishResult.serials.length); // Update the message to create version history Message updateMessage1 = new Message(); - updateMessage1.serial = publishedMessage.serial; + updateMessage1.serial = publishResult.serials[0]; updateMessage1.data = "First update"; channel.updateMessage(updateMessage1); Message updateMessage2 = new Message(); - updateMessage2.serial = publishedMessage.serial; + updateMessage2.serial = publishResult.serials[0]; updateMessage2.data = "Second update"; MessageOperation messageOperation = new MessageOperation(); messageOperation.description = "description"; @@ -228,7 +223,7 @@ public void getMessageVersions_retrieveHistory() throws Exception { channel.updateMessage(updateMessage2, messageOperation); // Retrieve version history - PaginatedResult versions = waitForMessageAppearInVersionHistory(channel, publishedMessage.serial, null, msgs -> + PaginatedResult versions = waitForMessageAppearInVersionHistory(channel, publishResult.serials[0], null, msgs -> msgs.length >= 3 ); @@ -307,13 +302,16 @@ public void completeWorkflow_publishUpdateVersionsDelete() throws Exception { String channelName = "mutable:complete_workflow_" + UUID.randomUUID() + "_" + testParams.name; Channel channel = ably.channels.get(channelName); + Helpers.AsyncWaiter publishResultAsyncWaiter = new Helpers.AsyncWaiter<>(); + // 1. Publish a message - channel.publish("workflow_event", "Initial data"); + channel.publish("workflow_event", "Initial data", publishResultAsyncWaiter); // Get the published message - PaginatedResult history = waitForMessageAppearInHistory(channel); - Message publishedMessage = history.items()[0]; - String serial = publishedMessage.serial; + publishResultAsyncWaiter.waitFor(); + PublishResult publishResult = publishResultAsyncWaiter.result; + assertEquals("Expected to have one serial", 1, publishResult.serials.length); + String serial = publishResult.serials[0]; // 2. Update the message Message updateMessage = new Message(); @@ -359,13 +357,17 @@ public void appendMessage_checkUpdatedData() throws Exception { String channelName = "mutable:message_append_" + UUID.randomUUID() + "_" + testParams.name; Channel channel = ably.channels.get(channelName); + Helpers.AsyncWaiter publishResultAsyncWaiter = new Helpers.AsyncWaiter<>(); + // 1. Publish a message - channel.publish("append_event", "Initial data"); + channel.publish("append_event", "Initial data", publishResultAsyncWaiter); // Get the published message - PaginatedResult history = waitForMessageAppearInHistory(channel); - Message publishedMessage = history.items()[0]; - String serial = publishedMessage.serial; + publishResultAsyncWaiter.waitFor(); + PublishResult publishResult = publishResultAsyncWaiter.result; + assertEquals("Expected to have one serial", 1, publishResult.serials.length); + + String serial = publishResult.serials[0]; Helpers.AsyncWaiter appendWaiter = new Helpers.AsyncWaiter<>(); @@ -402,15 +404,6 @@ private PaginatedResult waitForMessageAppearInVersionHistory(Channel ch } } - private PaginatedResult waitForMessageAppearInHistory(Channel channel) throws Exception { - long timeout = System.currentTimeMillis() + 5_000; - while (true) { - PaginatedResult history = channel.history(null); - if (history.items().length > 0 || System.currentTimeMillis() > timeout) return history; - Thread.sleep(200); - } - } - private Message waitForUpdatedMessageAppear(Channel channel, String serial) throws Exception { long timeout = System.currentTimeMillis() + 5_000; while (true) { diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java index 828f3cbaa..a64f96daa 100644 --- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java +++ b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java @@ -692,7 +692,7 @@ public MessageListener setMessageStack(List messageStack) { channel2.subscribe(listener); /* Start emitting channel with ably client 1 (emitter) */ - channel1.publish(messages, null); + channel1.publish(messages); /* Wait until receiver client (ably2) observes {@code } * is emitted from emitter client (ably1) @@ -781,9 +781,9 @@ public MessageListener setMessageStack(List messageStack) { channel2.subscribe(messageNames, listener); /* Start emitting channel with ably client 1 (emitter) */ - channel1.publish("nonTrackedMessageName", "This message should be ignore by second client (ably2).", null); - channel1.publish(messages, null); - channel1.publish("nonTrackedMessageName", "This message should be ignore by second client (ably2).", null); + channel1.publish("nonTrackedMessageName", "This message should be ignore by second client (ably2)."); + channel1.publish(messages); + channel1.publish("nonTrackedMessageName", "This message should be ignore by second client (ably2)."); /* Wait until receiver client (ably2) observes {@code Message} * on subscribed channel (channel2) emitted by emitter client (ably1) @@ -866,9 +866,9 @@ public MessageListener setMessageStack(List messageStack) { channel2.subscribe(messageName, listener); /* Start emitting channel with ably client 1 (emitter) */ - channel1.publish("nonTrackedMessageName", "This message should be ignore by second client (ably2).", null); - channel1.publish(messages, null); - channel1.publish("nonTrackedMessageName", "This message should be ignore by second client (ably2).", null); + channel1.publish("nonTrackedMessageName", "This message should be ignore by second client (ably2)."); + channel1.publish(messages); + channel1.publish("nonTrackedMessageName", "This message should be ignore by second client (ably2)."); /* Wait until receiver client (ably2) observes {@code Message} * on subscribed channel (channel2) emitted by emitter client (ably1) diff --git a/lib/src/test/java/io/ably/lib/test/rest/RestChannelMessageEditTest.java b/lib/src/test/java/io/ably/lib/test/rest/RestChannelMessageEditTest.java index 4a69037b8..651900d83 100644 --- a/lib/src/test/java/io/ably/lib/test/rest/RestChannelMessageEditTest.java +++ b/lib/src/test/java/io/ably/lib/test/rest/RestChannelMessageEditTest.java @@ -14,6 +14,7 @@ import io.ably.lib.types.MessageOperation; import io.ably.lib.types.PaginatedResult; import io.ably.lib.types.Param; +import io.ably.lib.types.PublishResult; import io.ably.lib.types.UpdateDeleteResult; import io.ably.lib.util.Crypto; import org.junit.Before; @@ -58,24 +59,17 @@ public void getMessage_retrieveBySerial() throws Exception { Channel channel = ably.channels.get(channelName); // Publish a message - channel.publish("test_event", "Test message data"); - - // Get the message from history to obtain its serial - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); - - Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); + PublishResult publishResult = channel.publishWithResult("test_event", "Test message data"); + assertNotNull("Expected message to have a serial", publishResult.serials[0]); // Retrieve the message by serial - Message retrievedMessage = waitForUpdatedMessageAppear(channel, publishedMessage.serial); + Message retrievedMessage = waitForUpdatedMessageAppear(channel, publishResult.serials[0]); // Verify the retrieved message assertNotNull("Expected non-null retrieved message", retrievedMessage); - assertEquals("Expected same message name", publishedMessage.name, retrievedMessage.name); - assertEquals("Expected same message data", publishedMessage.data, retrievedMessage.data); - assertEquals("Expected same serial", publishedMessage.serial, retrievedMessage.serial); + assertEquals("Expected same message name", "test_event", retrievedMessage.name); + assertEquals("Expected same message data", "Test message data", retrievedMessage.data); + assertEquals("Expected same serial", publishResult.serials[0], retrievedMessage.serial); } /** @@ -89,26 +83,19 @@ public void updateMessage_updateData() throws Exception { Channel channel = ably.channels.get(channelName); // Publish a message - channel.publish("test_event", "Original message data"); - - // Get the message from history to obtain its serial - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); - - Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); + PublishResult publishResult = channel.publishWithResult("test_event", "Original message data"); + assertNotNull("Expected message to have a serial", publishResult.serials[0]); // Update the message Message updateMessage = new Message(); - updateMessage.serial = publishedMessage.serial; + updateMessage.serial = publishResult.serials[0]; updateMessage.data = "Updated message data"; updateMessage.name = "updated_event"; UpdateDeleteResult result = channel.updateMessage(updateMessage); // Retrieve the updated message - Message updatedMessage = waitForUpdatedMessageAppear(channel, publishedMessage.serial); + Message updatedMessage = waitForUpdatedMessageAppear(channel, publishResult.serials[0]); // Verify the message was updated assertNotNull("Expected non-null updated message", updatedMessage); @@ -130,26 +117,19 @@ public void updateMessage_updateEncodedData() throws Exception { Channel channel = ably.channels.get(channelName, channelOptions); // Publish a message - channel.publish("test_event", "Original message data"); - - // Get the message from history to obtain its serial - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); - - Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); + PublishResult publishResult = channel.publishWithResult("test_event", "Original message data"); + assertNotNull("Expected message to have a serial", publishResult.serials[0]); // Update the message Message updateMessage = new Message(); - updateMessage.serial = publishedMessage.serial; + updateMessage.serial = publishResult.serials[0]; updateMessage.data = "Updated message data"; updateMessage.name = "updated_event"; UpdateDeleteResult result = channel.updateMessage(updateMessage); // Retrieve the updated message - Message updatedMessage = waitForUpdatedMessageAppear(channel, publishedMessage.serial); + Message updatedMessage = waitForUpdatedMessageAppear(channel, publishResult.serials[0]); // Verify the message was updated assertNotNull("Expected non-null updated message", updatedMessage); @@ -169,20 +149,17 @@ public void updateMessage_async() throws Exception { String channelName = "mutable:update_message_async_" + UUID.randomUUID() + "_" + testParams.name; Channel channel = ably.channels.get(channelName); - // Publish a message - channel.publish("test_event", "Original message data"); - - // Get the message from history - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); + Helpers.AsyncWaiter publishWaiter = new Helpers.AsyncWaiter<>(); - final Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); + // Publish a message + channel.publishAsync("test_event", "Original message data", publishWaiter); + publishWaiter.waitFor(); + PublishResult publishResult = publishWaiter.result; + assertNotNull("Expected message to have a serial", publishResult.serials[0]); // Update the message using async API Message updateMessage = new Message(); - updateMessage.serial = publishedMessage.serial; + updateMessage.serial = publishResult.serials[0]; updateMessage.data = "Updated message data async"; Helpers.AsyncWaiter updateWaiter = new Helpers.AsyncWaiter<>(); @@ -191,7 +168,7 @@ public void updateMessage_async() throws Exception { updateWaiter.waitFor(); // Retrieve the updated message - Message updatedMessage = waitForUpdatedMessageAppear(channel, publishedMessage.serial); + Message updatedMessage = waitForUpdatedMessageAppear(channel, publishResult.serials[0]); assertNotNull("Expected non-null updated message", updatedMessage); assertEquals("Expected updated message data", "Updated message data async", updatedMessage.data); assertEquals(updateWaiter.result.versionSerial, updatedMessage.version.serial); @@ -208,25 +185,18 @@ public void deleteMessage_softDelete() throws Exception { Channel channel = ably.channels.get(channelName); // Publish a message - channel.publish("test_event", "Message to be deleted"); - - // Get the message from history - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); - - Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); + PublishResult publishResult = channel.publishWithResult("test_event", "Message to be deleted"); + assertNotNull("Expected message to have a serial", publishResult.serials[0]); // Delete the message Message deleteMessage = new Message(); - deleteMessage.serial = publishedMessage.serial; + deleteMessage.serial = publishResult.serials[0]; deleteMessage.data = "Message deleted"; UpdateDeleteResult result = channel.deleteMessage(deleteMessage); // Retrieve the deleted message - Message deletedMessage = waitForDeletedMessageAppear(channel, publishedMessage.serial); + Message deletedMessage = waitForDeletedMessageAppear(channel, publishResult.serials[0]); // Verify the message was soft deleted assertNotNull("Expected non-null deleted message", deletedMessage); @@ -245,25 +215,18 @@ public void appendMessage_checkUpdatedData() throws Exception { Channel channel = ably.channels.get(channelName); // Publish a message - channel.publish("test_event", "Initial message"); - - // Get the message from history - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); - - Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); + PublishResult publishResult = channel.publishWithResult("test_event", "Initial message"); + assertNotNull("Expected message to have a serial", publishResult.serials[0]); // Append the message Message appendMessage = new Message(); - appendMessage.serial = publishedMessage.serial; + appendMessage.serial = publishResult.serials[0]; appendMessage.data = "Message append"; UpdateDeleteResult result = channel.appendMessage(appendMessage); // Retrieve the updated message - Message updatedMessage = waitForUpdatedMessageAppear(channel, publishedMessage.serial); + Message updatedMessage = waitForUpdatedMessageAppear(channel, publishResult.serials[0]); // Verify the message was appended assertNotNull("Expected non-null append message", updatedMessage); @@ -283,19 +246,12 @@ public void deleteMessage_async() throws Exception { Channel channel = ably.channels.get(channelName); // Publish a message - channel.publish("test_event", "Message to be deleted async"); - - // Get the message from history - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); - - final Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); + PublishResult publishResult = channel.publishWithResult("test_event", "Message to be deleted async"); + assertNotNull("Expected message to have a serial", publishResult.serials[0]); // Delete the message using async API Message deleteMessage = new Message(); - deleteMessage.serial = publishedMessage.serial; + deleteMessage.serial = publishResult.serials[0]; deleteMessage.data = "Message deleted async"; Helpers.AsyncWaiter deleteWaiter = new Helpers.AsyncWaiter<>(); @@ -304,7 +260,7 @@ public void deleteMessage_async() throws Exception { deleteWaiter.waitFor(); // Retrieve the deleted message - Message deletedMessage = waitForDeletedMessageAppear(channel, publishedMessage.serial); + Message deletedMessage = waitForDeletedMessageAppear(channel, publishResult.serials[0]); assertNotNull("Expected non-null deleted message", deletedMessage); assertEquals("Expected action to be MESSAGE_DELETE", MessageAction.MESSAGE_DELETE, deletedMessage.action); assertEquals(deleteWaiter.result.versionSerial, deletedMessage.version.serial); @@ -321,24 +277,17 @@ public void getMessageVersions_retrieveHistory() throws Exception { Channel channel = ably.channels.get(channelName); // Publish a message - channel.publish("test_event", "Original data"); - - // Get the message from history - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); + PublishResult publishResult = channel.publishWithResult("test_event", "Original data"); - Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); // Update the message to create version history Message updateMessage1 = new Message(); - updateMessage1.serial = publishedMessage.serial; + updateMessage1.serial = publishResult.serials[0]; updateMessage1.data = "First update"; channel.updateMessage(updateMessage1); Message updateMessage2 = new Message(); - updateMessage2.serial = publishedMessage.serial; + updateMessage2.serial = publishResult.serials[0]; updateMessage2.data = "Second update"; MessageOperation messageOperation = new MessageOperation(); messageOperation.description = "description"; @@ -347,7 +296,7 @@ public void getMessageVersions_retrieveHistory() throws Exception { channel.updateMessage(updateMessage2, messageOperation); // Retrieve version history - PaginatedResult versions = waitForMessageAppearInVersionHistory(channel, publishedMessage.serial, null, msgs -> + PaginatedResult versions = waitForMessageAppearInVersionHistory(channel, publishResult.serials[0], null, msgs -> msgs.length >= 3 ); @@ -372,15 +321,28 @@ public void getMessageVersions_async() throws Exception { Channel channel = ably.channels.get(channelName); // Publish a message - channel.publish("test_event", "Original data"); + PublishResult publishResult = channel.publishWithResult("test_event", "Original data"); + + // Update the message to create version history + Message updateMessage1 = new Message(); + updateMessage1.serial = publishResult.serials[0]; + updateMessage1.data = "Update"; + MessageOperation messageOperation1 = new MessageOperation(); + messageOperation1.description = "description"; + channel.updateMessage(updateMessage1, messageOperation1); + + // Retrieve version history + PaginatedResult versions = waitForMessageAppearInVersionHistory(channel, publishResult.serials[0], null, msgs -> + msgs.length >= 2 + ); - // Get the message from history - PaginatedResult history = waitForMessageAppearInHistory(channel); - assertNotNull("Expected non-null history", history); - assertEquals(1, history.items().length); + // Verify version history + assertNotNull("Expected non-null versions", versions); + assertTrue("Expected at least 2 versions (original + 2 updates)", versions.items().length >= 2); - final Message publishedMessage = history.items()[0]; - assertNotNull("Expected message to have a serial", publishedMessage.serial); + Message latestVersion = versions.items()[versions.items().length - 1]; + assertEquals("Expected latest version to have second update data", "Update", latestVersion.data); + assertEquals("description", latestVersion.version.description); } /** @@ -451,12 +413,10 @@ public void completeWorkflow_publishUpdateVersionsDelete() throws Exception { Channel channel = ably.channels.get(channelName); // 1. Publish a message - channel.publish("workflow_event", "Initial data"); + PublishResult publishResult = channel.publishWithResult("workflow_event", "Initial data"); // Get the published message - PaginatedResult history = waitForMessageAppearInHistory(channel); - Message publishedMessage = history.items()[0]; - String serial = publishedMessage.serial; + String serial = publishResult.serials[0]; // 2. Update the message Message updateMessage = new Message(); @@ -509,20 +469,30 @@ private PaginatedResult waitForMessageAppearInHistory(Channel channel) private Message waitForUpdatedMessageAppear(Channel channel, String serial) throws Exception { long timeout = System.currentTimeMillis() + 5_000; while (true) { - Message message = channel.getMessage(serial); - if ((message != null && message.action == MessageAction.MESSAGE_UPDATE) || System.currentTimeMillis() > timeout) - return message; - Thread.sleep(200); + try { + Message message = channel.getMessage(serial); + if ((message != null && message.action == MessageAction.MESSAGE_UPDATE) || System.currentTimeMillis() > timeout) + return message; + Thread.sleep(200); + } catch (AblyException e) { + // skip not found errors + if (e.errorInfo.statusCode != 404) throw e; + } } } private Message waitForDeletedMessageAppear(Channel channel, String serial) throws Exception { long timeout = System.currentTimeMillis() + 5_000; while (true) { - Message message = channel.getMessage(serial); - if ((message != null && message.action == MessageAction.MESSAGE_DELETE) || System.currentTimeMillis() > timeout) - return message; - Thread.sleep(200); + try { + Message message = channel.getMessage(serial); + if ((message != null && message.action == MessageAction.MESSAGE_DELETE) || System.currentTimeMillis() > timeout) + return message; + Thread.sleep(200); + } catch (AblyException e) { + // skip not found errors + if (e.errorInfo.statusCode != 404) throw e; + } } } }