service = client.service().messaging();

val service = client.service().messaging()

Depending on the client you initialised — ComapiClient or RxComapiClient — you have access to either a callback or a reactive messaging API.

The reactive flavour returns Observables you subscribe to. The callback flavour takes a Callback<T> as its last argument; the request runs on a background I/O thread and the result is delivered on the main (UI) thread.

public interface Callback<T> {
    void success(T result);
    void error(Throwable t);
}

🚧
The reactive API uses RxJava 1.x
RxComapiClient returns rx.Observable (RxJava 1.x), not io.reactivex.Observable (RxJava 2 / 3). If you don't already have RxJava 1.x in your project, the callback flavour or the Kotlin coroutine bridge shown below will usually be the simpler choice.

Calling the API from Kotlin coroutines

If your codebase is on coroutines, wrap any callback-style method in a small suspend helper. The pattern is the same for every call.

suspend fun <T> awaitResult(
    call: (Callback<T>) -> Unit
): T = suspendCancellableCoroutine { cont ->
    call(object : Callback<T> {
        override fun success(result: T) {
            if (cont.isActive) cont.resume(result) {}
        }
        override fun error(t: Throwable) {
            if (cont.isActive) cont.resumeWithException(t)
        }
    })
}

// Usage:
val result: ComapiResult<ConversationDetails> = awaitResult { cb ->
    service.createConversation(conversation, cb)
}

Data consistency

To manage concurrent updates of a conversation from many devices or websites, use the ETag.

When obtaining conversation details from our services, you can find an ETag in the ComapiResult object (via result.getETag()). It describes the version of the server data you received. Pass that same ETag with your next update; if the server data has changed in the meantime, the service rejects the modification until you obtain the most recent version and supply its ETag. This keeps the data consistent across many devices.

📘
Where to read the ETag
ComapiResult.getETag() works for any call that returns a ComapiResult. Conversation objects returned by getConversations(...) also expose getETag() directly. ConversationDetails (returned by createConversation / getConversation) does not — read the ETag from the surrounding ComapiResult.

Messaging service API

Create conversation

ConversationCreate conversation = ConversationCreate.builder()
    // Unique conversation identifier.
    .setId("1234")
    // Description
    .setDescription("This is my first conversation")
    // Name
    .setName("Awesome chat")
    // Is this conversation visible to users who are not participants
    .setPublic(false)
    /* Sets what permissions 'owner' and 'participant' have in this conversation.
       You can set whether they can: add new participants, remove participants, send messages.
       By default both can send messages and add participants, neither can remove participants.
       Here we additionally grant CanRemoveParticipants to the 'owner'. */
    .setRoles(new Roles(
        Role.builder().setCanRemoveParticipants().build(),
        new Role()
    ))
    .build();

val conversation = ConversationCreate.builder()
    .setId("1234")
    .setDescription("This is my first conversation")
    .setName("Awesome chat")
    .setPublic(false)
    .setRoles(
        Roles(
            Role.builder().setCanRemoveParticipants().build(),
            Role()
        )
    )
    .build()

Then pass this object to the service:

service.createConversation(conversation, new Callback<ComapiResult<ConversationDetails>>() {
    @Override public void success(ComapiResult<ConversationDetails> result) { /* implement */ }
    @Override public void error(Throwable t) { /* implement */ }
});

service.createConversation(conversation, object : Callback<ComapiResult<ConversationDetails>> {
    override fun success(result: ComapiResult<ConversationDetails>) { /* implement */ }
    override fun error(t: Throwable) { /* implement */ }
})

// Or with the coroutine bridge:
val result = awaitResult<ComapiResult<ConversationDetails>> { cb ->
    service.createConversation(conversation, cb)
}

rxService.createConversation(conversation)
    .subscribe(new Observer<ComapiResult<ConversationDetails>>() { /* implement */ });

rxService.createConversation(conversation)
    .subscribe(object : Observer<ComapiResult<ConversationDetails>> { /* implement */ })

