Rename incoming and outgoing ports to frontend and backend

- This refactors the `self.incoming` and `self.outgoing` sockets to
  `self.frontend` (client facing) and `self.backend` (manager facing). This
  change should make the purpose of these ports more clear for new people
  working in the code.

Resolves: #16

12 files changed, 101 insertions, 98 deletions

diff --git a/etc/eventmq.conf-dist b/etc/eventmq.conf-dist
index 43ec6f2..d492269 100644
--- a/etc/eventmq.conf-dist
+++ b/etc/eventmq.conf-dist
@@ -23,5 +23,5 @@ queues=[[50,"google"], [40,"pushes"], [10,"default"]]
 concurrent_jobs=2
 
 [publisher]
-publisher_incoming_addr=tcp://0.0.0.0:47298
-publisher_outgoing_addr=tcp://0.0.0.0:47299
\ No newline at end of file
+publisher_frontend_addr=tcp://0.0.0.0:47298
+publisher_backend_addr=tcp://0.0.0.0:47299
\ No newline at end of file
diff --git a/eventmq/conf.py b/eventmq/conf.py
index cec0249..16b1f78 100644
--- a/eventmq/conf.py
+++ b/eventmq/conf.py
@@ -73,8 +73,8 @@ WORKER_ADDR_FAILOVER = 'tcp://127.0.0.1:47290'
 ADMINISTRATIVE_ADDR = 'tcp://127.0.0.1:47293'
 
 # PubSub
-PUBLISHER_INCOMING_ADDR = 'tcp://127.0.0.1:47298'
-PUBLISHER_OUTGOING_ADDR = 'tcp://127.0.0.1:47299'
+PUBLISHER_FRONTEND_ADDR = 'tcp://127.0.0.1:47298'
+PUBLISHER_BACKEND_ADDR = 'tcp://127.0.0.1:47299'
 
 # How many jobs should the job manager concurrently handle?
 CONCURRENT_JOBS = 4
diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index ea5af07..831cee5 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -95,7 +95,7 @@ class JobManager(HeartbeatMixin, EMQPService):
         #: then telling the router that it is READY. The reply will be the unit
         #: of work.
         # Despite the name, jobs are received on this socket
-        self.outgoing = Sender(name=self.name)
+        self.frontend = Sender(name=self.name)
 
         self.poller = Poller()
 
@@ -149,8 +149,8 @@ class JobManager(HeartbeatMixin, EMQPService):
                 self.received_disconnect = True
                 continue
 
-            if events.get(self.outgoing) == POLLIN:
-                msg = self.outgoing.recv_multipart()
+            if events.get(self.frontend) == POLLIN:
+                msg = self.frontend.recv_multipart()
                 self.process_message(msg)
 
             # Call appropiate callbacks for each finished job
@@ -238,7 +238,7 @@ class JobManager(HeartbeatMixin, EMQPService):
         send the READY command upstream to indicate that JobManager is ready
         for another REQUEST message.
         """
-        sendmsg(self.outgoing, 'READY')
+        sendmsg(self.frontend, 'READY')
 
     def send_reply(self, reply, msgid):
         """
@@ -249,7 +249,7 @@ class JobManager(HeartbeatMixin, EMQPService):
              recipient (str): The recipient id for the ack
              msgid: The unique id that we are acknowledging
          """
-        sendmsg(self.outgoing, 'REPLY', [reply, msgid])
+        sendmsg(self.frontend, 'REPLY', [reply, msgid])
 
     def on_heartbeat(self, msgid, message):
         """
@@ -277,8 +277,8 @@ class JobManager(HeartbeatMixin, EMQPService):
             self._workers.append(w)
 
     def on_disconnect(self, msgid, msg):
-        sendmsg(self.outgoing, KBYE)
-        self.outgoing.unbind(conf.WORKER_ADDR)
+        sendmsg(self.frontend, KBYE)
+        self.frontend.unbind(conf.WORKER_ADDR)
         super(JobManager, self).on_disconnect(msgid, msg)
 
     def on_kbye(self, msgid, msg):
