Merge lp:~verterok/lalita/zmq-proxy into lp:lalita

[Guillermo Gonzalez]

ZeroMQ proxy plugin, converts lalita into a irc <-> ZeroMQ bridge (to create plugins in external processes)

Add a plugin that converts lalita into a irc <-> ZeroMQ bridge, this allow to create plugins in external processes.

[lp:~verterok/lalita/zmq-proxy]

178. By Guillermo Gonzalez on 2012-07-08

add info about zmq-proxy plugin in the docs.

[Guillermo Gonzalez (verterok)]

added info in the docs/*

[lp:~verterok/lalita/zmq-proxy]

179. By Guillermo Gonzalez on 2012-07-09

fix typo

[Facundo Batista (facundo)]

Awesome, thanks!

review: Approve

[lp:~verterok/lalita/zmq-proxy]

180. By Guillermo Gonzalez on 2012-07-09

add copyright notice

=== modified file 'docs/tutorial_en.rst'
--- docs/tutorial_en.rst	2011-04-11 02:06:18 +0000
+++ docs/tutorial_en.rst	2012-07-09 13:48:35 +0000
@@ -633,6 +633,11 @@
 - url.py: Recollects all the URLs that are mentioned on the different channels,
   and you can search them afterwards.
 
+- zmq_proxy.py: It's a IRC-ZeroMQ <http://www.zeromq.org/> proxy/bridge, that 
+  publish all lalita events to a PUB/SUB ZeroMQ socket and listen for commands
+  in other socket (in json format). See zmq_plugins/example.py for an example 
+  plugin using this.
+
 
 Advanced configuration
 ======================

=== modified file 'docs/tutorial_sp.rst'
--- docs/tutorial_sp.rst	2011-04-11 02:06:18 +0000
+++ docs/tutorial_sp.rst	2012-07-09 13:48:35 +0000
@@ -670,6 +670,10 @@
 - url.py: Va juntando todas las URLs que se van mencionando en los
   distintos canales, y luego nos permite buscar en las mismas.
 
+- zmq_proxy.py: Es un proxy/bridge entre IRC y ZeroMQ <http://www.zeromq.org/>
+  que publica todos los eventos de lalita en un socket pub/sub de ZeroMQ y 
+  escucha 'comandos' en otro socket (en formato json). 
+  Ver zmq_plugins/example.py para un ejemplo de plugin en otro proceso.
 
 Configuración avanzada
 ======================

=== modified file 'lalita.cfg.sample'
--- lalita.cfg.sample	2011-04-11 02:05:31 +0000
+++ lalita.cfg.sample	2012-07-09 13:48:35 +0000
@@ -55,6 +55,10 @@
                 # 'url.Url': { },
                 'lalita.plugins.photo.Photo': { },
                 'lalita.plugins.misc.Ping': { },
+                'zmq_proxy.ZMQPlugin':{
+                    'events_address':'tcp://127.0.0.1:9090',
+                    'bot_address':'tcp://127.0.0.1:9091',
+                },
             },
         ),
         'example': dict (

=== modified file 'lalita/core/tests/test_dispatcher.py'
--- lalita/core/tests/test_dispatcher.py	2010-06-22 00:12:42 +0000
+++ lalita/core/tests/test_dispatcher.py	2012-07-09 13:48:35 +0000
@@ -2,6 +2,7 @@
 # License: GPL v3
 # For further info, see LICENSE file
 
+import logging
 import re
 
 from collections import defaultdict
@@ -26,7 +27,7 @@
         )
 
         ircbot_factory = ircbot.IRCBotFactory(server)
-        ircbot.logger.setLevel("error")
+        ircbot.logger.setLevel(logging.ERROR)
         self.bot = ircbot.IrcBot()
         self.bot.factory = ircbot_factory
         self.bot.config = ircbot_factory.config
@@ -213,7 +214,7 @@
                             if name == name.upper()]
         for event in supported_events:
             self.disp.register(event, self.helper.f)
-            if (event in dispatcher.USER_POS and 
+            if (event in dispatcher.USER_POS and
                 dispatcher.USER_POS[event] is not None):
                 self.disp.push(event, 'user', 'channel', 'msg')
             elif (event in dispatcher.CHANNEL_POS and

=== modified file 'lalita/core/tests/test_events.py'
--- lalita/core/tests/test_events.py	2010-06-20 16:34:35 +0000
+++ lalita/core/tests/test_events.py	2012-07-09 13:48:35 +0000
@@ -2,6 +2,7 @@
 # License: GPL v3
 # For further info, see LICENSE file
 
+import logging
 
 from twisted.trial.unittest import TestCase as TwistedTestCase
 from twisted.internet import defer
@@ -21,7 +22,7 @@
     def write(*a): pass
 
 ircbot_factory = ircbot.IRCBotFactory(server)
-ircbot.logger.setLevel("error")
+ircbot.logger.setLevel(logging.ERROR)
 bot = ircbot.IrcBot()
 bot.factory = ircbot_factory
 bot.msg = lambda *a:None

=== modified file 'lalita/core/tests/test_ircbot.py'
--- lalita/core/tests/test_ircbot.py	2010-06-16 11:45:13 +0000
+++ lalita/core/tests/test_ircbot.py	2012-07-09 13:48:35 +0000
@@ -47,7 +47,7 @@
     def setUp(self):
         server["log_config"] = {}
         ircbot_factory = ircbot.IRCBotFactory(server)
-        ircbot.logger.setLevel("error")
+        ircbot.logger.setLevel(logging.ERROR)
         self.bot = bot = ircbot.IrcBot()
         bot.factory = ircbot_factory
         bot.config = ircbot_factory.config
@@ -129,7 +129,7 @@
 
     def setUp(self):
         ircbot_factory = ircbot.IRCBotFactory(server)
-        ircbot.logger.setLevel("error")
+        ircbot.logger.setLevel(logging.ERROR)
         self.bot = bot = ircbot.IrcBot()
         bot.factory = ircbot_factory
         bot.config = ircbot_factory.config
@@ -168,7 +168,7 @@
 
     def setUp(self):
         ircbot_factory = ircbot.IRCBotFactory(server.copy())
-        ircbot.logger.setLevel("error")
+        ircbot.logger.setLevel(logging.ERROR)
         self.bot = bot = ircbot.IrcBot()
         bot.factory = ircbot_factory
         bot.config = ircbot_factory.config

=== modified file 'lalita/core/tests/test_metacommands.py'
--- lalita/core/tests/test_metacommands.py	2010-01-17 19:03:28 +0000
+++ lalita/core/tests/test_metacommands.py	2012-07-09 13:48:35 +0000
@@ -4,6 +4,7 @@
 # License: GPL v3
 # For further info, see LICENSE file
 
+import logging
 import unittest
 
 from collections import defaultdict
@@ -11,7 +12,7 @@
 from lalita import dispatcher, events, ircbot
 
 ircbot_factory = ircbot.IRCBotFactory(dict(log_config="error", channels=defaultdict(lambda: {})))
-ircbot.logger.setLevel("error")
+ircbot.logger.setLevel(logging.ERROR)
 bot = ircbot.IrcBot()
 bot.factory = ircbot_factory
 bot.config = ircbot_factory.config

=== modified file 'lalita/plugins/tests/helper.py'
--- lalita/plugins/tests/helper.py	2010-03-08 13:52:37 +0000
+++ lalita/plugins/tests/helper.py	2012-07-09 13:48:35 +0000
@@ -2,6 +2,7 @@
 # License: GPL v3
 # For further info, see LICENSE file
 
+import logging
 import unittest
 
 from lalita import ircbot
@@ -27,8 +28,8 @@
             (plugin_name, config) = server_plugin
             self.test_server["plugins"] = { plugin_name: config }
 
-        self.test_server["log_config"] = { plugin_name: "error" }
-        ircbot.logger.setLevel("error")
+        self.test_server["log_config"] = { plugin_name: "ERROR" }
+        ircbot.logger.setLevel(logging.ERROR)
         ircbot_factory = ircbot.IRCBotFactory(self.test_server)
         self.bot = ircbot.IrcBot()
         self.bot.factory = ircbot_factory
@@ -61,6 +62,11 @@
         else:
             raise ValueError("The searched plugin does not exist!")
 
+    def tearDown(self):
+        if hasattr(self, 'bot') and hasattr(self.bot, 'dispatcher'):
+            self.bot.dispatcher.shutdown()
+
+
     def assertMessageInAnswer(self, message_idx, expected):
         """assert the content of message with index: message_idx in self.answer
         and handle unexpected errors properly."""

=== added file 'lalita/plugins/tests/test_zmq_proxy.py'
--- lalita/plugins/tests/test_zmq_proxy.py	1970-01-01 00:00:00 +0000
+++ lalita/plugins/tests/test_zmq_proxy.py	2012-07-09 13:48:35 +0000
@@ -0,0 +1,204 @@
+# -*- coding: utf8 -*-
+
+# Copyright 2010 laliputienses
+# License: GPL v3
+# For further info, see LICENSE file
+import json
+import time
+
+from twisted.trial.unittest import TestCase as TwistedTestCase
+from twisted.internet import defer, reactor, error
+
+from lalita import events
+from lalita.core.dispatcher import USER_POS, CHANNEL_POS
+
+from .helper import PluginTest
+from lalita.plugins.zmq_proxy import EVENT_MAP, PluginProcess
+
+try:
+    import zmq, txzmq
+    zmq_available = True
+except ImportError:
+    zmq_available = False
+
+
+class TestZMQPlugin(TwistedTestCase, PluginTest):
+
+    if not zmq_available:
+        skip = "pyzmq and txzmq required."
+
+    def setUp(self):
+        super(TestZMQPlugin, self).setUp()
+        self.init(server_plugin=("lalita.plugins.zmq_proxy.ZMQPlugin",
+                                 {"events_address":"inproc://pub_addr",
+                                  "bot_address":"inproc://sub_addr"}))
+        self.ctx = zmq.Context.instance()
+        self.sub = self.ctx.socket(zmq.SUB)
+        while True:
+            try:
+                self.sub.connect("inproc://pub_addr")
+            except zmq.ZMQError:
+                continue
+            else:
+                break
+        self.sub.setsockopt(zmq.SUBSCRIBE, "")
+        self.cmd = self.ctx.socket(zmq.PUB)
+        while True:
+            try:
+                self.cmd.connect("inproc://sub_addr")
+            except zmq.ZMQError:
+                time.sleep(0.2)
+                continue
+            else:
+                break
+
+    def tearDown(self):
+        self.sub.close()
+        self.cmd.close()
+        self.plugin.shutdown()
+        return super(TestZMQPlugin, self).tearDown()
+
+    @defer.inlineCallbacks
+    def test_say_action(self):
+        """Say something via zmq."""
+        called = []
+        self.patch(self.plugin, 'say', lambda *a: called.append(a))
+        msg = {'action':'say', 'to_whom':"channel", "msg":"hola", "args":[]}
+        self.cmd.send(json.dumps(msg))
+        d = defer.Deferred()
+        reactor.callLater(0.2, lambda: d.callback(None))
+        yield d
+        self.assertEqual(called[0], (msg['to_whom'].encode('utf-8'), msg['msg'].decode("utf-8")))
+
+    def test_event_handler(self):
+        """Handle all events."""
+        called = []
+        self.patch(self.plugin, 'publish', lambda *a: called.append(a))
+        tested_events = {
+            events.CONNECTION_MADE:(),
+            events.CONNECTION_LOST:(),
+            events.SIGNED_ON:(),
+            events.JOINED:("channel",),
+            events.PRIVATE_MESSAGE:("channel",),
+            events.TALKED_TO_ME:("channel", "user"),
+            events.PUBLIC_MESSAGE:("channel", "user"),
+            events.ACTION:("channel", "user"),
+            events.JOIN:("channel", "user"),
+            events.LEFT:("channel", "user"),
+            events.QUIT:("user",),
+            events.KICK:("channel", "user"),
+            events.RENAME:("user",)
+        }
+        for event, fixed_args in tested_events.items():
+            msg = "a message"
+            args = fixed_args + (msg,)
+            self.disp.push(event, *args)
+            kwargs = {}
+            if CHANNEL_POS[event] is not None:
+                kwargs['channel'] = args[CHANNEL_POS[event]]
+            if USER_POS[event] is not None:
+                kwargs['user'] = args[USER_POS[event]]
+            kwargs['args'] = [msg]
+            self.assertIn((EVENT_MAP[event][0], kwargs), called)
+
+
+class TestPluginProcess(TwistedTestCase, PluginTest):
+    """Tests for PluginProcess."""
+
+    class TestPlugin(PluginProcess):
+        def __init__(self, addr1, addr2, called, config=None):
+            self.called = called
+            super(TestPluginProcess.TestPlugin, self).__init__(addr1, addr2,
+                                                               config=config)
+
+        def init(self, config):
+            self.config = config
+            self.called.append(("init", config))
+
+        def _connect(self, events_address, bot_address):
+            self.ctx = zmq.Context.instance()
+            self.sub_socket = self.ctx.socket(zmq.SUB)
+            while True:
+                try:
+                    self.sub_socket.connect(events_address)
+                except zmq.ZMQError:
+                    time.sleep(0.1)
+                    continue
+                else:
+                    break
+            self.sub_socket.setsockopt(zmq.SUBSCRIBE, "")
+            self.bot_socket = self.ctx.socket(zmq.PUB)
+            while True:
+                try:
+                    self.bot_socket.connect(bot_address)
+                except zmq.ZMQError:
+                    time.sleep(0.1)
+                    continue
+                else:
+                    break
+
+    def setUp(self):
+        super(TestPluginProcess, self).setUp()
+        events_address = "inproc://pub_addr"
+        bot_address = "inproc://sub_addr"
+        self.init(server_plugin=("lalita.plugins.zmq_proxy.ZMQPlugin",
+                                 {"events_address":events_address,
+                                  "bot_address":bot_address}))
+        self.called = []
+        try:
+            self.zmq_plugin = TestPluginProcess.TestPlugin(events_address,
+                                                           bot_address,
+                                                           self.called)
+        except Exception, e:
+            import traceback;
+            traceback.print_exc()
+
+    def tearDown(self):
+        self.zmq_plugin.bot_socket.close()
+        self.zmq_plugin.sub_socket.close()
+        self.plugin.shutdown()
+        return super(TestPluginProcess, self).tearDown()
+
+    def test_init(self):
+        """Test init is called."""
+        self.assertEquals(self.called, [("init", None)])
+
+    def test_register(self):
+        """Register to en event."""
+        matcher = lambda a: True
+        func = lambda *a: True
+        self.zmq_plugin.register(events.JOINED, func, matcher)
+        self.assertIn(events.JOINED, self.zmq_plugin._events)
+        self.assertEquals(self.zmq_plugin._events[events.JOINED], (func, matcher))
+
+    def test_register_command(self):
+        func = lambda a: None
+        self.patch(self.zmq_plugin, '_send', self.called.append)
+        self.zmq_plugin.register_command(func, "command")
+        self.assertIn("irc.command", self.zmq_plugin._events)
+        self.assertEquals(self.zmq_plugin._events["irc.command"][0][0], func)
+        self.assertIn({'action':'register_command',
+                       'command':["command"]}, self.called)
+
+    def test_register_commands(self):
+        func = lambda a: None
+        self.patch(self.zmq_plugin, '_send', self.called.append)
+        self.zmq_plugin.register_command(func, "command")
+        self.assertIn("irc.command", self.zmq_plugin._events)
+        self.assertEquals(self.zmq_plugin._events["irc.command"][0][0], func)
+        self.assertIn({'action':'register_command',
+                       'command':["command"]}, self.called)
+        func1 = lambda a: None
+        self.zmq_plugin.register_command(func1, "command1")
+        self.assertEqual(len(self.zmq_plugin._events["irc.command"]), 2)
+        self.assertEquals(self.zmq_plugin._events["irc.command"][1][0], func1)
+        self.assertIn({'action':'register_command',
+                       'command':["command1"]}, self.called)
+
+
+
+    def test_say(self):
+        self.patch(self.zmq_plugin, '_send', self.called.append)
+        self.zmq_plugin.say("me", "message")
+        self.assertIn({'action':'say', 'to_whom':'me', 'msg':"message",
+                                  'args':()}, self.called)

=== added directory 'lalita/plugins/zmq_plugins'
=== added file 'lalita/plugins/zmq_plugins/example.py'
--- lalita/plugins/zmq_plugins/example.py	1970-01-01 00:00:00 +0000
+++ lalita/plugins/zmq_plugins/example.py	2012-07-09 13:48:35 +0000
@@ -0,0 +1,52 @@
+# -*- coding: utf8 -*-
+# Copyright 2009-2012 laliputienses
+# License: GPL v3
+# For further info, see LICENSE file
+
+from lalita.plugins.zmq_proxy import PluginProcess
+
+
+class Example(PluginProcess):
+    """Example zmq-based plugin."""
+
+    def init(self, config):
+        self.logger.info("Configuring Example Plugin!")
+        # register the commands
+        self.register_command(self.cmd_example, "example")
+        self.register_command(self.cmd_example1, "example1")
+        self.register("irc.private_message", self.example_priv)
+        self.register("irc.talked_to_me", self.cmd_example)
+
+    def example_priv(self, user, command, *args):
+        """Just say something."""
+        self.say(user, "This is an example plugin.")
+
+    def cmd_example(self, user, channel, command, *args):
+        """Just say something."""
+        self.logger.debug("command %s from %s (args: %s)", command, user, args)
+        self.say(channel, "This is an example plugin.")
+
+    def cmd_example1(self, user, channel, command, *args):
+        """Just say something."""
+        self.logger.debug("command %s from %s (args: %s)", command, user, args)
+        self.say(channel, "Another example.")
+
+
+if __name__ == "__main__":
+    import optparse
+    parser = optparse.OptionParser()
+    parser.add_option("-s", "--events-address", dest="events_address",
+                      default="tcp://127.0.0.1:9090")
+    parser.add_option("-b", "--bot-address", dest="bot_address",
+                      default="tcp://127.0.0.1:9091")
+    options, args = parser.parse_args()
+    import logging
+    logging.basicConfig()
+    logging.getLogger().setLevel(logging.DEBUG)
+
+    try:
+        Example(options.events_address, options.bot_address).run()
+    except:
+        import traceback;
+        traceback.print_exc()
+

=== added file 'lalita/plugins/zmq_proxy.py'
--- lalita/plugins/zmq_proxy.py	1970-01-01 00:00:00 +0000
+++ lalita/plugins/zmq_proxy.py	2012-07-09 13:48:35 +0000
@@ -0,0 +1,210 @@
+#
+# Copyright 2009-2012 laliputienses
+# License: GPL v3
+# For further info, see LICENSE file
+#
+"""ZeroMQ proxy plugin."""
+
+import json
+import logging
+import re
+
+import zmq
+
+from txzmq.pubsub import ZmqSubConnection
+from txzmq import ZmqFactory, ZmqEndpoint
+
+from lalita import Plugin
+from lalita.core import events
+
+
+EVENT_MAP = {
+    events.CONNECTION_MADE:("irc.connection_made", None, None),
+    events.CONNECTION_LOST:("irc.connection_lost", None, None),
+    events.SIGNED_ON:("irc.signed_on", None, None),
+    events.JOINED:("irc.joined", 0, None),
+    events.PRIVATE_MESSAGE:("irc.private_message", None, 0),
+    events.TALKED_TO_ME:("irc.talked_to_me", 1, 0),
+    events.COMMAND:("irc.command", 1, 0),
+    events.PUBLIC_MESSAGE:("irc.public_message", 1, 0),
+    events.ACTION:("irc.action", 1, 0),
+    events.JOIN:("irc.join", 1, 0),
+    events.LEFT:("irc.left", 1, 0),
+    events.QUIT:("irc.quit", None, 0),
+    events.KICK:("irc.kick", 1, 0),
+    events.RENAME:("irc.rename", None, 0),
+}
+
+
+class BotConnection(ZmqSubConnection):
+    """Bot zmq connection."""
+
+    def __init__(self, plugin, factory, endpoint):
+        ZmqSubConnection.__init__(self, factory, endpoint)
+        self.plugin = plugin
+
+    def messageReceived(self, *a, **kw):
+        return ZmqSubConnection.messageReceived(self, *a, **kw)
+
+    def gotMessage(self, message):
+        info = json.loads(message)
+        if info['action'] == "say":
+            if isinstance(info['to_whom'], unicode):
+                self.plugin.say(info['to_whom'].encode('utf-8'),
+                                info['msg'], *info['args'])
+            else:
+                self.plugin.say(info['to_whom'], info['msg'], *info['args'])
+        elif info["action"] == "register_command":
+            self.plugin.register_command(info['command'])
+        else:
+            self.plugin.log.error("Invalid Action %s", message)
+
+
+class EventHandler(object):
+    """Bridge of lalita event's to zmq messages."""
+
+    def __init__(self, plugin, event_name, channel_pos, user_pos):
+        self.name = event_name
+        self.channel_pos = channel_pos
+        self.user_pos = user_pos
+        self.plugin = plugin
+        self.im_self = plugin
+        self.im_func = self.__call__
+        self.logger = plugin.logger
+
+    def __call__(self, *args):
+        self.logger.debug("event: %s", args)
+        msg_args = {}
+        if self.channel_pos is not None:
+            msg_args['channel'] = args[self.channel_pos]
+        if self.user_pos is not None:
+            msg_args['user'] = args[self.user_pos]
+        msg_args['args'] = [a for i, a in enumerate(args) \
+                            if i != self.channel_pos and i != self.user_pos]
+        self.logger.debug("publishing: %s - %s", self.name, msg_args)
+        self.plugin.publish(self.name, msg_args)
+
+
+class ZMQPlugin(Plugin):
+    """ZeroMQ plugin."""
+
+    def init(self, config):
+        self._plugins = {}
+        self.config = config
+        pub_address = self.config['events_address']
+        cmd_address = self.config['bot_address']
+        self.commands = set() # hold all the commands
+        self.ctx = zmq.Context.instance()
+        self.pub_socket = self.ctx.socket(zmq.PUB)
+        self.pub_socket.bind(pub_address)
+        # callback/command socket
+        zmq_factory = ZmqFactory(context=self.ctx)
+        rpc_endpoint = ZmqEndpoint("bind", cmd_address)
+        self.cmd_socket = BotConnection(self, zmq_factory, rpc_endpoint)
+        self.cmd_socket.subscribe("")
+        for event, info in EVENT_MAP.items():
+            if event != events.COMMAND:
+                self.register(event, EventHandler(self, *info), re.compile(".*"))
+
+    def shutdown(self):
+        if hasattr(self, 'cmd_socket'):
+            self.cmd_socket.shutdown()
+        if hasattr(self, 'pub_socket'):
+            self.pub_socket.close()
+        self.ctx.term()
+        for proc in self._plugins.values():
+            proc.loseConnection()
+
+    def register_command(self, commands):
+        """Register a list of commands."""
+        if not self.commands.intersection(set(commands)):
+            for command in commands:
+                self.commands.add(command)
+            self.register(events.COMMAND,
+                          EventHandler(self, *EVENT_MAP[events.COMMAND]),
+                          commands)
+        # else, already registered.
+
+    def publish(self, name, msg_args):
+        """Publish a message/event."""
+        self.pub_socket.send(name, zmq.SNDMORE)
+        self.pub_socket.send(json.dumps(msg_args))
+
+
+class PluginProcess(object):
+    """Base class for ZeroMQ plugins."""
+
+    def __init__(self, events_address, bot_address, config=None):
+        self.ctx = None
+        self.sub_socket = None
+        self.cmd_socket = None
+        self.config = config or {}
+        self.logger = logging.getLogger("zmq_plugin.%s" %
+                                        (self.__class__.__name__,))
+        self._connect(events_address, bot_address)
+        # setup the event handler
+        self._events = {}
+        self.init(config)
+
+    def _connect(self, events_address, bot_address):
+        self.ctx = zmq.Context.instance()
+        self.sub_socket = self.ctx.socket(zmq.SUB)
+        self.sub_socket.connect(events_address)
+        self.sub_socket.setsockopt(zmq.SUBSCRIBE, "irc")
+        # create the bot socket
+        self.bot_socket = self.ctx.socket(zmq.PUB)
+        self.bot_socket.connect(bot_address)
+
+    def init(self, config):
+        """Subclass responsability."""
+        pass
+
+    def _send(self, info):
+        self.bot_socket.send(json.dumps(info))
+
+    def register_command(self, function, command_name):
+        """Register a command."""
+        self._events.setdefault('irc.command', [])\
+                .append((function, lambda a: command_name in a))
+        new_msg = {'action':'register_command', 'command':[command_name]}
+        self._send(new_msg)
+
+    def register(self, event, function, matcher=None):
+        """Register a event handler."""
+        self._events[event] = (function, matcher)
+
+    def say(self, to_whom, msg, *args):
+        """Say something."""
+        new_msg = {'action':'say', 'to_whom':to_whom, "msg":msg, "args":args}
+        self._send(new_msg)
+
+    def run(self):
+        """Main loop"""
+        while True:
+            # block waiting for a message
+            match = False
+            event = self.sub_socket.recv()
+            payload = json.loads(self.sub_socket.recv())
+            if event == "irc.command":
+                commands = self._events[event]
+                for handler, matcher in commands:
+                    if matcher(payload['args']):
+                        match = True
+                        break
+            else:
+                match = True
+                try:
+                    handler, matcher = self._events[event]
+
+                except KeyError:
+                    self.logger.error("No handler for %s", event)
+                    continue
+            if match:
+                user = payload.get('user')
+                channel = payload.get('channel')
+                args = [a for a in [user, channel] + payload['args'] \
+                        if a is not None]
+                handler(*args)
+            else:
+                self.logger.debug("No match for %s", payload)
+