Update conversation

ConversationUpdate update = ConversationUpdate.builder()
    .setDescription("New description")
    .setName("Different name")
    .build();

val update = ConversationUpdate.builder()
    .setDescription("New description")
    .setName("Different name")
    .build()

Then pass this object to the service. The ETag must be the value you most recently obtained for that conversation.

service.updateConversation(conversationId, update, eTag,
    new Callback<ComapiResult<ConversationDetails>>() { /* implement */ });

service.updateConversation(conversationId, update, eTag,
    object : Callback<ComapiResult<ConversationDetails>> { /* implement */ })

rxService.updateConversation(conversationId, update, eTag)
    .subscribe(new Observer<ComapiResult<ConversationDetails>>() { /* implement */ });

rxService.updateConversation(conversationId, update, eTag)
    .subscribe(object : Observer<ComapiResult<ConversationDetails>> { /* implement */ })

Delete conversation

service.deleteConversation(conversationId, eTag,
    new Callback<ComapiResult<Void>>() { /* implement */ });

service.deleteConversation(conversationId, eTag,
    object : Callback<ComapiResult<Void>> { /* implement */ })

rxService.deleteConversation(conversationId, eTag)
    .subscribe(new Observer<ComapiResult<Void>>() { /* implement */ });

rxService.deleteConversation(conversationId, eTag)
    .subscribe(object : Observer<ComapiResult<Void>> { /* implement */ })

Get conversation

service.getConversation(conversationId,
    new Callback<ComapiResult<ConversationDetails>>() { /* implement */ });

service.getConversation(conversationId,
    object : Callback<ComapiResult<ConversationDetails>> { /* implement */ })

rxService.getConversation(conversationId)
    .subscribe(new Observer<ComapiResult<ConversationDetails>>() { /* implement */ });

rxService.getConversation(conversationId)
    .subscribe(object : Observer<ComapiResult<ConversationDetails>> { /* implement */ })

ConversationDetails exposes:

// Is the conversation public or private (Boolean, may be null)
details.isPublic();
// Unique identifier
details.getId();
// Name of the conversation
details.getName();
// Description of the conversation
details.getDescription();
// Privileges of owner and participants in this conversation
details.getRoles();

details.isPublic       // Boolean? — may be null
details.id
details.name
details.description
details.roles

Query conversations

service.getConversations(isPublic,
    new Callback<ComapiResult<List<Conversation>>>() { /* implement */ });

service.getConversations(isPublic,
    object : Callback<ComapiResult<List<Conversation>>> { /* implement */ })

rxService.getConversations(isPublic)
    .subscribe(new Observer<ComapiResult<List<Conversation>>>() { /* implement */ });

rxService.getConversations(isPublic)
    .subscribe(object : Observer<ComapiResult<List<Conversation>>> { /* implement */ })

📘
Deprecated overload
There is also a getConversations(Scope, callback) overload that takes a Scope enum. It is deprecated; prefer the boolean isPublic form shown above.

Conversation extends ConversationDetails and adds three more fields:

conversation.isPublic();
conversation.getId();
conversation.getName();
conversation.getDescription();
conversation.getRoles();
// ETag for comparing the local data version against the server version
conversation.getETag();
// Number of participants in the conversation
conversation.getParticipantCount();
// Latest event id of a sent message; null if there are no messages
conversation.getLatestSentEventId();

conversation.isPublic
conversation.id
conversation.name
conversation.description
conversation.roles
conversation.eTag
conversation.participantCount
val latest: Long? = conversation.latestSentEventId

Remove conversation participants

Provide a list of profile ids to be removed from the conversation.

service.removeParticipants(conversationId, ids,
    new Callback<ComapiResult<Void>>() { /* implement */ });

service.removeParticipants(conversationId, ids,
    object : Callback<ComapiResult<Void>> { /* implement */ })

rxService.removeParticipants(conversationId, ids)
    .subscribe(new Observer<ComapiResult<Void>>() { /* implement */ });

