com.r3corda.core.messaging / MessagingService

# MessagingService

`interface MessagingService`

A MessagingService sits at the boundary between a message routing / networking layer and the core platform code.

A messaging system must provide the ability to send 1:many messages, potentially to an abstract "group", the membership of which is defined elsewhere. Messages are atomic and the system guarantees that a sent message eventually will arrive in the exact form it was sent, however, messages can be arbitrarily re-ordered or delayed.

Example implementations might be a custom P2P layer, Akka, Apache Kafka, etc. It is assumed that the message layer is reliable and as such messages may be stored to disk once queued.

### Properties

myAddress | `abstract val myAddress: SingleMessageRecipient` Returns an address that refers to this node.

### Functions

addMessageHandler | `abstract fun addMessageHandler(topic: String = "", sessionID: Long = DEFAULT_SESSION_ID, callback: (Message, MessageHandlerRegistration) -> Unit): MessageHandlerRegistration` The provided function will be invoked for each received message whose topic matches the given string. The callback will run on threads provided by the messaging service, and the callback is expected to be thread safe as a result. `abstract fun addMessageHandler(topicSession: TopicSession, callback: (Message, MessageHandlerRegistration) -> Unit): MessageHandlerRegistration` The provided function will be invoked for each received message whose topic and session matches. The callback will run on threads provided by the messaging service, and the callback is expected to be thread safe as a result.
createMessage | `abstract fun createMessage(topicSession: TopicSession, data: ByteArray, uuid: UUID = UUID.randomUUID()): Message` Returns an initialised Message with the current time, etc, already filled in.
removeMessageHandler | `abstract fun removeMessageHandler(registration: MessageHandlerRegistration): Unit` Removes a handler given the object returned from addMessageHandler. The callback will no longer be invoked once this method has returned, although executions that are currently in flight will not be interrupted.
send | `abstract fun send(message: Message, target: MessageRecipients): Unit` Sends a message to the given receiver. The details of how receivers are identified is up to the messaging implementation: the type system provides an opaque high level view, with more fine grained control being available via type casting. Once this function returns the message is queued for delivery but not necessarily delivered: if the recipients are offline then the message could be queued hours or days later.

### Extension Functions

createMessage | `fun MessagingService.createMessage(topic: String, sessionID: Long = DEFAULT_SESSION_ID, data: ByteArray): Message` Returns an initialised Message with the current time, etc, already filled in.
onNext | `fun <M : Any> MessagingService.onNext(topic: String, sessionId: Long): <ERROR CLASS><M>` Returns a ListenableFuture of the next message payload (Message.data) which is received on the given topic and sessionId. The payload is deserilaized to an object of type M. Any exceptions thrown will be captured by the future.
runOnNextMessage | `fun MessagingService.runOnNextMessage(topic: String, sessionID: Long, callback: (Message) -> Unit): Unit` Registers a handler for the given topic and session ID that runs the given callback with the message and then removes itself. This is useful for one-shot handlers that arent supposed to stick around permanently. Note that this callback doesnt take the registration object, unlike the callback to MessagingService.addMessageHandler, as the handler is automatically deregistered before the callback runs. `fun MessagingService.runOnNextMessage(topicSession: TopicSession, callback: (Message) -> Unit): Unit` Registers a handler for the given topic and session that runs the given callback with the message and then removes itself. This is useful for one-shot handlers that arent supposed to stick around permanently. Note that this callback doesnt take the registration object, unlike the callback to MessagingService.addMessageHandler.
send | `fun MessagingService.send(topic: String, sessionID: Long, payload: Any, to: MessageRecipients, uuid: UUID = UUID.randomUUID()): Unit` `fun MessagingService.send(topicSession: TopicSession, payload: Any, to: MessageRecipients, uuid: UUID = UUID.randomUUID()): Unit`
sendRequest | `fun <R : Any> MessagingService.sendRequest(topic: String, request: ServiceRequestMessage, target: SingleMessageRecipient): <ERROR CLASS><R>` Sends a ServiceRequestMessage to target and returns a ListenableFuture of the response.

### Inheritors

MessagingServiceInternal | `interface MessagingServiceInternal : MessagingService`