@@ -295,7 +295,7 @@ class JobManager(HeartbeatMixin, EMQPService):
 
     def sigterm_handler(self, signum, frame):
         logger.info('Shutting down..')
-        sendmsg(self.outgoing, KBYE)
+        sendmsg(self.frontend, KBYE)
 
         self.awaiting_startup_ack = False
         self.received_disconnect = True
diff --git a/eventmq/pub.py b/eventmq/pub.py
index 3918580..487e8c5 100644
--- a/eventmq/pub.py
+++ b/eventmq/pub.py
@@ -32,26 +32,26 @@ logger = logging.getLogger(__name__)
 class Pub(HeartbeatMixin):
     def __init__(self):
         self.poller = poller.Poller()
-        self.incoming = receiver.Receiver()
-        self.outgoing = publisher.Publisher()
+        self.frontend = receiver.Receiver()
+        self.backend = publisher.Publisher()
 
         self.received_disconnect = False
 
-        self.poller.register(self.incoming, poller.POLLIN)
+        self.poller.register(self.frontend, poller.POLLIN)
         return
 
     def start(self,
-              incoming_addr=conf.PUBLISHER_INCOMING_ADDR,
-              outgoing_addr=conf.PUBLISHER_OUTGOING_ADDR):
+              frontend_addr=conf.PUBLISHER_FRONTEND_ADDR,
+              backend_addr=conf.PUBLISHER_BACKEND_ADDR):
 
         self.status = STATUS.starting
 
-        self.incoming.listen(incoming_addr)
-        self.outgoing.listen(outgoing_addr)
+        self.frontend.listen(frontend_addr)
+        self.backend.listen(backend_addr)
 
         logger.info('Listening for publish requests on {}'.format(
-            incoming_addr))
-        logger.info('Listening for subscribers on {}'.format(outgoing_addr))
+            frontend_addr))
+        logger.info('Listening for subscribers on {}'.format(backend_addr))
 
         self._start_event_loop()
 
@@ -63,8 +63,8 @@ class Pub(HeartbeatMixin):
 
             events = self.poller.poll()
 
-            if events.get(self.incoming) == poller.POLLIN:
-                msg = self.incoming.recv_multipart()
+            if events.get(self.frontend) == poller.POLLIN:
+                msg = self.frontend.recv_multipart()
                 self.process_client_message(msg)
 
     def process_client_message(self, msg):
@@ -77,7 +77,7 @@ class Pub(HeartbeatMixin):
             logger.debug('Got Publish command')
             topic = msg[5]
             sub_message = msg[6]
-            logger.debug(self.outgoing.publish(topic, sub_message))
+            logger.debug(self.backend.publish(topic, sub_message))
 
         return
 
@@ -87,8 +87,8 @@ class Pub(HeartbeatMixin):
         """
         setup_logger('eventmq')
         import_settings(section='publisher')
-        self.start(incoming_addr=conf.PUBLISHER_INCOMING_ADDR,
-                   outgoing_addr=conf.PUBLISHER_OUTGOING_ADDR)
+        self.start(frontend_addr=conf.PUBLISHER_FRONTEND_ADDR,
+                   backend_addr=conf.PUBLISHER_BACKEND_ADDR)
 
 
 # Entry point for pip console scripts
diff --git a/eventmq/router.py b/eventmq/router.py
index 2943e3c..4ff3732 100644
--- a/eventmq/router.py
+++ b/eventmq/router.py
@@ -55,12 +55,15 @@ class Router(HeartbeatMixin):
 
         self.poller = poller.Poller()
 
-        self.incoming = receiver.Receiver()
-        self.outgoing = receiver.Receiver()
+        #: Port clients connect on.
+        self.frontend = receiver.Receiver()
+        #: Port job managers connect on
+        self.backend = receiver.Receiver()
+        #: Port for administrative commands
         self.administrative_socket = receiver.Receiver()
 
-        self.poller.register(self.incoming, poller.POLLIN)
-        self.poller.register(self.outgoing, poller.POLLIN)
+        self.poller.register(self.frontend, poller.POLLIN)
+        self.poller.register(self.backend, poller.POLLIN)
         self.poller.register(self.administrative_socket, poller.POLLIN)
 
         self.status = STATUS.ready
@@ -147,8 +150,8 @@ class Router(HeartbeatMixin):
         """
         self.status = STATUS.starting
 