rxService.removeParticipants(conversationId, ids)
    .subscribe(object : Observer<ComapiResult<Void>> { /* implement */ })

Query conversation participants

service.getParticipants(conversationId,
    new Callback<ComapiResult<List<Participant>>>() { /* implement */ });

service.getParticipants(conversationId,
    object : Callback<ComapiResult<List<Participant>>> { /* implement */ })

rxService.getParticipants(conversationId)
    .subscribe(new Observer<ComapiResult<List<Participant>>>() { /* implement */ });

rxService.getParticipants(conversationId)
    .subscribe(object : Observer<ComapiResult<List<Participant>>> { /* implement */ })

Add participants to conversation

Build each Participant with Participant.builder().setId(...).setIsOwner() or .setIsParticipant().

List<Participant> participants = new ArrayList<>();
participants.add(Participant.builder().setId("profile-1").setIsParticipant().build());

service.addParticipants(conversationId, participants,
    new Callback<ComapiResult<Void>>() { /* implement */ });

val participants = listOf(
    Participant.builder().setId("profile-1").setIsParticipant().build()
)

service.addParticipants(conversationId, participants,
    object : Callback<ComapiResult<Void>> { /* implement */ })

rxService.addParticipants(conversationId, participants)
    .subscribe(new Observer<ComapiResult<Void>>() { /* implement */ });

rxService.addParticipants(conversationId, participants)
    .subscribe(object : Observer<ComapiResult<Void>> { /* implement */ })

Send a message in a conversation

Simple text body:

service.sendMessage(conversationId, body,
    new Callback<ComapiResult<MessageSentResponse>>() { /* implement */ });

service.sendMessage(conversationId, body,
    object : Callback<ComapiResult<MessageSentResponse>> { /* implement */ })

rxService.sendMessage(conversationId, body)
    .subscribe(new Observer<ComapiResult<MessageSentResponse>>() { /* implement */ });

rxService.sendMessage(conversationId, body)
    .subscribe(object : Observer<ComapiResult<MessageSentResponse>> { /* implement */ })

You can later update the message status using the conversationId and the id returned in MessageSentResponse:

// Globally unique message identifier
response.getId();
// Conversation event identifier — monotonically increasing, used to order messages
response.getEventId();

response.id
response.eventId

The more advanced form lets you attach FCM/APNS push details, custom metadata, and one or more Parts:

Map<String, Object> data = new HashMap<>();
data.put("key", "value");

Map<String, Object> fcm = new HashMap<>();
fcm.put("data", data);
fcm.put("notification", "{ \"title\":\"Message\", \"body\":\"Hi!\" }");

Map<String, Object> apns = new HashMap<>();
apns.put("alert", "Hi!");

String text = "Hi";
Part part = Part.builder()
    .setData(text)
    .setName("body")
    .setSize(text.getBytes(StandardCharsets.UTF_8).length)
    .setType("text/plain")
    .build();

MessageToSend message = MessageToSend.builder()
    .setAlert(fcm, apns)
    .setMetadata(data)
    .addPart(part)
    .build();

val data: Map<String, Any> = mapOf("key" to "value")

val fcm: Map<String, Any> = mapOf(
    "data" to data,
    "notification" to "{ \"title\":\"Message\", \"body\":\"Hi!\" }"
)

val apns: Map<String, Any> = mapOf("alert" to "Hi!")

val text = "Hi"
val part = Part.builder()
    .setData(text)
    .setName("body")
    .setSize(text.toByteArray(Charsets.UTF_8).size.toLong())
    .setType("text/plain")
    .build()

val message = MessageToSend.builder()
    .setAlert(fcm, apns)
    .setMetadata(data)
    .addPart(part)
    .build()

Then call:

service.sendMessage(conversationId, message,
    new Callback<ComapiResult<MessageSentResponse>>() { /* implement */ });

