From a796725057837c7bebcf4b6a831790883b943ba7 Mon Sep 17 00:00:00 2001
From: Sven Slootweg <jamsoftgamedev@gmail.com>
Date: Wed, 18 Dec 2013 17:22:13 +0100
Subject: [PATCH] Switch to pubsub for over-the-wire communication, to prevent
 memory leaks

---
 ccollectd/ccollectd | 3 ++-
 cstatsd/cstatsd     | 2 +-
 2 files changed, 3 insertions(+), 2 deletions(-)

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: