Skip to content

[AIT-98] feat: realtime edits and deletes #1183

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ttypic wants to merge 1 commit into
base: AIT-99/message-appends
Choose a base branch
from
Open

[AIT-98] feat: realtime edits and deletes #1183

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
168 changes: 99 additions & 69 deletions lib/src/main/java/io/ably/lib/realtime/ChannelBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import io.ably.lib.types.DeltaExtras;
import io.ably.lib.types.ErrorInfo;
import io.ably.lib.types.Message;
import io.ably.lib.types.MessageAction;
import io.ably.lib.types.MessageAnnotations;
import io.ably.lib.types.MessageDecodeException;
import io.ably.lib.types.MessageOperation;
Expand All @@ -46,6 +47,7 @@
import io.ably.lib.types.UpdateDeleteResult;
import io.ably.lib.util.CollectionUtils;
import io.ably.lib.util.EventEmitter;
import io.ably.lib.util.Listeners;
import io.ably.lib.util.Log;
import io.ably.lib.util.ReconnectionStrategy;
import io.ably.lib.util.StringUtils;
Expand Down Expand Up @@ -1123,7 +1125,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, listener);
connectionManager.send(msg, queueMessages, Listeners.fromCompletionListener(listener));
}
}

Expand Down Expand Up @@ -1206,103 +1208,89 @@ public void getMessageAsync(String serial, Callback<Message> callback) {
}

/**
* Updates an existing message using patch semantics.
* <p>
* Non-null fields in the provided message (name, data, extras) will replace the corresponding
* fields in the existing message, while null fields will be left unchanged.
* Asynchronously updates an existing message.
*
* @param message A {@link Message} object containing the fields to update and the serial identifier.
* Only non-null fields will be applied to the existing message.
* @param operation operation metadata such as clientId, description, or metadata in the version field
* @throws AblyException If the update operation fails.
* @return A {@link UpdateDeleteResult} containing the updated message version serial.
* <p>
* This callback is invoked on a background thread.
*/
public UpdateDeleteResult updateMessage(Message message, MessageOperation operation) throws AblyException {
return messageEditsMixin.updateMessage(ably.http, message, operation);
public void updateMessage(Message message) throws AblyException {
updateMessage(message, null, null);
}

/**
* Updates an existing message using patch semantics.
* <p>
* Non-null fields in the provided message (name, data, extras) will replace the corresponding
* fields in the existing message, while null fields will be left unchanged.
* Asynchronously updates an existing message.
*
* @param message A {@link Message} object containing the fields to update and the serial identifier.
* Only non-null fields will be applied to the existing message.
* @throws AblyException If the update operation fails.
* @return A {@link UpdateDeleteResult} containing the updated message version serial.
* @param operation operation metadata such as clientId, description, or metadata in the version field
* <p>
* This callback is invoked on a background thread.
*/
public UpdateDeleteResult updateMessage(Message message) throws AblyException {
return updateMessage(message, null);
public void updateMessage(Message message, MessageOperation operation) throws AblyException {
updateMessage(message, operation, null);
}

/**
* Asynchronously updates an existing message.
*
* @param message A {@link Message} object containing the fields to update and the serial identifier.
* @param operation operation metadata such as clientId, description, or metadata in the version field
* @param callback A callback to be notified of the outcome of this operation.
* @param listener A callback to be notified of the outcome of this operation.
* <p>
* This callback is invoked on a background thread.
*/
public void updateMessageAsync(Message message, MessageOperation operation, Callback<UpdateDeleteResult> callback) {
messageEditsMixin.updateMessageAsync(ably.http, message, operation, callback);
public void updateMessage(Message message, MessageOperation operation, Callback<UpdateDeleteResult> listener) throws AblyException {
Log.v(TAG, "updateMessage(Message); channel = " + this.name + "; serial = " + message.serial);
updateDeleteImpl(message, operation, MessageAction.MESSAGE_UPDATE, listener);
}

/**
* Asynchronously updates an existing message.
*
* @param message A {@link Message} object containing the fields to update and the serial identifier.
* @param callback A callback to be notified of the outcome of this operation.
* @param listener A callback to be notified of the outcome of this operation.
* <p>
* This callback is invoked on a background thread.
*/
public void updateMessageAsync(Message message, Callback<UpdateDeleteResult> callback) {
updateMessageAsync(message, null, callback);
public void updateMessage(Message message, Callback<UpdateDeleteResult> listener) throws AblyException {
updateMessage(message, null, listener);
}

/**
* Marks a message as deleted.
* <p>
* This operation does not remove the message from history; it marks it as deleted
* while preserving the full message history. The deleted message can still be
* retrieved and will have its action set to MESSAGE_DELETE.
* Asynchronously marks a message as deleted.
*
* @param message A {@link Message} message containing the serial identifier.
* @param operation operation metadata such as clientId, description, or metadata in the version field
* @throws AblyException If the delete operation fails.
* @return A {@link UpdateDeleteResult} containing the deleted message version serial.
* @param message A {@link Message} object containing the serial identifier and operation metadata.
* <p>
* This callback is invoked on a background thread.
*/
public UpdateDeleteResult deleteMessage(Message message, MessageOperation operation) throws AblyException {
return messageEditsMixin.deleteMessage(ably.http, message, operation);
public void deleteMessage(Message message) throws AblyException {
deleteMessage(message, null, null);
}

/**
* Marks a message as deleted.
* <p>
* This operation does not remove the message from history; it marks it as deleted
* while preserving the full message history. The deleted message can still be
* retrieved and will have its action set to MESSAGE_DELETE.
* Asynchronously marks a message as deleted.
*
* @param message A {@link Message} message containing the serial identifier.
* @throws AblyException If the delete operation fails.
* @return A {@link UpdateDeleteResult} containing the deleted message version serial.
* @param message A {@link Message} object containing the serial identifier and operation metadata.
* @param operation operation metadata such as clientId, description, or metadata in the version field
* <p>
* This callback is invoked on a background thread.
*/
public UpdateDeleteResult deleteMessage(Message message) throws AblyException {
return deleteMessage(message, null);
public void deleteMessage(Message message, MessageOperation operation) throws AblyException {
deleteMessage(message, operation, null);
}

/**
* Asynchronously marks a message as deleted.
*
* @param message A {@link Message} object containing the serial identifier and operation metadata.
* @param operation operation metadata such as clientId, description, or metadata in the version field
* @param callback A callback to be notified of the outcome of this operation.
* @param listener A callback to be notified of the outcome of this operation.
* <p>
* This callback is invoked on a background thread.
*/
public void deleteMessageAsync(Message message, MessageOperation operation, Callback<UpdateDeleteResult> callback) {
messageEditsMixin.deleteMessageAsync(ably.http, message, operation, callback);
public void deleteMessage(Message message, MessageOperation operation, Callback<UpdateDeleteResult> listener) throws AblyException {
Log.v(TAG, "deleteMessage(Message); channel = " + this.name + "; serial = " + message.serial);
updateDeleteImpl(message, operation, MessageAction.MESSAGE_DELETE, listener);
}

/**
Expand All @@ -1313,44 +1301,45 @@ public void deleteMessageAsync(Message message, MessageOperation operation, Call
* <p>
* This callback is invoked on a background thread.
*/
public void deleteMessageAsync(Message message, Callback<UpdateDeleteResult> callback) {
deleteMessageAsync(message, null, callback);
public void deleteMessage(Message message, Callback<UpdateDeleteResult> callback) throws AblyException {
deleteMessage(message, null, callback);
}

/**
* Appends message text to the end of the message.
* Asynchronously appends message text to the end of the message.
*
* @param message A {@link Message} object containing the serial identifier and data to append.
* @param operation operation details such as clientId, description, or metadata
* @return A {@link UpdateDeleteResult} containing the updated message version serial.
* @throws AblyException If the append operation fails.
* <p>
* This callback is invoked on a background thread.
*/
public UpdateDeleteResult appendMessage(Message message, MessageOperation operation) throws AblyException {
return messageEditsMixin.appendMessage(ably.http, message, operation);
public void appendMessage(Message message) throws AblyException {
appendMessage(message, null, null);
}

/**
* Appends message text to the end of the message.
* Asynchronously appends message text to the end of the message.
*
* @param message A {@link Message} object containing the serial identifier and data to append.
* @return A {@link UpdateDeleteResult} containing the updated message version serial.
* @throws AblyException If the append operation fails.
* @param operation operation details such as clientId, description, or metadata
* <p>
* This callback is invoked on a background thread.
*/
public UpdateDeleteResult appendMessage(Message message) throws AblyException {
return appendMessage(message, null);
public void appendMessage(Message message, MessageOperation operation) throws AblyException {
appendMessage(message, operation, null);
}

/**
* Asynchronously appends message text to the end of the message.
*
* @param message A {@link Message} object containing the serial identifier and data to append.
* @param operation operation details such as clientId, description, or metadata
* @param callback A callback to be notified of the outcome of this operation.
* @param listener A callback to be notified of the outcome of this operation.
* <p>
* This callback is invoked on a background thread.
*/
public void appendMessageAsync(Message message, MessageOperation operation, Callback<UpdateDeleteResult> callback) {
messageEditsMixin.appendMessageAsync(ably.http, message, operation, callback);
public void appendMessage(Message message, MessageOperation operation, Callback<UpdateDeleteResult> listener) throws AblyException {
Log.v(TAG, "appendMessage(Message); channel = " + this.name + "; serial = " + message.serial);
updateDeleteImpl(message, operation, MessageAction.MESSAGE_APPEND, listener);
}

/**
Expand All @@ -1361,8 +1350,49 @@ public void appendMessageAsync(Message message, MessageOperation operation, Call
* <p>
* This callback is invoked on a background thread.
*/
public void appendMessageAsync(Message message, Callback<UpdateDeleteResult> callback) {
appendMessageAsync(message, null, callback);
public void appendMessage(Message message, Callback<UpdateDeleteResult> callback) throws AblyException {
appendMessage(message, null, callback);
}

private void updateDeleteImpl(
Message message,
MessageOperation operation,
MessageAction action,
Callback<UpdateDeleteResult> listener
) throws AblyException {
if (message.serial == null || message.serial.isEmpty()) {
throw AblyException.fromErrorInfo(new ErrorInfo("Message serial cannot be empty", 400, 40003));
}
ConnectionManager connectionManager = ably.connection.connectionManager;
ConnectionManager.State connectionState = connectionManager.getConnectionState();
boolean queueMessages = ably.options.queueMessages;
if (!connectionManager.isActive() || (connectionState.queueEvents && !queueMessages)) {
throw AblyException.fromErrorInfo(connectionState.defaultErrorInfo);
}
boolean connected = (connectionState.sendEvents);

Message updatedMessage = new Message(message.name, message.data, message.extras);
updatedMessage.action = action;
updatedMessage.version = new MessageVersion();
if (operation != null) {
updatedMessage.version.clientId = operation.clientId;
updatedMessage.version.description = operation.description;
updatedMessage.version.metadata = operation.metadata;
}
Comment on lines +1374 to +1381
Copy link

@coderabbitai coderabbitai bot Jan 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Check the Message class structure and serial field usage
echo "=== Checking Message.java for serial field ==="
find . -name "Message.java" -path "*/types/*" -type f | head -5

Repository: ably/ably-java

Length of output: 155

🏁 Script executed:

#!/bin/bash
# Search for serial field in Message class
rg -n -A 3 -B 1 'serial' lib/src/main/java/io/ably/lib/types/Message.java | head -50

Repository: ably/ably-java

Length of output: 2091

🏁 Script executed:

#!/bin/bash
# Check the test file for how serial is used in update operations
rg -n 'serial' lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelMessageEditTest.java

Repository: ably/ably-java

Length of output: 3966

🏁 Script executed:

#!/bin/bash
# Check the full context of lines 1374-1381 in ChannelBase.java
sed -n '1360,1395p' lib/src/main/java/io/ably/lib/realtime/ChannelBase.java

Repository: ably/ably-java

Length of output: 1766

🏁 Script executed:

#!/bin/bash
# Get more context around the message sending to understand if updatedMessage should be used
sed -n '1358,1405p' lib/src/main/java/io/ably/lib/realtime/ChannelBase.java

Repository: ably/ably-java

Length of output: 2195

🏁 Script executed:

#!/bin/bash
# Check if there are other places where updatedMessage is used after its creation
sed -n '1374,1420p' lib/src/main/java/io/ably/lib/realtime/ChannelBase.java

Repository: ably/ably-java

Length of output: 2174

Critical issue: Wrong message object being sent, and updatedMessage missing the serial field.

The code creates and configures updatedMessage with the action and version metadata (lines 1374-1380), but then sends the original message object instead (line 1400), which lacks this metadata. Additionally, if updatedMessage were to be sent, it would need the serial field copied from the original message for the server to identify which message to update/delete. Either msg.messages = new Message[] { updatedMessage }; should be used, or the entire updatedMessage construction should be removed if the original message is the intended payload.

🤖 Prompt for AI Agents 
In @lib/src/main/java/io/ably/lib/realtime/ChannelBase.java around lines 1374 -
1381, The code in ChannelBase constructs an updatedMessage (setting action and
version fields) but then sends the original message, and updatedMessage also
never copies the original message.serial; fix by sending the updatedMessage and
ensuring its serial is copied from the original message: populate
updatedMessage.serial = message.serial after building updatedMessage, and
replace msg.messages = new Message[] { message } with msg.messages = new
Message[] { updatedMessage } (or remove updatedMessage construction if you
intend to send the original message instead).


try {
ably.auth.checkClientId(message, true, connected);
updatedMessage.encode(options);
} catch (AblyException e) {
if (listener != null) {
listener.onError(e.errorInfo);
}
return;
}

ProtocolMessage msg = new ProtocolMessage(Action.message, this.name);
msg.messages = new Message[] { message };
connectionManager.send(msg, queueMessages, Listeners.toPublishResultListener(listener));
}
Comment on lines +1357 to 1396
Copy link

@coderabbitai coderabbitai bot Jan 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Critical bug: Wrong message object used in ProtocolMessage.

At line 1394, the ProtocolMessage uses the original message instead of updatedMessage. This means:

  • The action (UPDATE/DELETE/APPEND) is not sent
  • The version metadata (clientId, description, metadata) is not sent
  • The encoded message data is not sent

The server will receive the unmodified original message instead of the properly constructed update/delete message.

🔎 Proposed fix 
         ProtocolMessage msg = new ProtocolMessage(Action.message, this.name);
-        msg.messages = new Message[] { message };
+        msg.messages = new Message[] { updatedMessage };
         connectionManager.send(msg, queueMessages, Listeners.toPublishResultListener(listener));
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
private void updateDeleteImpl(
Message message,
MessageOperation operation,
MessageAction action,
Callback<UpdateDeleteResult> listener
) throws AblyException {
if (message.serial == null || message.serial.isEmpty()) {
throw AblyException.fromErrorInfo(new ErrorInfo("Message serial cannot be empty", 400, 40003));
}
ConnectionManager connectionManager = ably.connection.connectionManager;
ConnectionManager.State connectionState = connectionManager.getConnectionState();
boolean queueMessages = ably.options.queueMessages;
if (!connectionManager.isActive() || (connectionState.queueEvents && !queueMessages)) {
throw AblyException.fromErrorInfo(connectionState.defaultErrorInfo);
}
boolean connected = (connectionState.sendEvents);
Message updatedMessage = new Message(message.name, message.data, message.extras);
updatedMessage.action = action;
updatedMessage.version = new MessageVersion();
if (operation != null) {
updatedMessage.version.clientId = operation.clientId;
updatedMessage.version.description = operation.description;
updatedMessage.version.metadata = operation.metadata;
}
try {
ably.auth.checkClientId(message, true, connected);
updatedMessage.encode(options);
} catch (AblyException e) {
if (listener != null) {
listener.onError(e.errorInfo);
}
return;
}
ProtocolMessage msg = new ProtocolMessage(Action.message, this.name);
msg.messages = new Message[] { message };
connectionManager.send(msg, queueMessages, Listeners.toPublishResultListener(listener));
}
private void updateDeleteImpl(
Message message,
MessageOperation operation,
MessageAction action,
Callback<UpdateDeleteResult> listener
) throws AblyException {
if (message.serial == null || message.serial.isEmpty()) {
throw AblyException.fromErrorInfo(new ErrorInfo("Message serial cannot be empty", 400, 40003));
}
ConnectionManager connectionManager = ably.connection.connectionManager;
ConnectionManager.State connectionState = connectionManager.getConnectionState();
boolean queueMessages = ably.options.queueMessages;
if (!connectionManager.isActive() || (connectionState.queueEvents && !queueMessages)) {
throw AblyException.fromErrorInfo(connectionState.defaultErrorInfo);
}
boolean connected = (connectionState.sendEvents);
Message updatedMessage = new Message(message.name, message.data, message.extras);
updatedMessage.action = action;
updatedMessage.version = new MessageVersion();
if (operation != null) {
updatedMessage.version.clientId = operation.clientId;
updatedMessage.version.description = operation.description;
updatedMessage.version.metadata = operation.metadata;
}
try {
ably.auth.checkClientId(message, true, connected);
updatedMessage.encode(options);
} catch (AblyException e) {
if (listener != null) {
listener.onError(e.errorInfo);
}
return;
}
ProtocolMessage msg = new ProtocolMessage(Action.message, this.name);
msg.messages = new Message[] { updatedMessage };
connectionManager.send(msg, queueMessages, Listeners.toPublishResultListener(listener));
}
🤖 Prompt for AI Agents 
In @lib/src/main/java/io/ably/lib/realtime/ChannelBase.java around lines 1357 -
1396, The ProtocolMessage is sending the original Message object instead of the
constructed updatedMessage, so the action, version and encoded data are lost;
change the send call to use updatedMessage (i.e., set msg.messages = new
Message[] { updatedMessage }) so the ProtocolMessage includes the modified
action/version/encoded payload before connectionManager.send(...).


/**
Expand Down Expand Up @@ -1681,7 +1711,7 @@ public void once(ChannelState state, ChannelStateListener listener) {
*/
public void sendProtocolMessage(ProtocolMessage protocolMessage, CompletionListener listener) throws AblyException {
ConnectionManager connectionManager = ably.connection.connectionManager;
connectionManager.send(protocolMessage, ably.options.queueMessages, listener);
connectionManager.send(protocolMessage, ably.options.queueMessages, Listeners.fromCompletionListener(listener));
}

private static final String TAG = Channel.class.getName();
Expand Down
10 changes: 6 additions & 4 deletions lib/src/main/java/io/ably/lib/realtime/Presence.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
import io.ably.lib.types.PresenceMessage;
import io.ably.lib.types.PresenceSerializer;
import io.ably.lib.types.ProtocolMessage;
import io.ably.lib.types.PublishResult;
import io.ably.lib.util.Listeners;
import io.ably.lib.util.Log;
import io.ably.lib.util.StringUtils;

Expand Down Expand Up @@ -120,9 +122,9 @@ public synchronized PresenceMessage[] get(String clientId, boolean wait) throws
return get(new Param(GET_WAITFORSYNC, String.valueOf(wait)), new Param(GET_CLIENTID, clientId));
}

void addPendingPresence(PresenceMessage presenceMessage, CompletionListener listener) {
void addPendingPresence(PresenceMessage presenceMessage, Callback<PublishResult> listener) {
synchronized(channel) {
final QueuedPresence queuedPresence = new QueuedPresence(presenceMessage,listener);
final QueuedPresence queuedPresence = new QueuedPresence(presenceMessage, Listeners.unwrap(listener));
pendingPresence.add(queuedPresence);
}
}
Expand Down Expand Up @@ -763,7 +765,7 @@ public void updatePresence(PresenceMessage msg, CompletionListener listener) thr
ProtocolMessage message = new ProtocolMessage(ProtocolMessage.Action.presence, channel.name);
message.presence = new PresenceMessage[] { msg };
ConnectionManager connectionManager = ably.connection.connectionManager;
connectionManager.send(message, ably.options.queueMessages, listener);
connectionManager.send(message, ably.options.queueMessages, Listeners.fromCompletionListener(listener));
break;
default:
throw AblyException.fromErrorInfo(new ErrorInfo("Unable to enter presence channel in detached or failed state", 400, 91001));
Expand Down Expand Up @@ -892,7 +894,7 @@ private void sendQueuedMessages() {
pendingPresence.clear();

try {
connectionManager.send(message, queueMessages, listener);
connectionManager.send(message, queueMessages, Listeners.fromCompletionListener(listener));
} catch(AblyException e) {
Log.e(TAG, "sendQueuedMessages(): Unexpected exception sending message", e);
if(listener != null)
Expand Down
Loading
Loading