Integrate Custom Listener

Couchbase Lite database peer-to-peer sync — integrate a custom-built listener

Overview

This is an Enterprise Edition feature.

This content covers how to integrate a custom MessageEndpointListener solution with Couchbase Lite to handle the data transfer, which is the sending and receiving of data. Where applicable, we discuss how to integrate Couchbase Lite into the workflow.

The following sections describe a typical Peer-to-Peer workflow.

Peer Discovery

Peer discovery is the first step. The communication framework will generally include a peer discovery API for devices to advertise themselves on the network and to browse for other peers.

Active Peer

The first step is to initialize the Couchbase Lite database.

Passive Peer

In addition to initializing the database, the Passive Peer must initialize the MessageEndpointListener. The MessageEndpointListener acts as a listener for incoming connections.

val listener = MessageEndpointListener(
    MessageEndpointListenerConfigurationFactory.newConfig(collections, ProtocolType.MESSAGE_STREAM)
)

Peer Selection and Connection Setup

Once a peer device is found, the application code must decide whether it should establish a connection with that peer. This step includes inviting a peer to a session and peer authentication.

This is handled by the Communication Framework.

Once the remote peer has been authenticated, the next step is to connect with that peer and initialize the MessageEndpoint API.

Replication Setup

Active Peer

When the connection is established, the Active Peer must instantiate a MessageEndpoint object corresponding to the remote peer.

// The delegate must implement the `MessageEndpointDelegate` protocol.
val messageEndpoint = MessageEndpoint("UID:123", "active", ProtocolType.MESSAGE_STREAM, delegate)

The MessageEndpoint constructor takes the following arguments:

1. uid: A unique ID that represents the remote Active Peer.
2. target: This represents the remote Passive Peer and could be any suitable representation of the remote peer. It could be an ID, URL, etc. If using the Multipeer Connectivity Framework, this could be the MCPeerID.
3. protocolType: Specifies the kind of transport you intend to implement. There are two options:
   • The default (MESSAGE_STREAM) means that you want to "send a series of messages", or in other words the Communication Framework will control the formatting of messages so that there are clear boundaries between messages.
   • The alternative (BYTE_STREAM) means that you just want to send raw bytes over the stream and Couchbase should format for you to ensure that messages get delivered in full.
     
     Typically, the Communication Framework will handle message assembly and disassembly, so you would use the MESSAGE_STREAM option in most cases.
4. delegate: The delegate that will implement the MessageEndpointDelegate protocol, which is a factory for MessageEndpointConnection.

Then, a Replicator is instantiated with the initialized MessageEndpoint as the target.

// Create the replicator object.
val repl = Replicator(
    ReplicatorConfigurationFactory.newConfig(
        collections = mapOf(collections to null),
        target = messageEndpoint
    )
)

// Start the replication.
repl.start()
this.replicator = repl

Next, Couchbase Lite will call back the application code through the MessageEndpointDelegate lambda. When the application receives the callback, it must create an instance of MessageEndpointConnection and return it.

/* implementation of MessageEndpointDelegate */
val delegate: MessageEndpointDelegate = { endpoint ->
    ActivePeerConnection()
}

Next, Couchbase Lite will call back the application code through the MessageEndpointConnection.open() method.

/* implementation of MessageEndpointConnection */
override fun open(connection: ReplicatorConnection, completion: MessagingCompletion) {
    replicatorConnection = connection
    completion(true, null)
}

The connection argument is then set on an instance variable. The application code must keep track of every ReplicatorConnection associated with every MessageEndpointConnection.

The MessageError argument in the completion block specifies whether the error is recoverable or not. If it is a recoverable error, the replicator will begin a retry process, creating a new MessageEndpointConnection instance.

Passive Peer

After connection establishment on the Passive Peer, the first step is to initialize a new MessageEndpointConnection and pass it to the listener. This message tells the listener to accept incoming data from that peer.

/* implements MessageEndpointConnection */
val connection = PassivePeerConnection()
listener?.accept(connection)

listener is the instance of the MessageEndpointListener that was created in the first step (Peer Discovery ).

Couchbase Lite will call the application code back through the MessageEndpointConnection.open() method.

/* implementation of MessageEndpointConnection */
override fun open(connection: ReplicatorConnection, completion: MessagingCompletion) {
    replicatorConnection = connection
    completion(true, null)
}

The connection argument is then set on an instance variable. The application code must keep track of every ReplicatorConnection associated with every MessageEndpointConnection.