-        self.incoming.listen(frontend_addr)
-        self.outgoing.listen(backend_addr)
+        self.frontend.listen(frontend_addr)
+        self.backend.listen(backend_addr)
         self.administrative_socket.listen(administrative_addr)
 
         self.status = STATUS.listening
@@ -169,12 +172,12 @@ class Router(HeartbeatMixin):
             now = monotonic()
             events = self.poller.poll()
 
-            if events.get(self.incoming) == poller.POLLIN:
-                msg = self.incoming.recv_multipart()
+            if events.get(self.frontend) == poller.POLLIN:
+                msg = self.frontend.recv_multipart()
                 self.process_client_message(msg)
 
-            if events.get(self.outgoing) == poller.POLLIN:
-                msg = self.outgoing.recv_multipart()
+            if events.get(self.backend) == poller.POLLIN:
+                msg = self.backend.recv_multipart()
                 self.process_worker_message(msg)
 
             if events.get(self.administrative_socket) == poller.POLLIN:
@@ -275,7 +278,7 @@ class Router(HeartbeatMixin):
         self._meta['last_sent_heartbeat'] = monotonic()
 
         for worker_id in self.workers:
-            self.send_heartbeat(self.outgoing, worker_id)
+            self.send_heartbeat(self.backend, worker_id)
 
     def send_schedulers_heartbeats(self):
         """
@@ -284,7 +287,7 @@ class Router(HeartbeatMixin):
         self._meta['last_sent_scheduler_heartbeat'] = monotonic()
 
         for scheduler_id in self.schedulers:
-            self.send_heartbeat(self.incoming, scheduler_id)
+            self.send_heartbeat(self.frontend, scheduler_id)
 
     def on_heartbeat(self, sender, msgid, msg):
         """
@@ -319,10 +322,10 @@ class Router(HeartbeatMixin):
 
         if client_type == CLIENT_TYPE.worker:
             self.add_worker(sender, queues)
-            self.send_ack(self.outgoing, sender, msgid)
+            self.send_ack(self.backend, sender, msgid)
         elif client_type == CLIENT_TYPE.scheduler:
             self.add_scheduler(sender)