service.sendMessage(conversationId, message,
    object : Callback<ComapiResult<MessageSentResponse>> { /* implement */ })

rxService.sendMessage(conversationId, message)
    .subscribe(new Observer<ComapiResult<MessageSentResponse>>() { /* implement */ });

rxService.sendMessage(conversationId, message)
    .subscribe(object : Observer<ComapiResult<MessageSentResponse>> { /* implement */ })

Update message status

MessageStatus is an enum with two lowercase values: delivered and read.

MessageStatusUpdate update = MessageStatusUpdate.builder()
    .addMessageId("id")
    .setStatus(MessageStatus.delivered)
    .build();

List<MessageStatusUpdate> updates = new ArrayList<>();
updates.add(update);

val update = MessageStatusUpdate.builder()
    .addMessageId("id")
    .setStatus(MessageStatus.delivered)
    .build()

val updates = listOf(update)

Then pass the list of delivered / read statuses to the service:

service.updateMessageStatus(conversationId, updates,
    new Callback<ComapiResult<Void>>() { /* implement */ });

service.updateMessageStatus(conversationId, updates,
    object : Callback<ComapiResult<Void>> { /* implement */ })

rxService.updateMessageStatus(conversationId, updates)
    .subscribe(new Observer<ComapiResult<Void>>() { /* implement */ });

rxService.updateMessageStatus(conversationId, updates)
    .subscribe(object : Observer<ComapiResult<Void>> { /* implement */ })

Query conversation events

Provide a conversation event ID (from) and a limit. The response groups events by type so you don't have to instanceof-check.

service.queryConversationEvents(conversationId, from, limit,
    new Callback<ComapiResult<ConversationEventsResponse>>() { /* implement */ });

service.queryConversationEvents(conversationId, from, limit,
    object : Callback<ComapiResult<ConversationEventsResponse>> { /* implement */ })

rxService.queryConversationEvents(conversationId, from, limit)
    .subscribe(new Observer<ComapiResult<ConversationEventsResponse>>() { /* implement */ });

rxService.queryConversationEvents(conversationId, from, limit)
    .subscribe(object : Observer<ComapiResult<ConversationEventsResponse>> { /* implement */ })

📘
Deprecated overload
An older queryEvents(...) method returns EventsQueryResponse. It is deprecated; queryConversationEvents(...) provides better typing of the events in the response.

/* Events in the order they were received.
   The collection can contain MessageSentEvent, MessageDeliveredEvent and MessageReadEvent.
   Cast elements as appropriate. */
response.getEventsInOrder();
// Parsed message-sent events
response.getMessageSent();
// Parsed message-delivered events
response.getMessageDelivered();
// Parsed message-read events
response.getMessageRead();

response.eventsInOrder
response.messageSent
response.messageDelivered
response.messageRead

Learn more in Listen to events.

Query messages

To obtain a page of messages in a conversation, supply a from event ID and a limit. This lets you implement a classic "pull to load more" timeline that walks backwards from the most recent message.

service.queryMessages(conversationId, from, limit,
    new Callback<ComapiResult<MessagesQueryResponse>>() { /* implement */ });

service.queryMessages(conversationId, from, limit,
    object : Callback<ComapiResult<MessagesQueryResponse>> { /* implement */ })

rxService.queryMessages(conversationId, from, limit)
    .subscribe(new Observer<ComapiResult<MessagesQueryResponse>>() { /* implement */ });

rxService.queryMessages(conversationId, from, limit)
    .subscribe(object : Observer<ComapiResult<MessagesQueryResponse>> { /* implement */ })

The result contains the messages and any "orphaned" events — events that update messages with event IDs higher than from (for example, status updates for messages you obtained in an earlier page). Apply those events to the messages you've already loaded.

// Messages matching the query
response.getMessages();
// Latest event id taken into account when constructing the result
response.getLatestEventId();
// Earliest event id taken into account when constructing the result
response.getEarliestEventId();
// Events updating messages for conversation event ids higher than `from`
response.getOrphanedEvents();