At this point, the connection is established, and both peers are ready to exchange data.

Push/Pull Replication

Typically, an application needs to send data and receive data. The directionality of the replication could be any of the following:

• Push only: The data is pushed from the local database to the remote database.
• Pull only: The data is pulled from the remote database to the local database.
• Push and Pull: The data is exchanged both ways.

Usually, the remote is a Sync Gateway database identified through a URL. In Peer-to-Peer syncing, the remote is another Couchbase Lite database.

The replication lifecycle is handled through the MessageEndpointConnection.

Active Peer

When Couchbase Lite calls back the application code through the MessageEndpointConnection.send() method, you should send that data to the other peer using the Communication Framework.

/* implementation of MessageEndpointConnection */
override fun send(message: Message, completion: MessagingCompletion) {
    /* send the data to the other peer */
    /* ... */
    /* call the completion handler once the message is sent */
    completion(true, null)
}

Once the data is sent, call the completion block to acknowledge the completion. You can use the MessageError in the completion block to specify whether the error is recoverable. If it is a recoverable error, the replicator will begin a retry process, creating a new MessageEndpointConnection.

When data is received from the Passive Peer via the Communication Framework, you call the ReplicatorConnection.receive() method.

replicatorConnection?.receive(message)

The ReplicatorConnection’s receive() method is called. Which then processes the data to persist to the local database.

Passive Peer

As in the case of the Active Peer, the Passive Peer must implement the MessageEndpointConnection.send() method to send data to the other peer.

/* implementation of MessageEndpointConnection */
override fun send(message: Message, completion: MessagingCompletion) {
    /* send the data to the other peer */
    /* ... */
    /* call the completion handler once the message is sent */
    completion(true, null)
}

Once the data is sent, call the completion block to acknowledge the completion. You can use the MessageError in the completion block to specify whether the error is recoverable. If it is a recoverable error, the replicator will begin a retry process, creating a new MessageEndpointConnection.

When data is received from the Active Peer via the Communication Framework, you call the ReplicatorConnection.receive() method.

replicatorConnection?.receive(message)

Connection Teardown

When a peer disconnects from a peer-to-peer network, all connected peers are notified. The disconnect notification is a good opportunity to close and remove a replication connection. The steps to tear down the connection are slightly different depending on whether the active or passive peer disconnects first. We will cover each case below.

Initiated by Active Peer

Active Peer

When an Active Peer disconnects, it must call the ReplicatorConnection.close() method.

fun disconnect() {
    replicatorConnection?.close(null)
    replicatorConnection = null
}

Then, Couchbase Lite will call back your code through the MessageEndpointConnection.close() to allow the application to disconnect with the Communication Framework.

override fun close(error: Exception?, completion: MessagingCloseCompletion) {
    /* disconnect with communications framework */
    /* ... */
    /* call completion handler */
    completion()
}

Passive Peer

When the Passive Peer receives the corresponding disconnect notification from the Communication Framework, it must call the ReplicatorConnection.close() method.

replicatorConnection?.close(null)

Then, Couchbase Lite will call back your code through the MessageEndpointConnection.close() to allow the application to disconnect with the Communication Framework.

/* implementation of MessageEndpointConnection */
override fun close(error: Exception?, completion: MessagingCloseCompletion) {
    /* disconnect with communications framework */
    /* ... */
    /* call completion handler */
    completion()
}

Initiated by Passive Peer

Passive Peer

When the Passive Peer disconnects, it must class the MessageEndpointListener.closeAll() method.

listener?.closeAll()

Then, Couchbase Lite will call back your code through the MessageEndpointConnection.close() to allow the application to disconnect with the Communication Framework.

/* implementation of MessageEndpointConnection */
override fun close(error: Exception?, completion: MessagingCloseCompletion) {
    /* disconnect with communications framework */
    /* ... */
    /* call completion handler */
    completion()
}

Active Peer

When the Active Peer receives the corresponding disconnect notification from the Communication Framework, it must call the ReplicatorConnection.close() method.

fun disconnect() {
    replicatorConnection?.close(null)
    replicatorConnection = null
}

Then, Couchbase Lite will call back your code through the MessageEndpointConnection.close() to allow the application to disconnect with the Communication Framework.

override fun close(error: Exception?, completion: MessagingCloseCompletion) {
    /* disconnect with communications framework */
    /* ... */
    /* call completion handler */
    completion()
}