Peer-to-Peer Synchronization — Custom

    +

    Description — Couchbase Lite database peer-to-peer sync concepts using message endpoints
    Related Content — Peer-to-Peer Synchronization over web sockets

    Overview

    Enterprise Edition only
    Peer-to-Peer Synchronization is an Enterprise Edition feature. You must purchase the Enterprise License which includes official Couchbase Support to use it in production (also see the FAQ).

    This content covers how to integrate a custom MessageEndpointListener solution with Couchbase Lite to handle the data transfer; 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.

    discovery

    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 as a listener for incoming connections.

    DatabaseConfiguration databaseConfiguration = new DatabaseConfiguration();
    Database database = new Database(DB_NAME, databaseConfiguration);
    MessageEndpointListenerConfiguration listenerConfiguration = new MessageEndpointListenerConfiguration(
        database,
        ProtocolType.MESSAGE_STREAM);
    this.messageEndpointListener = new MessageEndpointListener(listenerConfiguration);

    Peer Selection and Connection Setup

    Once a peer device is found, it is the application code’s responsibility to 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.

    selection

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

    Replication Setup

    connection

    Active Peer

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

    DatabaseConfiguration databaseConfiguration = new DatabaseConfiguration(context);
    Database database = new Database(DB_NAME, databaseConfiguration);
    
    // The delegate must implement the `MessageEndpointDelegate` protocol.
    MessageEndpoint messageEndpointTarget = new MessageEndpoint(
        "UID:123",
        "active",
        ProtocolType.MESSAGE_STREAM,
        this);

    The MessageEndpoint initializer 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 MultiPeerConnectivity Framework, this could be the MCPeerID.

    3. protocolType: specifies the kind of transport you intend to implement. There are two options.

      • The default (MessageStream) 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 (ByteStream) 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 MessageType 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.

    ReplicatorConfiguration config = new ReplicatorConfiguration(database, messageEndpointTarget);
    
    // Create the replicator object.
    replicator = new Replicator(config);
    // Start the replication.
    replicator.start();

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

    /* implementation of MessageEndpointDelegate */
    @NonNull
    @Override
    public MessageEndpointConnection createConnection(@NonNull MessageEndpoint endpoint) {
        return new ActivePeerConnection(); /* implements MessageEndpointConnection */
    }

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

    /* implementation of MessageEndpointConnection */
    @Override
    public void open(@NonNull ReplicatorConnection connection, @NonNull MessagingCompletion completion) {
        replicatorConnection = connection;
        completion.complete(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 is used to specify if the error is recoverable or not. If it is a recoverable error, the replicator will kick off a retry process which will result to creating a new MessageEndpointConnection instance.

    Passive Peer

    The first step after connection establishment on the passive peer is to initialize a new MessageEndpointConnection and pass it to the listener. This tells the listener to accept incoming data from that peer.

    PassivePeerConnection connection = new PassivePeerConnection(context); /* implements
    MessageEndpointConnection */
    messageEndpointListener.accept(connection);

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

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

    /* implementation of MessageEndpointConnection */
    @Override
    public void open(@NonNull ReplicatorConnection connection, @NonNull MessagingCompletion completion) {
        replicatorConnection = connection;
        completion.complete(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. 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 which is identified through a URL. In the context of peer-to-peer syncing, the remote is another Couchbase Lite database.

    replication

    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
    public void send(@NonNull Message message, @NonNull MessagingCompletion completion) {
        /* send the data to the other peer */
        /* ... */
        /* call the completion handler once the message is sent */
        completion.complete(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 if the error is recoverable or not. If it is a recoverable error, the replicator will kick off a retry process which will result to 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 replication connection’s receive method is called which then processes the data in order to persist it 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
    public void send(@NonNull Message message, @NonNull MessagingCompletion completion) {
        /* send the data to the other peer */
        /* ... */
        /* call the completion handler once the message is sent */
        completion.complete(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 if the error is recoverable or not. If it is a recoverable error, the replicator will kick off a retry process which will result to 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 teardown the connection are slightly different depending on whether it is the active or passive peer that disconnects first. We will cover each case below.

    Initiated by Active Peer

    dis active
    Active Peer

    When an active peer disconnects, it must call the ReplicatorConnection.close method.

    replicatorConnection.close(null);

    Then, Couchbase Lite will call back your code through the MessageEndpointConnection.close to give the application a chance to disconnect with the Communication Framework.

    @Override
    public void close(Exception error, @NonNull MessagingCloseCompletion completion) {
        /* disconnect with communications framework */
        /* ... */
        /* call completion handler */
        completion.complete();
    }
    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 give the application a chance to disconnect with the Communication Framework.

    /* implementation of MessageEndpointConnection */
    @Override
    public void close(Exception error, @NonNull MessagingCloseCompletion completion) {
        /* disconnect with communications framework */
        /* ... */
        /* call completion handler */
        completion.complete();
    }

    Initiated by Passive Peer

    dis passive
    Passive Peer

    When the passive disconnects, it must class the MessageEndpointListener.closeAll method.

    messageEndpointListener.closeAll();

    Then, Couchbase Lite will call back your code through the MessageEndpointConnection.close to give the application a chance to disconnect with the Communication Framework.

    /* implementation of MessageEndpointConnection */
    @Override
    public void close(Exception error, @NonNull MessagingCloseCompletion completion) {
        /* disconnect with communications framework */
        /* ... */
        /* call completion handler */
        completion.complete();
    }
    Active Peer

    When the active 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 give the application a chance to disconnect with the Communication Framework.

    @Override
    public void close(Exception error, @NonNull MessagingCloseCompletion completion) {
        /* disconnect with communications framework */
        /* ... */
        /* call completion handler */
        completion.complete();
    }