response.messages
response.latestEventId
response.earliestEventId
response.orphanedEvents

getMessages() returns a List<MessageReceived>:

// Message unique identifier
message.getMessageId();
// Monotonically increasing event id for this message in this conversation
message.getSentEventId();
// Sender (Sender — exposes getId(), getName(), getAvatarUrl())
message.getFromWhom();
// Server-side sender id (internal — typically not surfaced in the app)
message.getSentBy();
// When the message was sent (ISO-8601 String)
message.getSentOn();
// Conversation unique identifier
message.getConversationId();
/* Map<String, MessageReceived.Status> keyed by profile id.
   Each Status has getStatus() (MessageStatus) and getTimestamp() (String). */
message.getStatusUpdate();

message.messageId
message.sentEventId
val sender: Sender? = message.fromWhom
message.sentBy
message.sentOn
message.conversationId
val statuses: Map<String, MessageReceived.Status>? = message.statusUpdate

getOrphanedEvents() returns a List<OrphanedEvent>:

// Conversation event id
orphan.getConversationEventId();
// Event name
orphan.getName();
// Unique event identifier
orphan.getEventId();
// Id of the updated message
orphan.getMessageId();
// Id of the conversation the updated message belongs to
orphan.getConversationId();
// Profile id of the user who changed the message status
orphan.getProfileId();
// When the message status changed (ISO-8601)
orphan.getTimestamp();
// true if this event is of type "message delivered"
orphan.isEventTypeDelivered();
// true if this event is of type "message read"
orphan.isEventTypeRead();

orphan.conversationEventId
orphan.name
orphan.eventId
orphan.messageId
orphan.conversationId
orphan.profileId
orphan.timestamp
orphan.isEventTypeDelivered
orphan.isEventTypeRead

Send "user is typing" event

Notify other participants when the user starts or stops typing.

service.isTyping(conversationId, isTyping,
    new Callback<ComapiResult<Void>>() { /* implement */ });

service.isTyping(conversationId, isTyping,
    object : Callback<ComapiResult<Void>> { /* implement */ })

rxService.isTyping(conversationId, isTyping)
    .subscribe(new Observer<ComapiResult<Void>>() { /* implement */ });

rxService.isTyping(conversationId, isTyping)
    .subscribe(object : Observer<ComapiResult<Void>> { /* implement */ })

📘
Throttle these calls
Typing events are best fired on a debounce — e.g. send isTyping=true when the user types and hasn't typed in the last few seconds, and isTyping=false after a short idle timeout. Avoid sending one event per keystroke.

Upload content data

service.uploadContent(folder, contentData,
    new Callback<ComapiResult<UploadContentResponse>>() { /* implement */ });

service.uploadContent(folder, contentData,
    object : Callback<ComapiResult<UploadContentResponse>> { /* implement */ })

rxService.uploadContent(folder, contentData)
    .subscribe(new Observer<ComapiResult<UploadContentResponse>>() { /* implement */ });

rxService.uploadContent(folder, contentData)
    .subscribe(object : Observer<ComapiResult<UploadContentResponse>> { /* implement */ })

Content data can be created from a File, a byte[], or a base64-encoded String:

ContentData fromFile   = ContentData.create(file,   type, name);
ContentData fromBytes  = ContentData.create(bytes,  type, name);
ContentData fromBase64 = ContentData.create(base64, type, name);

val fromFile   = ContentData.create(file,   type, name)
val fromBytes  = ContentData.create(bytes,  type, name)
val fromBase64 = ContentData.create(base64, type, name)

UploadContentResponse includes the full URL of the uploaded file (response.getUrl()); you can then send a message whose Part references that URL.

🚧
Scoped storage on Android 10+ (API 29)
If the file you pass to ContentData.create(file, ...) came from a content:// URI (e.g. the document picker or photo picker), copy it into your app's cache directory first — direct file-path access to user-selected media is restricted on Android 10 and above.