initial commit

hbmqtt/broker.py

@@ -0,0 +1,85 @@
+# Copyright (c) 2015 Nicolas JOUANIN
+#
+# See the file license.txt for copying permission.
+import asyncio
+import threading
+import logging
+from hbmqtt.errors import BrokerException
+from transitions import Machine, MachineError
+
+
+class BrokerProtocol(asyncio.Protocol):
+    def connection_made(self, transport):
+        pass
+
+    def data_received(self, data):
+        pass
+
+
+class Broker:
+    states = ['new', 'starting', 'started', 'not_started', 'stopping', 'stopped', 'not_stopped', 'stopped']
+
+    def __init__(self, host='127.0.0.1', port=1883, loop = None):
+        self.logger = logging.getLogger(__name__)
+        self._init_states()
+        self.host = host
+        self.port = port
+        if loop is not None:
+            self._loop = loop
+        else:
+            self._loop = asyncio.get_event_loop()
+        self._server = None
+        self._loop_thread = None
+
+    def _init_states(self):
+        self.machine = Machine(states=Broker.states, initial='new')
+        self.machine.add_transition(trigger='start', source='new', dest='starting')
+        self.machine.add_transition(trigger='starting_fail', source='starting', dest='not_started')
+        self.machine.add_transition(trigger='starting_success', source='starting', dest='started')
+        self.machine.add_transition(trigger='restart', source='not_started', dest='starting')
+        self.machine.add_transition(trigger='shutdown', source='started', dest='stopping')
+        self.machine.add_transition(trigger='stopping_success', source='stopping', dest='stopped')
+        self.machine.add_transition(trigger='stopping_failure', source='stopping', dest='not_stopped')
+        self.machine.add_transition(trigger='abort', source='not_stopped', dest='stopped')
+        self.machine.add_transition(trigger='abort', source='not_started', dest='stopped')
+        self.machine.add_transition(trigger='start', source='stopped', dest='starting')
+
+    def start(self):
+        try:
+            self.machine.start()
+            self.logger.debug("Broker starting")
+        except MachineError as me:
+            self.logger.debug("Invalid method call at this moment: %s" % me)
+            raise BrokerException("Broker instance can't be started: %s" % me)
+
+        try:
+            self._loop_thread = threading.Thread(target=self._run_server_loop, args=(self._loop,))
+            self._loop_thread.setDaemon(True)
+            self._loop_thread.start()
+        except Exception as e:
+            self.logger.error("Broker startup failed: %s" % e)
+            self.machine.starting_fail()
+            raise BrokerException("Broker instance can't be started: %s" % e)
+
+    def shutdown(self):
+        try:
+            self.machine.shutdown()
+        except MachineError as me:
+            self.logger.debug("Invalid method call at this moment: %s" % me)
+            raise BrokerException("Broker instance can't be stopped: %s" % me)
+        self._loop.call_soon_threadsafe(self._loop.stop)
+        self._loop_thread.join()
+        self._server.close()
+        self._loop.run_until_complete(self._server.wait_closed())
+        self._loop.close()
+        self._server = None
+        self.machine.stopping_success()
+
+    def _run_server_loop(self, loop):
+        asyncio.set_event_loop(loop)
+        coro = loop.create_server(BrokerProtocol, self.host, self.port)
+        self._server = loop.run_until_complete(coro)
+        self.logger.debug("Broker listening %s:%s" % (self.host, self.port))
+        self.machine.starting_success()
+        self.logger.debug("Broker started, ready to serve")
+        loop.run_forever()

hbmqtt/errors.py

@@ -0,0 +1,6 @@
+# Copyright (c) 2015 Nicolas JOUANIN
+#
+# See the file license.txt for copying permission.
+
+class BrokerException(BaseException):
+    pass

license.txt

@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2015 Nicolas JOUANIN
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

setup.py

@@ -0,0 +1,3 @@
+# Copyright (c) 2015 Nicolas JOUANIN
+#
+# See the file license.txt for copying permission.

tests/test_broker.py

@@ -0,0 +1,14 @@
+import unittest
+import logging
+import time
+from hbmqtt.broker import Broker
+
+logging.basicConfig(level=logging.DEBUG)
+
+class TestBroker(unittest.TestCase):
+    def test_start_broker(self):
+        b = Broker()
+        b.start()
+        time.sleep(1)
+        self.assertEqual(b.machine.state, 'started')
+        b.shutdown()