-            self.send_ack(self.incoming, sender, msgid)
+            self.send_ack(self.frontend, sender, msgid)
 
     def on_reply(self, sender, msgid, msg):
         """
@@ -353,10 +356,10 @@ class Router(HeartbeatMixin):
         # Remove schedulers and send them a kbye
         logger.info("Router preparing to disconnect...")
         for scheduler in self.schedulers:
-            self.send_kbye(self.incoming, scheduler)
+            self.send_kbye(self.frontend, scheduler)
 
         self.schedulers.clear()
-        self.incoming.unbind(conf.FRONTEND_ADDR)
+        self.frontend.unbind(conf.FRONTEND_ADDR)
 
         if len(self.waiting_messages) > 0:
             logger.info("Router processing messages in queue.")
@@ -366,10 +369,10 @@ class Router(HeartbeatMixin):
                     self.process_worker_message(msg)
 
         for worker in self.workers.keys():
-            self.send_kbye(self.outgoing, worker)
+            self.send_kbye(self.backend, worker)
 
         self.workers.clear()
-        self.outgoing.unbind(conf.BACKEND_ADDR)
+        self.backend.unbind(conf.BACKEND_ADDR)
 
         # Loops event loops should check for this and break out
         self.received_disconnect = True
@@ -400,7 +403,7 @@ class Router(HeartbeatMixin):
                 msg = self.waiting_messages[queue_name].peekleft()
 
                 try:
-                    fwdmsg(self.outgoing, sender, msg)
+                    fwdmsg(self.backend, sender, msg)
                     self.waiting_messages[queue_name].popleft()
                 except exceptions.PeerGoneAwayError:
                     # Cleanup a workerg that cannot be contacted, leaving the
@@ -494,12 +497,12 @@ class Router(HeartbeatMixin):
 
             # Rebuild the message to be sent to the worker. fwdmsg will
             # properly address the message.
-            fwdmsg(self.outgoing, worker_addr, ['', constants.PROTOCOL_VERSION,
-                                                'REQUEST', msgid, ] + msg)
+            fwdmsg(self.backend, worker_addr, ['', constants.PROTOCOL_VERSION,
+                                               'REQUEST', msgid, ] + msg)
 
             self.workers[worker_addr]['available_slots'] -= 1
             # Acknowledgment of the request being submitted to the client
-            sendmsg(self.incoming, sender, 'REPLY',
+            sendmsg(self.frontend, sender, 'REPLY',
                     (msgid,))
         except exceptions.PeerGoneAwayError:
             logger.debug(
@@ -760,7 +763,7 @@ class Router(HeartbeatMixin):
             try:
                 # Strips off the client id before forwarding because the
                 # scheduler isn't expecting it.
-                fwdmsg(self.incoming, scheduler_addr, original_msg[1:])
+                fwdmsg(self.frontend, scheduler_addr, original_msg[1:])
 
             except exceptions.PeerGoneAwayError:
                 logger.debug("Scheduler {} has unexpectedly gone away. Trying "
@@ -777,7 +780,7 @@ class Router(HeartbeatMixin):
                 try:
                     # Strips off the client id before forwarding because the
                     # scheduler isn't expecting it.
-                    fwdmsg(self.incoming, scheduler_addr, original_msg[1:])
+                    fwdmsg(self.frontend, scheduler_addr, original_msg[1:])
 
                 except exceptions.PeerGoneAwayError:
                     logger.debug("Scheduler {} has unexpectedly gone away."
@@ -902,8 +905,8 @@ class Router(HeartbeatMixin):
         process receives a SIGHUP from the system.
         """
         logger.info('Caught signame %s' % signum)
