diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransport.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransport.java
index 17fa742..45cc529 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransport.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransport.java
@@ -15,73 +15,59 @@
  */
 package org.onosproject.store.primitives.impl;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import com.google.common.base.Throwables;
+import com.google.common.collect.Maps;
+import io.atomix.catalyst.transport.Address;
+import io.atomix.catalyst.transport.Client;
+import io.atomix.catalyst.transport.Server;
+import io.atomix.catalyst.transport.Transport;
+import org.onlab.packet.IpAddress;
+import org.onosproject.cluster.PartitionId;
+import org.onosproject.store.cluster.messaging.Endpoint;
+import org.onosproject.store.cluster.messaging.MessagingService;
 
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.util.Map;
 
-import org.onlab.packet.IpAddress;
-import org.onosproject.cluster.PartitionId;
-import org.onosproject.store.cluster.messaging.Endpoint;
-import org.onosproject.store.cluster.messaging.MessagingService;
-
-import com.google.common.base.Throwables;
-import com.google.common.collect.Maps;
-
-import io.atomix.catalyst.transport.Address;
-import io.atomix.catalyst.transport.Client;
-import io.atomix.catalyst.transport.Server;
-import io.atomix.catalyst.transport.Transport;
+import static com.google.common.base.MoreObjects.toStringHelper;
+import static com.google.common.base.Preconditions.checkNotNull;
 
 /**
- * Custom {@link Transport transport} for Copycat interactions
- * built on top of {@link MessagingService}.
- *
- * @see CopycatTransportServer
- * @see CopycatTransportClient
+ * Copycat transport implementation built on {@link MessagingService}.
  */
 public class CopycatTransport implements Transport {
-
-    /**
-     * Transport Mode.
-     */
-    public enum Mode {
-        /**
-         * Signifies transport for client {@literal ->} server interaction.
-         */
-        CLIENT,
-
-        /**
-         * Signified transport for server {@literal ->} server interaction.
-         */
-        SERVER
-    }
-
-    private final Mode mode;
     private final PartitionId partitionId;
     private final MessagingService messagingService;
     private static final Map<Address, Endpoint> EP_LOOKUP_CACHE = Maps.newConcurrentMap();
     private static final Map<Endpoint, Address> ADDRESS_LOOKUP_CACHE = Maps.newConcurrentMap();
 
-    public CopycatTransport(Mode mode, PartitionId partitionId, MessagingService messagingService) {
-        this.mode = checkNotNull(mode);
-        this.partitionId = checkNotNull(partitionId);
-        this.messagingService = checkNotNull(messagingService);
+    static final byte MESSAGE = 0x01;
+    static final byte CONNECT = 0x02;
+    static final byte CLOSE = 0x03;
+
+    static final byte SUCCESS = 0x01;
+    static final byte FAILURE = 0x02;
+
+    public CopycatTransport(PartitionId partitionId, MessagingService messagingService) {
+        this.partitionId = checkNotNull(partitionId, "partitionId cannot be null");
+        this.messagingService = checkNotNull(messagingService, "messagingService cannot be null");
     }
 
     @Override
     public Client client() {
-        return new CopycatTransportClient(partitionId,
-                                          messagingService,
-                                          mode);
+        return new CopycatTransportClient(partitionId, messagingService);
     }
 
     @Override
     public Server server() {
-        return new CopycatTransportServer(partitionId,
-                                          messagingService);
+        return new CopycatTransportServer(partitionId, messagingService);
+    }
+
+    @Override
+    public String toString() {
+        return toStringHelper(this).toString();
     }
 
     /**
@@ -89,7 +75,7 @@
      * @param address address
      * @return end point
      */
-    public static Endpoint toEndpoint(Address address) {
+    static Endpoint toEndpoint(Address address) {
         return EP_LOOKUP_CACHE.computeIfAbsent(address, a -> {
             try {
                 return new Endpoint(IpAddress.valueOf(InetAddress.getByName(a.host())), a.port());
@@ -105,7 +91,7 @@
      * @param endpoint end point
      * @return address
      */
-    public static Address toAddress(Endpoint endpoint) {
+    static Address toAddress(Endpoint endpoint) {
         return ADDRESS_LOOKUP_CACHE.computeIfAbsent(endpoint, ep -> {
             try {
                 InetAddress host = InetAddress.getByAddress(endpoint.host().toOctets());
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransportClient.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransportClient.java
index 913dc6c..75a75d7 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransportClient.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransportClient.java
@@ -15,54 +15,92 @@
  */
 package org.onosproject.store.primitives.impl;
 
-import static com.google.common.base.Preconditions.checkNotNull;
-
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-
-import org.apache.commons.lang.math.RandomUtils;
-import org.onosproject.cluster.PartitionId;
-import org.onosproject.store.cluster.messaging.MessagingService;
-
+import com.google.common.base.Throwables;
 import com.google.common.collect.Sets;
-
+import io.atomix.catalyst.concurrent.ThreadContext;
 import io.atomix.catalyst.transport.Address;
 import io.atomix.catalyst.transport.Client;
 import io.atomix.catalyst.transport.Connection;
-import io.atomix.catalyst.concurrent.ThreadContext;
+import io.atomix.catalyst.transport.TransportException;
+import org.onosproject.cluster.PartitionId;
+import org.onosproject.store.cluster.messaging.Endpoint;
+import org.onosproject.store.cluster.messaging.MessagingException;
+import org.onosproject.store.cluster.messaging.MessagingService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.ConnectException;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import static com.google.common.base.MoreObjects.toStringHelper;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onosproject.store.primitives.impl.CopycatTransport.CONNECT;
+import static org.onosproject.store.primitives.impl.CopycatTransport.SUCCESS;
 
 /**
- * {@link Client} implementation for {@link CopycatTransport}.
+ * Copycat transport client implementation.
  */
 public class CopycatTransportClient implements Client {
-
+    private final Logger log = LoggerFactory.getLogger(getClass());
     private final PartitionId partitionId;
+    private final String serverSubject;
     private final MessagingService messagingService;
-    private final CopycatTransport.Mode mode;
     private final Set<CopycatTransportConnection> connections = Sets.newConcurrentHashSet();
 
-    CopycatTransportClient(PartitionId partitionId, MessagingService messagingService, CopycatTransport.Mode mode) {
-        this.partitionId = checkNotNull(partitionId);
-        this.messagingService = checkNotNull(messagingService);
-        this.mode = checkNotNull(mode);
+    public CopycatTransportClient(PartitionId partitionId, MessagingService messagingService) {
+        this.partitionId = checkNotNull(partitionId, "partitionId cannot be null");
+        this.serverSubject = String.format("onos-copycat-%s", partitionId);
+        this.messagingService = checkNotNull(messagingService, "messagingService cannot be null");
     }
 
     @Override
-    public CompletableFuture<Connection> connect(Address remoteAddress) {
+    public CompletableFuture<Connection> connect(Address address) {
+        CompletableFuture<Connection> future = new CompletableFuture<>();
         ThreadContext context = ThreadContext.currentContextOrThrow();
-        CopycatTransportConnection connection = new CopycatTransportConnection(
-                nextConnectionId(),
-                CopycatTransport.Mode.CLIENT,
-                partitionId,
-                remoteAddress,
-                messagingService,
-                context);
-        if (mode == CopycatTransport.Mode.CLIENT) {
-            connection.setBidirectional();
-        }
-        connection.closeListener(c -> connections.remove(c));
-        connections.add(connection);
-        return CompletableFuture.supplyAsync(() -> connection, context.executor());
+        Endpoint endpoint = CopycatTransport.toEndpoint(address);
+
+        log.debug("Connecting to {}", address);
+
+        ByteBuffer requestBuffer = ByteBuffer.allocate(1);
+        requestBuffer.put(CONNECT);
+
+        // Send a connect request to the server to get a unique connection ID.
+        messagingService.sendAndReceive(endpoint, serverSubject, requestBuffer.array(), context.executor())
+                .whenComplete((payload, error) -> {
+                    Throwable wrappedError = error;
+                    if (error != null) {
+                        Throwable rootCause = Throwables.getRootCause(error);
+                        if (MessagingException.class.isAssignableFrom(rootCause.getClass())) {
+                            wrappedError = new TransportException(error);
+                        }
+                        log.warn("Connection to {} failed! Reason: {}", address, wrappedError);
+                        future.completeExceptionally(wrappedError);
+                    } else {
+                        // If the connection is successful, the server will send back a
+                        // connection ID indicating where to send messages for the connection.
+                        ByteBuffer responseBuffer = ByteBuffer.wrap(payload);
+                        if (responseBuffer.get() == SUCCESS) {
+                            long connectionId = responseBuffer.getLong();
+                            CopycatTransportConnection connection = new CopycatTransportConnection(
+                                    connectionId,
+                                    CopycatTransportConnection.Mode.CLIENT,
+                                    partitionId,
+                                    endpoint,
+                                    messagingService,
+                                    context);
+                            connection.closeListener(connections::remove);
+                            connections.add(connection);
+                            future.complete(connection);
+                            log.debug("Created connection {}-{} to {}", partitionId, connectionId, address);
+                        } else {
+                            log.warn("Connection to {} failed!");
+                            future.completeExceptionally(new ConnectException());
+                        }
+                    }
+                });
+        return future;
     }
 
     @Override
@@ -70,7 +108,11 @@
         return CompletableFuture.allOf(connections.stream().map(Connection::close).toArray(CompletableFuture[]::new));
     }
 
-    private long nextConnectionId() {
-        return RandomUtils.nextLong();
+    @Override
+    public String toString() {
+        return toStringHelper(this)
+                .add("partitionId", partitionId)
+                .toString();
     }
- }
+}
+
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransportConnection.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransportConnection.java
index f2752cd..9d7edd0 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransportConnection.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransportConnection.java
@@ -15,24 +15,23 @@
  */
 package org.onosproject.store.primitives.impl;
 
-import com.google.common.base.MoreObjects;
 import com.google.common.base.Throwables;
-import com.google.common.collect.Maps;
 import io.atomix.catalyst.concurrent.Listener;
 import io.atomix.catalyst.concurrent.Listeners;
 import io.atomix.catalyst.concurrent.ThreadContext;
 import io.atomix.catalyst.serializer.SerializationException;
-import io.atomix.catalyst.transport.Address;
 import io.atomix.catalyst.transport.Connection;
 import io.atomix.catalyst.transport.MessageHandler;
 import io.atomix.catalyst.transport.TransportException;
-import io.atomix.catalyst.util.Assert;
 import io.atomix.catalyst.util.reference.ReferenceCounted;
 import org.apache.commons.io.IOUtils;
 import org.onlab.util.Tools;
 import org.onosproject.cluster.PartitionId;
+import org.onosproject.store.cluster.messaging.Endpoint;
 import org.onosproject.store.cluster.messaging.MessagingException;
 import org.onosproject.store.cluster.messaging.MessagingService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -40,79 +39,64 @@
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.ByteBuffer;
 import java.util.Map;
-import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Consumer;
 
 import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onosproject.store.primitives.impl.CopycatTransport.CLOSE;
+import static org.onosproject.store.primitives.impl.CopycatTransport.FAILURE;
+import static org.onosproject.store.primitives.impl.CopycatTransport.MESSAGE;
+import static org.onosproject.store.primitives.impl.CopycatTransport.SUCCESS;
 
 /**
- * {@link Connection} implementation for CopycatTransport.
+ * Base Copycat Transport connection.
  */
 public class CopycatTransportConnection implements Connection {
-
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private final long connectionId;
+    private final String localSubject;
+    private final String remoteSubject;
+    private final PartitionId partitionId;
+    private final Endpoint endpoint;
+    private final MessagingService messagingService;
+    private final ThreadContext context;
+    private final Map<Class, InternalHandler> handlers = new ConcurrentHashMap<>();
     private final Listeners<Throwable> exceptionListeners = new Listeners<>();
     private final Listeners<Connection> closeListeners = new Listeners<>();
 
-    static final byte SUCCESS = 0x03;
-    static final byte FAILURE = 0x04;
-
-    private final long connectionId;
-    private final CopycatTransport.Mode mode;
-    private final Address remoteAddress;
-    private final MessagingService messagingService;
-    private final String outboundMessageSubject;
-    private final String inboundMessageSubject;
-    private final ThreadContext context;
-    private final Map<Class<?>, InternalHandler> handlers = Maps.newConcurrentMap();
-
-    CopycatTransportConnection(long connectionId,
-            CopycatTransport.Mode mode,
+    CopycatTransportConnection(
+            long connectionId,
+            Mode mode,
             PartitionId partitionId,
-            Address address,
+            Endpoint endpoint,
             MessagingService messagingService,
             ThreadContext context) {
         this.connectionId = connectionId;
-        this.mode = checkNotNull(mode);
-        this.remoteAddress = checkNotNull(address);
-        this.messagingService = checkNotNull(messagingService);
-        if (mode == CopycatTransport.Mode.CLIENT) {
-            this.outboundMessageSubject = String.format("onos-copycat-%s", partitionId);
-            this.inboundMessageSubject = String.format("onos-copycat-%s-%d", partitionId, connectionId);
-        } else {
-            this.outboundMessageSubject = String.format("onos-copycat-%s-%d", partitionId, connectionId);
-            this.inboundMessageSubject = String.format("onos-copycat-%s", partitionId);
-        }
-        this.context = checkNotNull(context);
-    }
-
-    public void setBidirectional() {
-        messagingService.registerHandler(inboundMessageSubject, (sender, payload) -> {
-            try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(payload))) {
-                if (input.readLong() !=  connectionId) {
-                    throw new IllegalStateException("Invalid connection Id");
-                }
-                return handle(IOUtils.toByteArray(input));
-            } catch (IOException e) {
-                Throwables.propagate(e);
-                return null;
-            }
-        });
+        this.partitionId = checkNotNull(partitionId, "partitionId cannot be null");
+        this.localSubject = mode.getLocalSubject(partitionId, connectionId);
+        this.remoteSubject = mode.getRemoteSubject(partitionId, connectionId);
+        this.endpoint = checkNotNull(endpoint, "endpoint cannot be null");
+        this.messagingService = checkNotNull(messagingService, "messagingService cannot be null");
+        this.context = checkNotNull(context, "context cannot be null");
+        messagingService.registerHandler(localSubject, this::handle);
     }
 
     @Override
     public <T, U> CompletableFuture<U> send(T message) {
         ThreadContext context = ThreadContext.currentContextOrThrow();
-        CompletableFuture<U> result = new CompletableFuture<>();
+        CompletableFuture<U> future = new CompletableFuture<>();
         try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
-            new DataOutputStream(baos).writeLong(connectionId);
+            DataOutputStream dos = new DataOutputStream(baos);
+            dos.writeByte(MESSAGE);
             context.serializer().writeObject(message, baos);
             if (message instanceof ReferenceCounted) {
                 ((ReferenceCounted<?>) message).release();
             }
-            messagingService.sendAndReceive(CopycatTransport.toEndpoint(remoteAddress),
-                                            outboundMessageSubject,
+            messagingService.sendAndReceive(endpoint,
+                                            remoteSubject,
                                             baos.toByteArray(),
                                             context.executor())
                     .whenComplete((r, e) -> {
@@ -123,20 +107,23 @@
                                 wrappedError = new TransportException(e);
                             }
                         }
-                        handleResponse(r, wrappedError, result, context);
+                        handleResponse(r, wrappedError, future);
                     });
         } catch (SerializationException | IOException e) {
-            result.completeExceptionally(e);
+            future.completeExceptionally(e);
         }
-        return result;
+        return future;
     }
 
-    private <T> void handleResponse(byte[] response,
-                                    Throwable error,
-                                    CompletableFuture<T> future,
-                                    ThreadContext context) {
+    /**
+     * Handles a response received from the other side of the connection.
+     */
+    private <T> void handleResponse(
+            byte[] response,
+            Throwable error,
+            CompletableFuture<T> future) {
         if (error != null) {
-            context.execute(() -> future.completeExceptionally(error));
+            future.completeExceptionally(error);
             return;
         }
         checkNotNull(response);
@@ -145,36 +132,54 @@
             byte status = (byte) input.read();
             if (status == FAILURE) {
                 Throwable t = context.serializer().readObject(input);
-                context.execute(() -> future.completeExceptionally(t));
+                future.completeExceptionally(t);
             } else {
-                context.execute(() -> {
-                    try {
-                        future.complete(context.serializer().readObject(input));
-                    } catch (SerializationException e) {
-                        future.completeExceptionally(e);
-                    }
-                });
+                try {
+                    future.complete(context.serializer().readObject(input));
+                } catch (SerializationException e) {
+                    future.completeExceptionally(e);
+                }
             }
         } catch (IOException e) {
-            context.execute(() -> future.completeExceptionally(e));
+            future.completeExceptionally(e);
         }
     }
 
-    @Override
-    public <T, U> Connection handler(Class<T> type, MessageHandler<T, U> handler) {
-        Assert.notNull(type, "type");
-        handlers.put(type, new InternalHandler(handler, ThreadContext.currentContextOrThrow()));
-        return null;
+    /**
+     * Handles a message sent to the connection.
+     */
+    private CompletableFuture<byte[]> handle(Endpoint sender, byte[] payload) {
+        try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(payload))) {
+            byte type = input.readByte();
+            switch (type) {
+                case MESSAGE:
+                    return handleMessage(IOUtils.toByteArray(input));
+                case CLOSE:
+                    return handleClose();
+                default:
+                    throw new IllegalStateException("Invalid message type");
+            }
+        } catch (IOException e) {
+            Throwables.propagate(e);
+            return null;
+        }
     }
 
-   public CompletableFuture<byte[]> handle(byte[] message) {
+    /**
+     * Handles a message from the other side of the connection.
+     */
+    @SuppressWarnings("unchecked")
+    private CompletableFuture<byte[]> handleMessage(byte[] message) {
         try {
             Object request = context.serializer().readObject(new ByteArrayInputStream(message));
             InternalHandler handler = handlers.get(request.getClass());
             if (handler == null) {
+                log.warn("No handler registered on connection {}-{} for type {}",
+                         partitionId, connectionId, request.getClass());
                 return Tools.exceptionalFuture(new IllegalStateException(
                         "No handler registered for " + request.getClass()));
             }
+
             return handler.handle(request).handle((result, error) -> {
                 try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
                     baos.write(error != null ? FAILURE : SUCCESS);
@@ -190,67 +195,159 @@
         }
     }
 
-    @Override
-    public Listener<Throwable> exceptionListener(Consumer<Throwable> listener) {
-        return exceptionListeners.add(listener);
+    /**
+     * Handles a close request from the other side of the connection.
+     */
+    private CompletableFuture<byte[]> handleClose() {
+        CompletableFuture<byte[]> future = new CompletableFuture<>();
+        context.executor().execute(() -> {
+            cleanup();
+            ByteBuffer responseBuffer = ByteBuffer.allocate(1);
+            responseBuffer.put(SUCCESS);
+            future.complete(responseBuffer.array());
+        });
+        return future;
     }
 
     @Override
-    public Listener<Connection> closeListener(Consumer<Connection> listener) {
-        return closeListeners.add(listener);
+    public <T, U> Connection handler(Class<T> type, MessageHandler<T, U> handler) {
+        if (log.isTraceEnabled()) {
+            log.trace("Registered handler on connection {}-{}: {}", partitionId, connectionId, type);
+        }
+        handlers.put(type, new InternalHandler(handler, ThreadContext.currentContextOrThrow()));
+        return this;
+    }
+
+    @Override
+    public Listener<Throwable> exceptionListener(Consumer<Throwable> consumer) {
+        return exceptionListeners.add(consumer);
+    }
+
+    @Override
+    public Listener<Connection> closeListener(Consumer<Connection> consumer) {
+        return closeListeners.add(consumer);
     }
 
     @Override
     public CompletableFuture<Void> close() {
-        closeListeners.forEach(listener -> listener.accept(this));
-        if (mode == CopycatTransport.Mode.CLIENT) {
-            messagingService.unregisterHandler(inboundMessageSubject);
-        }
-        return CompletableFuture.completedFuture(null);
+        log.debug("Closing connection {}-{}", partitionId, connectionId);
+
+        ByteBuffer requestBuffer = ByteBuffer.allocate(1);
+        requestBuffer.put(CLOSE);
+
+        ThreadContext context = ThreadContext.currentContextOrThrow();
+        CompletableFuture<Void> future = new CompletableFuture<>();
+        messagingService.sendAndReceive(endpoint, remoteSubject, requestBuffer.array(), context.executor())
+                .whenComplete((payload, error) -> {
+                    cleanup();
+                    Throwable wrappedError = error;
+                    if (error != null) {
+                        Throwable rootCause = Throwables.getRootCause(error);
+                        if (MessagingException.class.isAssignableFrom(rootCause.getClass())) {
+                            wrappedError = new TransportException(error);
+                        }
+                        future.completeExceptionally(wrappedError);
+                    } else {
+                        ByteBuffer responseBuffer = ByteBuffer.wrap(payload);
+                        if (responseBuffer.get() == SUCCESS) {
+                            future.complete(null);
+                        } else {
+                            future.completeExceptionally(new TransportException("Failed to close connection"));
+                        }
+                    }
+                });
+        return future;
     }
 
-    @Override
-    public int hashCode() {
-        return Objects.hash(connectionId);
+    /**
+     * Cleans up the connection, unregistering handlers registered on the MessagingService.
+     */
+    private void cleanup() {
+        log.debug("Connection {}-{} closed", partitionId, connectionId);
+        messagingService.unregisterHandler(localSubject);
+        closeListeners.accept(this);
     }
 
-    @Override
-    public boolean equals(Object other) {
-        if (!(other instanceof CopycatTransportConnection)) {
-            return false;
-        }
-        return connectionId == ((CopycatTransportConnection) other).connectionId;
+    /**
+     * Connection mode used to indicate whether this side of the connection is
+     * a client or server.
+     */
+    enum Mode {
+
+        /**
+         * Represents the client side of a bi-directional connection.
+         */
+        CLIENT {
+            @Override
+            String getLocalSubject(PartitionId partitionId, long connectionId) {
+                return String.format("onos-copycat-%s-%d-client", partitionId, connectionId);
+            }
+
+            @Override
+            String getRemoteSubject(PartitionId partitionId, long connectionId) {
+                return String.format("onos-copycat-%s-%d-server", partitionId, connectionId);
+            }
+        },
+
+        /**
+         * Represents the server side of a bi-directional connection.
+         */
+        SERVER {
+            @Override
+            String getLocalSubject(PartitionId partitionId, long connectionId) {
+                return String.format("onos-copycat-%s-%d-server", partitionId, connectionId);
+            }
+
+            @Override
+            String getRemoteSubject(PartitionId partitionId, long connectionId) {
+                return String.format("onos-copycat-%s-%d-client", partitionId, connectionId);
+            }
+        };
+
+        /**
+         * Returns the local messaging service subject for the connection in this mode.
+         * Subjects generated by the connection mode are guaranteed to be globally unique.
+         *
+         * @param partitionId the partition ID to which the connection belongs.
+         * @param connectionId the connection ID.
+         * @return the globally unique local subject for the connection.
+         */
+        abstract String getLocalSubject(PartitionId partitionId, long connectionId);
+
+        /**
+         * Returns the remote messaging service subject for the connection in this mode.
+         * Subjects generated by the connection mode are guaranteed to be globally unique.
+         *
+         * @param partitionId the partition ID to which the connection belongs.
+         * @param connectionId the connection ID.
+         * @return the globally unique remote subject for the connection.
+         */
+        abstract String getRemoteSubject(PartitionId partitionId, long connectionId);
     }
 
-    @Override
-    public String toString() {
-        return MoreObjects.toStringHelper(getClass())
-                .add("id", connectionId)
-                .toString();
-    }
-
-    @SuppressWarnings("rawtypes")
-    private final class InternalHandler {
-
+    /**
+     * Internal container for a handler/context pair.
+     */
+    private static class InternalHandler {
         private final MessageHandler handler;
         private final ThreadContext context;
 
-        private InternalHandler(MessageHandler handler, ThreadContext context) {
+        InternalHandler(MessageHandler handler, ThreadContext context) {
             this.handler = handler;
             this.context = context;
         }
 
         @SuppressWarnings("unchecked")
-        public CompletableFuture<Object> handle(Object message) {
-            CompletableFuture<Object> answer = new CompletableFuture<>();
+        CompletableFuture<Object> handle(Object message) {
+            CompletableFuture<Object> future = new CompletableFuture<>();
             context.execute(() -> handler.handle(message).whenComplete((r, e) -> {
                 if (e != null) {
-                    answer.completeExceptionally((Throwable) e);
+                    future.completeExceptionally((Throwable) e);
                 } else {
-                    answer.complete(r);
+                    future.complete(r);
                 }
             }));
-            return answer;
+            return future;
         }
     }
 }
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransportServer.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransportServer.java
index ff64112..a38dea3 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransportServer.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CopycatTransportServer.java
@@ -15,113 +15,99 @@
  */
 package org.onosproject.store.primitives.impl;
 
-import static com.google.common.base.Preconditions.checkNotNull;
-import static org.slf4j.LoggerFactory.getLogger;
-import io.atomix.catalyst.concurrent.CatalystThreadFactory;
-import io.atomix.catalyst.concurrent.SingleThreadContext;
+import com.google.common.collect.Sets;
 import io.atomix.catalyst.concurrent.ThreadContext;
 import io.atomix.catalyst.transport.Address;
 import io.atomix.catalyst.transport.Connection;
 import io.atomix.catalyst.transport.Server;
-
-import java.io.ByteArrayInputStream;
-import java.io.DataInputStream;
-import java.io.IOException;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Consumer;
-
-import org.apache.commons.io.IOUtils;
-import org.onlab.util.Tools;
+import org.apache.commons.lang3.RandomUtils;
 import org.onosproject.cluster.PartitionId;
 import org.onosproject.store.cluster.messaging.MessagingService;
 import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import com.google.common.collect.Maps;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
+import static com.google.common.base.MoreObjects.toStringHelper;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onosproject.store.primitives.impl.CopycatTransport.CONNECT;
+import static org.onosproject.store.primitives.impl.CopycatTransport.FAILURE;
+import static org.onosproject.store.primitives.impl.CopycatTransport.SUCCESS;
 
 /**
- * {@link Server} implementation for {@link CopycatTransport}.
+ * Copycat transport server implementation.
  */
 public class CopycatTransportServer implements Server {
-
-    private final Logger log = getLogger(getClass());
-    private final AtomicBoolean listening = new AtomicBoolean(false);
-    private CompletableFuture<Void> listenFuture = new CompletableFuture<>();
-    private final ScheduledExecutorService executorService;
+    private final Logger log = LoggerFactory.getLogger(getClass());
     private final PartitionId partitionId;
+    private final String serverSubject;
     private final MessagingService messagingService;
-    private final String messageSubject;
-    private final Map<Long, CopycatTransportConnection> connections = Maps.newConcurrentMap();
+    private final Set<CopycatTransportConnection> connections = Sets.newConcurrentHashSet();
 
-    CopycatTransportServer(PartitionId partitionId, MessagingService messagingService) {
-        this.partitionId = checkNotNull(partitionId);
-        this.messagingService = checkNotNull(messagingService);
-        this.messageSubject = String.format("onos-copycat-%s", partitionId);
-        this.executorService = Executors.newScheduledThreadPool(Math.min(4, Runtime.getRuntime().availableProcessors()),
-                new CatalystThreadFactory("copycat-server-p" + partitionId + "-%d"));
+    public CopycatTransportServer(PartitionId partitionId, MessagingService messagingService) {
+        this.partitionId = checkNotNull(partitionId, "partitionId cannot be null");
+        this.serverSubject = String.format("onos-copycat-%s", partitionId);
+        this.messagingService = checkNotNull(messagingService, "messagingService cannot be null");
     }
 
     @Override
-    public CompletableFuture<Void> listen(Address address, Consumer<Connection> listener) {
-        if (listening.compareAndSet(false, true)) {
-            ThreadContext context = ThreadContext.currentContextOrThrow();
-            listen(address, listener, context);
-        }
-        return listenFuture;
-    }
+    public CompletableFuture<Void> listen(Address address, Consumer<Connection> consumer) {
+        ThreadContext context = ThreadContext.currentContextOrThrow();
+        messagingService.registerHandler(serverSubject, (sender, payload) -> {
 
-    private void listen(Address address, Consumer<Connection> listener, ThreadContext context) {
-        messagingService.registerHandler(messageSubject, (sender, payload) -> {
-            try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(payload))) {
-                long connectionId = input.readLong();
-                AtomicBoolean newConnectionCreated = new AtomicBoolean(false);
-                CopycatTransportConnection connection = connections.computeIfAbsent(connectionId, k -> {
-                    newConnectionCreated.set(true);
-                    CopycatTransportConnection newConnection = new CopycatTransportConnection(connectionId,
-                            CopycatTransport.Mode.SERVER,
-                            partitionId,
-                            CopycatTransport.toAddress(sender),
-                            messagingService,
-                            getOrCreateContext(context));
-                    log.debug("Created new incoming connection {}", connectionId);
-                    newConnection.closeListener(c -> connections.remove(connectionId, c));
-                    return newConnection;
-                });
-                byte[] request = IOUtils.toByteArray(input);
-                return CompletableFuture.supplyAsync(
-                        () -> {
-                            if (newConnectionCreated.get()) {
-                                listener.accept(connection);
-                            }
-                            return connection;
-                        }, context.executor()).thenCompose(c -> c.handle(request));
-            } catch (IOException e) {
-                return Tools.exceptionalFuture(e);
+            // Only connect messages can be sent to the server. Once a connect message
+            // is received, the connection will register a separate handler for messaging.
+            ByteBuffer requestBuffer = ByteBuffer.wrap(payload);
+            if (requestBuffer.get() != CONNECT) {
+                ByteBuffer responseBuffer = ByteBuffer.allocate(1);
+                responseBuffer.put(FAILURE);
+                return CompletableFuture.completedFuture(responseBuffer.array());
             }
+
+            // Create the connection and ensure state is cleaned up when the connection is closed.
+            long connectionId = RandomUtils.nextLong();
+            CopycatTransportConnection connection = new CopycatTransportConnection(
+                    connectionId,
+                    CopycatTransportConnection.Mode.SERVER,
+                    partitionId,
+                    sender,
+                    messagingService,
+                    context);
+            connection.closeListener(connections::remove);
+            connections.add(connection);
+
+            CompletableFuture<byte[]> future = new CompletableFuture<>();
+
+            // We need to ensure the connection event is called on the Copycat thread
+            // and that the future is not completed until the Copycat server has been
+            // able to register message handlers, otherwise some messages can be received
+            // prior to any handlers being registered.
+            context.executor().execute(() -> {
+                log.debug("Created connection {}-{}", partitionId, connectionId);
+                consumer.accept(connection);
+
+                ByteBuffer responseBuffer = ByteBuffer.allocate(9);
+                responseBuffer.put(SUCCESS);
+                responseBuffer.putLong(connectionId);
+                future.complete(responseBuffer.array());
+            });
+            return future;
         });
-        context.execute(() -> {
-            listenFuture.complete(null);
-        });
+        return CompletableFuture.completedFuture(null);
     }
 
     @Override
     public CompletableFuture<Void> close() {
-        messagingService.unregisterHandler(messageSubject);
-        executorService.shutdown();
-        return CompletableFuture.completedFuture(null);
+        return CompletableFuture.allOf(connections.stream().map(Connection::close).toArray(CompletableFuture[]::new));
     }
 
-    /**
-     * Returns the current execution context or creates one.
-     */
-    private ThreadContext getOrCreateContext(ThreadContext parentContext) {
-        ThreadContext context = ThreadContext.currentContext();
-        if (context != null) {
-            return context;
-        }
-        return new SingleThreadContext(executorService, parentContext.serializer().clone());
+    @Override
+    public String toString() {
+        return toStringHelper(this)
+                .add("partitionId", partitionId)
+                .toString();
     }
 }
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StoragePartition.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StoragePartition.java
index 7ed275a..a68b793 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StoragePartition.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StoragePartition.java
@@ -131,9 +131,7 @@
         StoragePartitionServer server = new StoragePartitionServer(toAddress(localNodeId),
                 this,
                 serializer,
-                () -> new CopycatTransport(CopycatTransport.Mode.SERVER,
-                                     partition.getId(),
-                                     messagingService),
+                () -> new CopycatTransport(partition.getId(), messagingService),
                 logFolder);
         return server.open().thenRun(() -> this.server = server);
     }
@@ -150,9 +148,7 @@
         StoragePartitionServer server = new StoragePartitionServer(toAddress(localNodeId),
                 this,
                 serializer,
-                () -> new CopycatTransport(CopycatTransport.Mode.SERVER,
-                                     partition.getId(),
-                                     messagingService),
+                () -> new CopycatTransport(partition.getId(), messagingService),
                 logFolder);
         return server.join(Collections2.transform(otherMembers, this::toAddress)).thenRun(() -> this.server = server);
     }
@@ -160,9 +156,7 @@
     private CompletableFuture<StoragePartitionClient> openClient() {
         client = new StoragePartitionClient(this,
                 serializer,
-                new CopycatTransport(CopycatTransport.Mode.CLIENT,
-                                     partition.getId(),
-                                     messagingService));
+                new CopycatTransport(partition.getId(), messagingService));
         return client.open().thenApply(v -> client);
     }
 
diff --git a/core/store/primitives/src/test/java/org/onosproject/store/primitives/impl/CopycatTransportTest.java b/core/store/primitives/src/test/java/org/onosproject/store/primitives/impl/CopycatTransportTest.java
new file mode 100644
index 0000000..21b4d70
--- /dev/null
+++ b/core/store/primitives/src/test/java/org/onosproject/store/primitives/impl/CopycatTransportTest.java
@@ -0,0 +1,360 @@
+/*
+ * Copyright 2017-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.store.primitives.impl;
+
+import com.google.common.collect.Lists;
+import io.atomix.catalyst.concurrent.SingleThreadContext;
+import io.atomix.catalyst.concurrent.ThreadContext;
+import io.atomix.catalyst.transport.Address;
+import io.atomix.catalyst.transport.Client;
+import io.atomix.catalyst.transport.Server;
+import io.atomix.catalyst.transport.Transport;
+import io.atomix.copycat.protocol.ConnectRequest;
+import io.atomix.copycat.protocol.ConnectResponse;
+import io.atomix.copycat.protocol.PublishRequest;
+import io.atomix.copycat.protocol.PublishResponse;
+import io.atomix.copycat.protocol.Response;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.onlab.packet.IpAddress;
+import org.onlab.util.Tools;
+import org.onosproject.cluster.PartitionId;
+import org.onosproject.store.cluster.messaging.Endpoint;
+import org.onosproject.store.cluster.messaging.MessagingService;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+import static org.onlab.junit.TestTools.findAvailablePort;
+
+/**
+ * Copycat transport test.
+ */
+public class CopycatTransportTest {
+
+    private static final String IP_STRING = "127.0.0.1";
+
+    private Endpoint endpoint1 = new Endpoint(IpAddress.valueOf(IP_STRING), 5001);
+    private Endpoint endpoint2 = new Endpoint(IpAddress.valueOf(IP_STRING), 5002);
+
+    private MessagingService service1;
+    private MessagingService service2;
+
+    private Transport clientTransport;
+    private ThreadContext clientContext;
+
+    private Transport serverTransport;
+    private ThreadContext serverContext;
+
+    @Before
+    public void setUp() throws Exception {
+        Map<Endpoint, TestMessagingService> services = new ConcurrentHashMap<>();
+
+        endpoint1 = new Endpoint(IpAddress.valueOf("127.0.0.1"), findAvailablePort(5001));
+        service1 = new TestMessagingService(endpoint1, services);
+        clientTransport = new CopycatTransport(PartitionId.from(1), service1);
+        clientContext = new SingleThreadContext("client-test-%d", CatalystSerializers.getSerializer());
+
+        endpoint2 = new Endpoint(IpAddress.valueOf("127.0.0.1"), findAvailablePort(5003));
+        service2 = new TestMessagingService(endpoint2, services);
+        serverTransport = new CopycatTransport(PartitionId.from(1), service2);
+        serverContext = new SingleThreadContext("server-test-%d", CatalystSerializers.getSerializer());
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        if (clientContext != null) {
+            clientContext.close();
+        }
+        if (serverContext != null) {
+            serverContext.close();
+        }
+    }
+
+    /**
+     * Tests sending a message from the client side of a Copycat connection to the server side.
+     */
+    @Test
+    public void testCopycatClientConnectionSend() throws Exception {
+        Client client = clientTransport.client();
+        Server server = serverTransport.server();
+
+        CountDownLatch latch = new CountDownLatch(4);
+        CountDownLatch listenLatch = new CountDownLatch(1);
+        CountDownLatch handlerLatch = new CountDownLatch(1);
+        serverContext.executor().execute(() -> {
+            server.listen(new Address(IP_STRING, endpoint2.port()), connection -> {
+                serverContext.checkThread();
+                latch.countDown();
+                connection.handler(ConnectRequest.class, request -> {
+                    serverContext.checkThread();
+                    latch.countDown();
+                    return CompletableFuture.completedFuture(ConnectResponse.builder()
+                            .withStatus(Response.Status.OK)
+                            .withLeader(new Address(IP_STRING, endpoint2.port()))
+                            .withMembers(Lists.newArrayList(new Address(IP_STRING, endpoint2.port())))
+                            .build());
+                });
+                handlerLatch.countDown();
+            }).thenRun(listenLatch::countDown);
+        });
+
+        listenLatch.await(5, TimeUnit.SECONDS);
+
+        clientContext.executor().execute(() -> {
+            client.connect(new Address(IP_STRING, endpoint2.port())).thenAccept(connection -> {
+                clientContext.checkThread();
+                latch.countDown();
+                try {
+                    handlerLatch.await(5, TimeUnit.SECONDS);
+                } catch (InterruptedException e) {
+                    fail();
+                }
+                connection.<ConnectRequest, ConnectResponse>send(ConnectRequest.builder()
+                        .withClientId(UUID.randomUUID().toString())
+                        .build())
+                        .thenAccept(response -> {
+                            clientContext.checkThread();
+                            assertNotNull(response);
+                            assertEquals(Response.Status.OK, response.status());
+                            latch.countDown();
+                        });
+            });
+        });
+
+        latch.await(5, TimeUnit.SECONDS);
+        assertEquals(0, latch.getCount());
+    }
+
+    /**
+     * Tests sending a message from the server side of a Copycat connection to the client side.
+     */
+    @Test
+    public void testCopycatServerConnectionSend() throws Exception {
+        Client client = clientTransport.client();
+        Server server = serverTransport.server();
+
+        CountDownLatch latch = new CountDownLatch(4);
+        CountDownLatch listenLatch = new CountDownLatch(1);
+        serverContext.executor().execute(() -> {
+            server.listen(new Address(IP_STRING, endpoint2.port()), connection -> {
+                serverContext.checkThread();
+                latch.countDown();
+                serverContext.schedule(Duration.ofMillis(100), () -> {
+                    connection.<PublishRequest, PublishResponse>send(PublishRequest.builder()
+                            .withSession(1)
+                            .withEventIndex(3)
+                            .withPreviousIndex(2)
+                            .build())
+                            .thenAccept(response -> {
+                                serverContext.checkThread();
+                                assertEquals(Response.Status.OK, response.status());
+                                assertEquals(1, response.index());
+                                latch.countDown();
+                            });
+                });
+            }).thenRun(listenLatch::countDown);
+        });
+
+        listenLatch.await(5, TimeUnit.SECONDS);
+
+        clientContext.executor().execute(() -> {
+            client.connect(new Address(IP_STRING, endpoint2.port())).thenAccept(connection -> {
+                clientContext.checkThread();
+                latch.countDown();
+                connection.handler(PublishRequest.class, request -> {
+                    clientContext.checkThread();
+                    latch.countDown();
+                    assertEquals(1, request.session());
+                    assertEquals(3, request.eventIndex());
+                    assertEquals(2, request.previousIndex());
+                    return CompletableFuture.completedFuture(PublishResponse.builder()
+                            .withStatus(Response.Status.OK)
+                            .withIndex(1)
+                            .build());
+                });
+            });
+        });
+
+        latch.await(5, TimeUnit.SECONDS);
+        assertEquals(0, latch.getCount());
+    }
+
+    /**
+     * Tests closing the server side of a Copycat connection.
+     */
+    @Test
+    public void testCopycatClientConnectionClose() throws Exception {
+        Client client = clientTransport.client();
+        Server server = serverTransport.server();
+
+        CountDownLatch latch = new CountDownLatch(5);
+        CountDownLatch listenLatch = new CountDownLatch(1);
+        serverContext.executor().execute(() -> {
+            server.listen(new Address(IP_STRING, endpoint2.port()), connection -> {
+                serverContext.checkThread();
+                latch.countDown();
+                connection.closeListener(c -> {
+                    serverContext.checkThread();
+                    latch.countDown();
+                });
+            }).thenRun(listenLatch::countDown);
+        });
+
+        listenLatch.await(5, TimeUnit.SECONDS);
+
+        clientContext.executor().execute(() -> {
+            client.connect(new Address(IP_STRING, endpoint2.port())).thenAccept(connection -> {
+                clientContext.checkThread();
+                latch.countDown();
+                connection.closeListener(c -> {
+                    clientContext.checkThread();
+                    latch.countDown();
+                });
+                clientContext.schedule(Duration.ofMillis(100), () -> {
+                    connection.close().whenComplete((result, error) -> {
+                        clientContext.checkThread();
+                        latch.countDown();
+                    });
+                });
+            });
+        });
+
+        latch.await(5, TimeUnit.SECONDS);
+        assertEquals(0, latch.getCount());
+    }
+
+    /**
+     * Tests closing the server side of a Copycat connection.
+     */
+    @Test
+    public void testCopycatServerConnectionClose() throws Exception {
+        Client client = clientTransport.client();
+        Server server = serverTransport.server();
+
+        CountDownLatch latch = new CountDownLatch(5);
+        CountDownLatch listenLatch = new CountDownLatch(1);
+        serverContext.executor().execute(() -> {
+            server.listen(new Address(IP_STRING, endpoint2.port()), connection -> {
+                serverContext.checkThread();
+                latch.countDown();
+                connection.closeListener(c -> {
+                    latch.countDown();
+                });
+                serverContext.schedule(Duration.ofMillis(100), () -> {
+                    connection.close().whenComplete((result, error) -> {
+                        serverContext.checkThread();
+                        latch.countDown();
+                    });
+                });
+            }).thenRun(listenLatch::countDown);
+        });
+
+        listenLatch.await(5, TimeUnit.SECONDS);
+
+        clientContext.executor().execute(() -> {
+            client.connect(new Address(IP_STRING, endpoint2.port())).thenAccept(connection -> {
+                clientContext.checkThread();
+                latch.countDown();
+                connection.closeListener(c -> {
+                    latch.countDown();
+                });
+            });
+        });
+
+        latch.await(5, TimeUnit.SECONDS);
+        assertEquals(0, latch.getCount());
+    }
+
+    /**
+     * Custom implementation of {@code MessagingService} used for testing. Really, this should
+     * be mocked but suffices for now.
+     */
+    public static final class TestMessagingService implements MessagingService {
+        private final Endpoint endpoint;
+        private final Map<Endpoint, TestMessagingService> services;
+        private final Map<String, BiFunction<Endpoint, byte[], CompletableFuture<byte[]>>> handlers =
+                new ConcurrentHashMap<>();
+
+        TestMessagingService(Endpoint endpoint, Map<Endpoint, TestMessagingService> services) {
+            this.endpoint = endpoint;
+            this.services = services;
+            services.put(endpoint, this);
+        }
+
+        private CompletableFuture<byte[]> handle(Endpoint ep, String type, byte[] message, Executor executor) {
+            BiFunction<Endpoint, byte[], CompletableFuture<byte[]>> handler = handlers.get(type);
+            if (handler == null) {
+                return Tools.exceptionalFuture(new IllegalStateException());
+            }
+            return handler.apply(ep, message).thenApplyAsync(r -> r, executor);
+        }
+
+        @Override
+        public CompletableFuture<Void> sendAsync(Endpoint ep, String type, byte[] payload) {
+            // Unused for testing
+            return null;
+        }
+
+        @Override
+        public CompletableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload) {
+            // Unused for testing
+            return null;
+        }
+
+        @Override
+        public CompletableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload, Executor executor) {
+            TestMessagingService service = services.get(ep);
+            if (service == null) {
+                return Tools.exceptionalFuture(new IllegalStateException());
+            }
+            return service.handle(endpoint, type, payload, executor);
+        }
+
+        @Override
+        public void registerHandler(String type, BiConsumer<Endpoint, byte[]> handler, Executor executor) {
+            // Unused for testing
+        }
+
+        @Override
+        public void registerHandler(String type, BiFunction<Endpoint, byte[], byte[]> handler, Executor executor) {
+            // Unused for testing
+        }
+
+        @Override
+        public void registerHandler(String type, BiFunction<Endpoint, byte[], CompletableFuture<byte[]>> handler) {
+            handlers.put(type, handler);
+        }
+
+        @Override
+        public void unregisterHandler(String type) {
+            handlers.remove(type);
+        }
+    }
+
+}
\ No newline at end of file