diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ConnectionManager.java b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ConnectionManager.java
new file mode 100644
index 0000000..fac3c21
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/cluster/impl/ConnectionManager.java
@@ -0,0 +1,255 @@
+package org.onlab.onos.store.cluster.impl;
+
+import org.onlab.onos.cluster.DefaultControllerNode;
+import org.onlab.onos.cluster.NodeId;
+import org.onlab.onos.store.cluster.messaging.ClusterMessage;
+import org.onlab.onos.store.cluster.messaging.ClusterMessageStream;
+import org.onlab.onos.store.cluster.messaging.HelloMessage;
+import org.onlab.onos.store.cluster.messaging.SerializationService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import static java.net.InetAddress.getByAddress;
+import static org.onlab.util.Tools.namedThreads;
+
+/**
+ * Manages connections to other controller cluster nodes.
+ */
+public class ConnectionManager implements MessageSender {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    private static final long CONNECTION_CUSTODIAN_DELAY = 1000L;
+    private static final long CONNECTION_CUSTODIAN_FREQUENCY = 5000;
+
+    private static final long START_TIMEOUT = 1000;
+    private static final int WORKERS = 3;
+
+    private ClusterConnectionListener connectionListener;
+    private List<ClusterIOWorker> workers = new ArrayList<>(WORKERS);
+
+    private final DefaultControllerNode localNode;
+    private final ClusterNodesDelegate nodesDelegate;
+    private final CommunicationsDelegate commsDelegate;
+    private final SerializationService serializationService;
+
+    // Nodes to be monitored to make sure they have a connection.
+    private final Set<DefaultControllerNode> nodes = new HashSet<>();
+
+    // Means to track message streams to other nodes.
+    private final Map<NodeId, ClusterMessageStream> streams = new ConcurrentHashMap<>();
+
+    // Executor pools for listening and managing connections to other nodes.
+    private final ExecutorService listenExecutor =
+            Executors.newSingleThreadExecutor(namedThreads("onos-comm-listen"));
+    private final ExecutorService commExecutors =
+            Executors.newFixedThreadPool(WORKERS, namedThreads("onos-comm-cluster"));
+    private final ExecutorService heartbeatExecutor =
+            Executors.newSingleThreadExecutor(namedThreads("onos-comm-heartbeat"));
+
+    private final Timer timer = new Timer("onos-comm-initiator");
+    private final TimerTask connectionCustodian = new ConnectionCustodian();
+
+    private final WorkerFinder workerFinder = new LeastUtilitiedWorkerFinder();
+
+
+    /**
+     * Creates a new connection manager.
+     */
+    ConnectionManager(DefaultControllerNode localNode,
+                      ClusterNodesDelegate nodesDelegate,
+                      CommunicationsDelegate commsDelegate,
+                      SerializationService serializationService) {
+        this.localNode = localNode;
+        this.nodesDelegate = nodesDelegate;
+        this.commsDelegate = commsDelegate;
+        this.serializationService = serializationService;
+
+        commsDelegate.setSender(this);
+        startCommunications();
+        startListening();
+        startInitiating();
+        log.info("Started");
+    }
+
+    /**
+     * Shuts down the connection manager.
+     */
+    void shutdown() {
+        connectionListener.shutdown();
+        for (ClusterIOWorker worker : workers) {
+            worker.shutdown();
+        }
+        log.info("Stopped");
+    }
+
+    /**
+     * Adds the node to the list of monitored nodes.
+     *
+     * @param node node to be added
+     */
+    void addNode(DefaultControllerNode node) {
+        nodes.add(node);
+    }
+
+    /**
+     * Removes the node from the list of monitored nodes.
+     *
+     * @param node node to be removed
+     */
+    void removeNode(DefaultControllerNode node) {
+        nodes.remove(node);
+        ClusterMessageStream stream = streams.remove(node.id());
+        if (stream != null) {
+            stream.close();
+        }
+    }
+
+    /**
+     * Removes the stream associated with the specified node.
+     *
+     * @param node node whose stream to remove
+     */
+    void removeNodeStream(DefaultControllerNode node) {
+        nodesDelegate.nodeVanished(node);
+        streams.remove(node.id());
+    }
+
+    @Override
+    public boolean send(NodeId nodeId, ClusterMessage message) {
+        ClusterMessageStream stream = streams.get(nodeId);
+        if (stream != null) {
+            try {
+                stream.write(message);
+                return true;
+            } catch (IOException e) {
+                log.warn("Unable to send a message about {} to node {}",
+                         message.subject(), nodeId);
+            }
+        }
+        return false;
+    }
+
+    /**
+     * Kicks off the IO loops and waits for them to startup.
+     */
+    private void startCommunications() {
+        HelloMessage hello = new HelloMessage(localNode.id(), localNode.ip(),
+                                              localNode.tcpPort());
+        for (int i = 0; i < WORKERS; i++) {
+            try {
+                ClusterIOWorker worker =
+                        new ClusterIOWorker(this, commsDelegate,
+                                            serializationService, hello);
+                workers.add(worker);
+                commExecutors.execute(worker);
+            } catch (IOException e) {
+                log.warn("Unable to start communication worker", e);
+            }
+        }
+
+        // Wait for the IO loops to start
+        for (ClusterIOWorker loop : workers) {
+            if (!loop.awaitStart(START_TIMEOUT)) {
+                log.warn("Comm loop did not start on-time; moving on...");
+            }
+        }
+    }
+
+    /**
+     * Starts listening for connections from peer cluster members.
+     */
+    private void startListening() {
+        try {
+            connectionListener =
+                    new ClusterConnectionListener(localNode.ip(), localNode.tcpPort(),
+                                                  workerFinder);
+            listenExecutor.execute(connectionListener);
+            if (!connectionListener.awaitStart(START_TIMEOUT)) {
+                log.warn("Listener did not start on-time; moving on...");
+            }
+        } catch (IOException e) {
+            log.error("Unable to listen for cluster connections", e);
+        }
+    }
+
+    /**
+     * Initiates open connection request and registers the pending socket
+     * channel with the given IO loop.
+     *
+     * @param loop loop with which the channel should be registered
+     * @throws java.io.IOException if the socket could not be open or connected
+     */
+    private void initiateConnection(DefaultControllerNode node,
+                                    ClusterIOWorker loop) throws IOException {
+        SocketAddress sa = new InetSocketAddress(getByAddress(node.ip().toOctets()), node.tcpPort());
+        SocketChannel ch = SocketChannel.open();
+        ch.configureBlocking(false);
+        ch.connect(sa);
+        loop.connectStream(ch);
+    }
+
+
+    /**
+     * Attempts to connect to any nodes that do not have an associated connection.
+     */
+    private void startInitiating() {
+        timer.schedule(connectionCustodian, CONNECTION_CUSTODIAN_DELAY,
+                       CONNECTION_CUSTODIAN_FREQUENCY);
+    }
+
+    // Sweeps through all controller nodes and attempts to open connection to
+    // those that presently do not have one.
+    private class ConnectionCustodian extends TimerTask {
+        @Override
+        public void run() {
+            for (DefaultControllerNode node : nodes) {
+                if (node != localNode && !streams.containsKey(node.id())) {
+                    try {
+                        initiateConnection(node, workerFinder.findWorker());
+                    } catch (IOException e) {
+                        log.debug("Unable to connect", e);
+                    }
+                }
+            }
+        }
+    }
+
+    // Finds the least utilitied IO loop.
+    private class LeastUtilitiedWorkerFinder implements WorkerFinder {
+
+        @Override
+        public ClusterIOWorker findWorker() {
+            ClusterIOWorker leastUtilized = null;
+            int minCount = Integer.MAX_VALUE;
+            for (ClusterIOWorker worker : workers) {
+                int count = worker.streamCount();
+                if (count == 0) {
+                    return worker;
+                }
+
+                if (count < minCount) {
+                    leastUtilized = worker;
+                    minCount = count;
+                }
+            }
+            return leastUtilized;
+        }
+    }
+
+}