-        self.incoming.unbind(conf.FRONTEND_ADDR)
-        self.outgoing.unbind(conf.BACKEND_ADDR)
+        self.frontend.unbind(conf.FRONTEND_ADDR)
+        self.backend.unbind(conf.BACKEND_ADDR)
         import_settings()
         self.start(frontend_addr=conf.FRONTEND_ADDR,
                    backend_addr=conf.BACKEND_ADDR,
diff --git a/eventmq/scheduler.py b/eventmq/scheduler.py
index 4bac781..0d953a6 100644
--- a/eventmq/scheduler.py
+++ b/eventmq/scheduler.py
@@ -56,7 +56,7 @@ class Scheduler(HeartbeatMixin, EMQPService):
         logger.info('Initializing Scheduler...')
         import_settings()
         super(Scheduler, self).__init__(*args, **kwargs)
-        self.outgoing = Sender()
+        self.frontend = Sender()
         self._redis_server = None
 
         # contains dict of 4-item lists representing cron jobs key of this
@@ -120,8 +120,8 @@ class Scheduler(HeartbeatMixin, EMQPService):
             m_now = monotonic()
             events = self.poller.poll()
 
-            if events.get(self.outgoing) == POLLIN:
-                msg = self.outgoing.recv_multipart()
+            if events.get(self.frontend) == POLLIN:
+                msg = self.frontend.recv_multipart()
                 self.process_message(msg)
 
             # TODO: distribute me!
@@ -213,7 +213,7 @@ class Scheduler(HeartbeatMixin, EMQPService):
             str: ID of the message
         """
         jobmsg = json.loads(jobmsg)
-        msgid = send_request(self.outgoing, jobmsg, queue=queue,
+        msgid = send_request(self.frontend, jobmsg, queue=queue,
                              reply_requested=True)
 
         return msgid
@@ -221,8 +221,8 @@ class Scheduler(HeartbeatMixin, EMQPService):
     def on_disconnect(self, msgid, message):
         logger.info("Received DISCONNECT request: {}".format(message))
         self._redis_server.connection_pool.disconnect()
-        sendmsg(self.outgoing, KBYE)
-        self.outgoing.unbind(conf.SCHEDULER_ADDR)
+        sendmsg(self.frontend, KBYE)
+        self.frontend.unbind(conf.SCHEDULER_ADDR)
         super(Scheduler, self).on_disconnect(msgid, message)
 
     def on_kbye(self, msgid, msg):
diff --git a/eventmq/tests/test_jobmanager.py b/eventmq/tests/test_jobmanager.py
index db82843..82fed3f 100644
--- a/eventmq/tests/test_jobmanager.py
+++ b/eventmq/tests/test_jobmanager.py
@@ -42,7 +42,7 @@ class TestCase(unittest.TestCase):
         jm = jobmanager.JobManager()
         jm.send_ready()
 
-        sndmsg_mock.assert_called_with(jm.outgoing, 'READY')
+        sndmsg_mock.assert_called_with(jm.frontend, 'READY')
 
     @mock.patch('multiprocessing.pool.Pool.close')
     @mock.patch('eventmq.jobmanager.JobManager.process_message')
@@ -55,7 +55,7 @@ class TestCase(unittest.TestCase):
                                pool_close_mock):
         jm = jobmanager.JobManager()
         maybe_send_hb_mock.return_value = False
-        poll_mock.return_value = {jm.outgoing: jobmanager.POLLIN}
+        poll_mock.return_value = {jm.frontend: jobmanager.POLLIN}
         sender_mock.return_value = [1, 2, 3]
 
         jm._start_event_loop()
@@ -106,7 +106,7 @@ class TestCase(unittest.TestCase):
         socket_mock.return_value = True
 
         jm = jobmanager.JobManager()
-        jm.outgoing.status = constants.STATUS.listening
+        jm.frontend.status = constants.STATUS.listening
         jm.on_disconnect(msgid, msg)
         self.assertTrue(jm.received_disconnect, "Did not receive disconnect.")
 
@@ -129,7 +129,7 @@ class TestCase(unittest.TestCase):
 
         jm.sigterm_handler(13231, "FRAMEY the evil frame")
 
-        sendmsg_mock.assert_called_with(jm.outgoing, constants.KBYE)
+        sendmsg_mock.assert_called_with(jm.frontend, constants.KBYE)
         self.assertFalse(jm.awaiting_startup_ack)
         self.assertTrue(jm.received_disconnect)
 
diff --git a/eventmq/tests/test_receiver.py b/eventmq/tests/test_receiver.py
index 4ee7170..49eef69 100644
--- a/eventmq/tests/test_receiver.py
+++ b/eventmq/tests/test_receiver.py
@@ -24,7 +24,7 @@ class TestCase(unittest.TestCase):
         self.zcontext = zmq.Context.instance()
 
         self.router = router.Router()
-        self.receiver = self.router.incoming
+        self.receiver = self.router.frontend
         self.sender = sender.Sender()
 
     def test_send_multipart_unicode(self):
diff --git a/eventmq/tests/test_router.py b/eventmq/tests/test_router.py
index fdb02d0..10441eb 100644
--- a/eventmq/tests/test_router.py
+++ b/eventmq/tests/test_router.py
@@ -29,16 +29,16 @@ class TestCase(unittest.TestCase):
     def setUp(self):
         self.router = router.Router(skip_signal=True)
         self.router.zcontext = mock.Mock(spec=zmq.Context)
-        self.router.incoming = mock.Mock(spec=receiver.Receiver)
-        self.router.outgoing = mock.Mock(spec=receiver.Receiver)
+        self.router.frontend = mock.Mock(spec=receiver.Receiver)
+        self.router.backend = mock.Mock(spec=receiver.Receiver)
 
     @mock.patch('eventmq.receiver.zmq.Socket.bind')
     @mock.patch('eventmq.router.Router._start_event_loop')
     def test_start(self, event_loop_mock, zsocket_bind_mock):
         # Test default args
         self.router.start()
-        self.router.incoming.listen.assert_called_with(conf.FRONTEND_ADDR)
-        self.router.outgoing.listen.assert_called_with(conf.BACKEND_ADDR)
+        self.router.frontend.listen.assert_called_with(conf.FRONTEND_ADDR)
+        self.router.backend.listen.assert_called_with(conf.BACKEND_ADDR)
         self.assertEqual(self.router.status, constants.STATUS.listening)
 
         # Test invalid args
@@ -55,7 +55,7 @@ class TestCase(unittest.TestCase):
             sender_id, inform_msgid, [queues, 'worker'])
 
         self.router.send_ack.assert_called_with(
-            self.router.outgoing, sender_id, inform_msgid)
+            self.router.backend, sender_id, inform_msgid)
 
         self.router.add_worker.assert_called_with(
             sender_id, [(32, 'top'), (23, 'drop'), (12, 'shop')])
@@ -75,7 +75,7 @@ class TestCase(unittest.TestCase):
             sender_id, inform_msgid, [queues, constants.CLIENT_TYPE.worker])
 
         self.router.send_ack.assert_called_with(
-            self.router.outgoing, sender_id, inform_msgid)
+            self.router.backend, sender_id, inform_msgid)
         self.router.add_worker.assert_called_with(
             sender_id, [(10, 'default'), ])
 
@@ -146,10 +146,10 @@ class TestCase(unittest.TestCase):
 
         generate_msgid_mock.return_value = ack_msgid
 
-        self.router.send_ack(self.router.outgoing, sender_id, orig_msgid)
+        self.router.send_ack(self.router.backend, sender_id, orig_msgid)
 
         # Verify that an ACK was sent for the INFORM
-        self.router.outgoing.send_multipart.assert_called_with(
+        self.router.backend.send_multipart.assert_called_with(
             ('ACK', ack_msgid, orig_msgid),
             constants.PROTOCOL_VERSION, _recipient_id=sender_id)
 
@@ -168,9 +168,9 @@ class TestCase(unittest.TestCase):
 
         generate_msgid_mock.return_value = msgid
 
-        self.router.send_heartbeat(self.router.incoming, recipient_id)
+        self.router.send_heartbeat(self.router.frontend, recipient_id)
 
-        self.router.incoming.send_multipart.assert_called_with(
+        self.router.frontend.send_multipart.assert_called_with(
             ('HEARTBEAT', msgid, str(ts)), constants.PROTOCOL_VERSION,
             _recipient_id=recipient_id)
 
@@ -197,8 +197,8 @@ class TestCase(unittest.TestCase):
         self.assertGreater(self.router._meta['last_sent_heartbeat'], 0)
 
         send_heartbeat_mock.assert_has_calls(
-            [mock.call(self.router.outgoing, 'w1'),
-             mock.call(self.router.outgoing, 'w2')], any_order=True)
+            [mock.call(self.router.backend, 'w1'),
+             mock.call(self.router.backend, 'w2')], any_order=True)
 
     @mock.patch('eventmq.router.Router.send_heartbeat')
     def test_send_schedulers_heartbeats(self, send_hb_mock):
@@ -215,7 +215,7 @@ class TestCase(unittest.TestCase):
 
         self.assertGreater(
             self.router._meta['last_sent_scheduler_heartbeat'], 0)
-        send_hb_mock.assert_called_with(self.router.incoming, scheduler_id)
+        send_hb_mock.assert_called_with(self.router.frontend, scheduler_id)
 
     def test_on_disconnect(self):
         self.assertFalse(self.router.received_disconnect)
@@ -245,7 +245,7 @@ class TestCase(unittest.TestCase):
 
         self.router.on_ready(worker_id, msgid, msg)
 
-        fwdmsg_mock.assert_called_with(self.router.outgoing, worker_id,
+        fwdmsg_mock.assert_called_with(self.router.backend, worker_id,
                                        waiting_msg)
 
         self.router.on_ready(worker_id, msgid + 'a', msg)
@@ -296,19 +296,19 @@ class TestCase(unittest.TestCase):
         # Forward waiting_msg1
         ready_msgid1 = 'ready23'
         self.router.on_ready(worker1_id, ready_msgid1, ['READY', ready_msgid1])
-        fwdmsg_mock.assert_called_with(self.router.outgoing, worker1_id,
+        fwdmsg_mock.assert_called_with(self.router.backend, worker1_id,
                                        waiting_msg1)
 
         # Forward waiting_msg3 -- blu is a higher priority for worker2
         ready_msgid3 = 'ready19'
         self.router.on_ready(worker2_id, ready_msgid3, ['READY', ready_msgid3])
-        fwdmsg_mock.assert_called_with(self.router.outgoing, worker2_id,
+        fwdmsg_mock.assert_called_with(self.router.backend, worker2_id,
                                        waiting_msg3)
 
         # Forward waiting_msg2
         ready_msgid2 = 'ready5'
         self.router.on_ready(worker2_id, ready_msgid2, ['READY', ready_msgid2])
-        fwdmsg_mock.assert_called_with(self.router.outgoing, worker2_id,
+        fwdmsg_mock.assert_called_with(self.router.backend, worker2_id,
                                        waiting_msg2)
 
         # There should be no keys because the code checks for their existence
@@ -348,7 +348,7 @@ class TestCase(unittest.TestCase):
 
         # Router accepts job for 1 available slot
         self.router.on_request(client_id, msgid, msg)
-        fwdmsg_mock.assert_called_with(self.router.outgoing, worker_id,
+        fwdmsg_mock.assert_called_with(self.router.backend, worker_id,
                                        ['', constants.PROTOCOL_VERSION,
                                         'REQUEST', msgid, ] + msg)
         self.assertEqual(self.router.workers[worker_id]['available_slots'], 0)
diff --git a/eventmq/tests/test_utils.py b/eventmq/tests/test_utils.py
index 03da18a..996cddb 100644
--- a/eventmq/tests/test_utils.py
+++ b/eventmq/tests/test_utils.py
@@ -128,13 +128,13 @@ class SettingsTestCase(unittest.TestCase):
 class EMQPServiceTestCase(unittest.TestCase):
 
     # pretend to be an emq socket
-    outgoing = 'some-outgoing-socket'
+    frontend = 'some-frontend-socket'
 
     def get_worker(self):
         """return an EMQPService mimicking a worker"""
         obj = classes.EMQPService()
         obj.SERVICE_TYPE = constants.CLIENT_TYPE.worker
-        obj.outgoing = self.outgoing
+        obj.frontend = self.frontend
         obj._meta = {
             'last_sent_heartbeat': 0
         }
@@ -145,7 +145,7 @@ class EMQPServiceTestCase(unittest.TestCase):
         """return an EMQPService mimicking a scheduler"""
         obj = classes.EMQPService()
         obj.SERVICE_TYPE = constants.CLIENT_TYPE.scheduler
-        obj.outgoing = self.outgoing
+        obj.frontend = self.frontend
         obj._meta = {
             'last_sent_heartbeat': 0
         }
@@ -168,7 +168,7 @@ class EMQPServiceTestCase(unittest.TestCase):
         obj.send_inform()
 
         sendmsg_mock.assert_called_with(
-            self.outgoing, 'INFORM',
+            self.frontend, 'INFORM',
             ['', constants.CLIENT_TYPE.worker])
 
     @mock.patch('eventmq.utils.classes.sendmsg')
@@ -178,7 +178,7 @@ class EMQPServiceTestCase(unittest.TestCase):
         obj.send_inform(queues="this shouldn't matter")
 
         sendmsg_mock.assert_called_with(
-            self.outgoing, 'INFORM',
+            self.frontend, 'INFORM',
             ['', constants.CLIENT_TYPE.scheduler]
         )
 
@@ -189,7 +189,7 @@ class EMQPServiceTestCase(unittest.TestCase):
         obj.send_inform(queues=([10, 'push'], [7, 'email'],
                                 [3, 'default']))
         sendmsg_mock.asert_called_with(
-            'some-outgoing-socket', 'INFORM',
+            'some-frontend-socket', 'INFORM',
             ['[10, "push"],[7, "email"],[3, "default"]',
              constants.CLIENT_TYPE.worker]
         )
diff --git a/eventmq/utils/classes.py b/eventmq/utils/classes.py
index 39c600b..b9e09fb 100644
--- a/eventmq/utils/classes.py
+++ b/eventmq/utils/classes.py
@@ -42,10 +42,10 @@ class EMQPService(object):
     Also implements utlitiy methods for managing long-running processes.
 
     To use you must define:
-     - `self.outgoing` - socket where messages can be sent to the Router
+     - `self.frontend` - socket where messages can be sent to the Router
      - `self.SERVICE_TYPE` - defines the service type for INFORM. See
        :meth:`send_inform` for more information.
-     - `self.poller` - the poller that `self.outgoing` will be using.
+     - `self.poller` - the poller that `self.frontend` will be using.
        Usually: `self.poller = eventmq.poller.Poller()`
 
     When messages are received from the router, they are processed in
@@ -99,7 +99,7 @@ class EMQPService(object):
         else:
             queues = ''
 
-        msgid = sendmsg(self.outgoing, 'INFORM', [
+        msgid = sendmsg(self.frontend, 'INFORM', [
             queues,
             self.SERVICE_TYPE
         ])
@@ -116,7 +116,7 @@ class EMQPService(object):
         also run on a reset are here.
         """
         # Look for incoming events
-        self.poller.register(self.outgoing, poller.POLLIN)
+        self.poller.register(self.frontend, poller.POLLIN)
         self.awaiting_startup_ack = False
         self.received_disconnect = False
         self.should_reset = False
@@ -132,7 +132,7 @@ class EMQPService(object):
         """
         while not self.received_disconnect:
             self.status = constants.STATUS.connecting
-            self.outgoing.connect(addr)
+            self.frontend.connect(addr)
 
             # Setting this to false is how the loop is broken and the
             # _event_loop is started.
@@ -149,8 +149,8 @@ class EMQPService(object):
                 # multiplied by 1000 to get seconds
                 events = self.poller.poll(conf.RECONNECT_TIMEOUT * 1000)
 
-                if self.outgoing in events:  # A message from the Router!
-                    msg = self.outgoing.recv_multipart()
+                if self.frontend in events:  # A message from the Router!
+                    msg = self.frontend.recv_multipart()
                     # TODO This will silently drop messages that aren't
                     # ACK/DISCONNECT
                     if msg[2] == "ACK" or msg[2] == "DISCONNECT":
@@ -176,10 +176,10 @@ class EMQPService(object):
         Resets the current connection by closing and reopening the socket
         """
         # Unregister the old socket from the poller
-        self.poller.unregister(self.outgoing)
+        self.poller.unregister(self.frontend)
 
         # Polish up a new socket to use
-        self.outgoing.rebuild()
+        self.frontend.rebuild()
 
         # Prepare the device to connect again
         self._setup()
@@ -323,7 +323,7 @@ class HeartbeatMixin(object):
             # Send a HEARTBEAT if necessary
             if now - self._meta['last_sent_heartbeat'] >= \
                conf.HEARTBEAT_INTERVAL:
-                self.send_heartbeat(self.outgoing)
+                self.send_heartbeat(self.frontend)
 
             # Do something about any missed HEARTBEAT, if we have nothing
             # waiting on the socket
diff --git a/setup.py b/setup.py
index 067b352..8e6e731 100644
--- a/setup.py
+++ b/setup.py
@@ -61,6 +61,6 @@ setup(
         'bin/emq-jobmanager',
         'bin/emq-router',
         'bin/emq-scheduler',
-        'bin/emq-pubsub'
+        'bin/emq-publisher'
     ],
 )