diff --git a/ccollectd/ccollectd b/ccollectd/ccollectd
index e44cccd..6e2129c 100755
--- a/ccollectd/ccollectd
+++ b/ccollectd/ccollectd
@@ -75,7 +75,8 @@ timers.add_timer(zmqtimer.ZmqTimer(config["heartbeat"]["interval"], heartbeat))
 
 for hostname, node in config["nodes"].iteritems():
 	boxes[hostname] = Box(privkey, PublicKey(binascii.unhexlify(node["pubkey"])))
-	grabber = ctx.socket(zmq.PULL)
+	grabber = ctx.socket(zmq.SUB)
+	grabber.setsockopt(zmq.SUBSCRIBE, "")
 	grabber.connect(node["endpoint"])
 	socket_map[grabber] = hostname
 	poller.register(grabber, zmq.POLLIN)
diff --git a/cstatsd/cstatsd b/cstatsd/cstatsd
index 1242585..b7b4e4c 100755
--- a/cstatsd/cstatsd
+++ b/cstatsd/cstatsd
@@ -23,7 +23,7 @@ box = Box(privkey, pubkey)
 collector = ctx.socket(zmq.PULL)
 collector.bind("ipc:///tmp/cstatsd")
 
-shipper = ctx.socket(zmq.PUSH)
+shipper = ctx.socket(zmq.PUB)
 shipper.bind(config["endpoint"])
 
 try: