Repository: vert-x3/vertx-eventbus-bridge-clients (language: python)

Issue:
Improve python support
#### Describe the feature

Improve official support for python

#### Use cases

Any python user wanting to use vert.x especially without using a Java server.

#### Contribution

* https://github.com/rc-dukes/vertx-eventbus-python
* https://pypi.org/project/vertx-python-client/

The docs say "There is no official support for python in Vert.x since there is no maintainer". There are the two above projects which unfortunately are not well integrated with the Vert.x eco system. Improving the integration is IMHO worthwhile.


Discussion / hints:

Fix patch:
diff --git a/python/README.md b/python/README.md
index 2276381..346b893 100644
--- a/python/README.md
+++ b/python/README.md
@@ -5,41 +5,47 @@ This is a TCP eventbus implementation for python clients.
 example:
 
 ```python
-class Client():
+from vertx import EventBus
 
-#Handler
-  def Handler(self,message):
-    if message != None:
-      print(message['body']['result'],'4');
+ebus = EventBus('localhost', 7000)
 
-eb=Vertx.Eventbus(Client(),'localhost', 7000)
+# Define a handler to handle message
+def handler(msg):
+    print("Got message from server: %s" % msg)
 
-#jsonObject -body
-body={'msg':'add 4 to 0',}
+# Connects to target bridge, it will connect automatically
+ebus.connect()
 
-#DeliveryOption
-do=Vertx.DeliveryOption();
-do.addHeader('type','text')
-do.addHeader('size','small')
-do.addReplyAddress('add')
-do.setTimeInterval(5)
+# register a handler on address `test`
+ebus.register_handler("test", handler)
 
-#register handler
-eb.registerHandler('add',Client.Handler);
+# send a json message to address: `test`, the handler will be called
+ebus.send("test", body={'name': 'python'})
 
-#send
-eb.send('add',body,do)
+# You can also send a message, and specify the reply_address
+ebus.send("echo", reply_address="test", body={'desc': 'send to echo, reply to test'})
+
+# publish message
+ebus.publish("publish-address", body={'name': 'python'})
+
+# unregister the handler after everything is done
+ebus.unregister_handler("test")
+
+# close the EventBus to release resources, after closed, it cannot be used anymore
+ebus.close()
 
-#close after 5 seconds
-eb.closeConnection(5)
 ```
 
-## Install locally
+## Build and install locally
 
 ```
 # build the package
-python setup.py sdist
+./build.sh --dist
 
 # With the tar file you can install it anywhere
-pip install --user path_to_tarfile.tar.gz
+pip install dist/vertx-eventbus-client*.tar.gz
+
+# Install from PyPI:
+pip install vertx-eventbus-client
 ```
+
diff --git a/python/Vertx/DeliveryOption.py b/python/Vertx/DeliveryOption.py
deleted file mode 100644
index 5506485..0000000
--- a/python/Vertx/DeliveryOption.py
+++ /dev/null
@@ -1,29 +0,0 @@
-import json
-
-# DeliveryOption constructor
-#	inside parameters
-#		1) replyAddress
-#		2) headers
-#		3) timeInterval for reply
-
-
-class DeliveryOption:
-    ('deliveryOption class describe headers and replyAddress')
-    replyAddress = None
-    headers = {}
-    timeInterval = 10.0
-
-    def addHeader(self, header, value):
-        self.headers[header] = value
-
-    def deleteHeader(self, header):
-        del self.headers[header]
-
-    def addReplyAddress(self, replyAddress):
-        self.replyAddress = replyAddress
-
-    def deleteReplyAddress(self):
-        self.replyAddress = None
-
-    def setTimeInterval(self, time):
-        self.timeInterval = time
diff --git a/python/Vertx/Eventbus.py b/python/Vertx/Eventbus.py
deleted file mode 100644
index 3ee26f0..0000000
--- a/python/Vertx/Eventbus.py
+++ /dev/null
@@ -1,285 +0,0 @@
-#!/usr/bin/python
-# Jayamine Alupotha
-
-import socket
-import json
-import struct
-import time
-import types
-import threading
-
-# Eventbus constructor
-#	input parameters
-#		1) instance
-#		1) host	- String
-#		2) port	- integer(>2^10-1)
-#		3) TimeOut - float- receive TimeOut
-#		4) TimeInterval -float -sleep time
-#	inside parameters
-#		1) socket
-#		2) handlers - List<String address,array<functions/handler>>
-#		3) state -integer
-#		4) ReplyHandler - <address,function>
-#		5) writable - boolean {1: sendFrame, 0: receiving
-#
-# Eventbus state
-#	0 - not connected/failed
-#	1 - connecting
-#	2 - connected /open
-#	3 - closing
-#	4 - closed
-
-
-class Eventbus:
-    ('TCP eventbus client for python')
-    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-    Handlers = {}
-    state = 0
-    ReplyHandler = {}
-    writable = True
-
-    # constructor
-    def __init__(self, instance, host='localhost', port=7000, TimeOut=0.1, TimeInterval=10.0):
-        self.host = host
-        self.port = port
-        self.this = instance
-        if TimeOut < 0.01:
-            self.TimeOut = 0.01
-        else:
-            self.TimeOut = TimeOut
-
-        self.TimeInterval = TimeInterval
-        # connect
-        try:
-            self.state = 1
-            self.sock.connect((self.host, self.port))
-            self.sock.settimeout(self.TimeOut)
-            t1 = threading.Thread(target=self.receivingThread)
-            t1.start()
-            self.state = 2
-        except IOError as e:
-            self.printErr(1, 'SEVERE', str(e))
-        except Exception as e:
-            self.printErr('Undefined Error', 'SEVERE', str(e))
-
-# Connection send and receive---------------------------------------------
-
-    def isConnected(self):
-        if self.state is 2:
-            return True
-        return False
-
-    def sendFrame(self, message_s):
-        message = message_s.encode('utf-8')
-        frame = struct.pack('!I', len(message)) + message
-        self.sock.sendall(frame)
-
-    def receive(self):
-        try:
-            if self.state < 3:  # closing socket
-                len_str = Eventbus.sock.recv(4)
-            else:
-                return False
-            len1 = struct.unpack("!i", len_str)[0]
-            if self.state < 3:  # closing socket
-                payload = Eventbus.sock.recv(len1)
-            else:
-                return False
-            json_message = payload.decode('utf-8')
-            message = json.loads(json_message)
-            # check
-            # print(message)
-            if message['type'] == 'message':
-                # failure message
-                if 'address' not in message.keys():
-                    self.printErr('message failure', 'SEVERE', message)
-                else:
-                    try:
-                        # handlers
-                        if self.Handlers[message['address']] != None:
-                            for handler in self.Handlers[message['address']]:
-                                handler(self.this, message)
-                            self.ReplyHandler = None
-
-                    except KeyError:
-                        # replyHandler
-                        try:
-                            if self.ReplyHandler['address'] == message['address']:
-                                self.ReplyHandler['replyHandler'](
-                                    self.this, None, message)
-                                self.ReplyHandler = None
-                        except KeyError:
-                            print('no handlers for ' + message['address'])
-
-            elif message['type'] == 'err':
-                try:
-                    self.ReplyHandler['replyHandler'](self.this, message, None)
-                    self.ReplyHandler = None
-                except:
-                    pass
-            else:
-                self.printErr(2, 'SEVERE', 'Unknown type')
-            return True
-        except socket.timeout:
-            return True
-        except Exception as e:
-            # 1) close socket while thread is running
-            # 2) function error in the client code
-            self.printErr('Undefined Error', 'SEVERE', str(e))
-            return False
-            # send error message
-
-    def receivingThread(self):
-        while self.state < 3:  # 0,1,2
-            if self.writable == False:
-                if self.receive() == False:
-                    break
-
-    def closeConnection(self, timeInterval=30):
-        if self.state == 1:
-            self.sock.close()
-            return
-        self.state = 3
-        time.sleep(timeInterval)
-        try:
-            self.sock.close()
-        except Exception as e:
-            self.printErr('Undefined Error', 'SEVERE', str(e))
-        self.state = 4
-
-
-# send, receive, register, unregister ------------------------------------
-
-    # address-string
-    # body - json object
-    #deliveryOption - object
-    #replyHandler -function
-
-    def send(self, address, body=None, deliveryOption=None, replyHandler=None):
-        if self.isConnected() is True:
-            message = None
-
-            if callable(deliveryOption) == True:
-                replyHandler = deliveryOption
-                deliveryOption = None
-
-            if deliveryOption != None:
-                headers = deliveryOption.headers
-                replyAddress = deliveryOption.replyAddress
-                timeInterval = deliveryOption.timeInterval
-            else:
-                headers = None
-                replyAddress = None
-                timeInterval = self.TimeInterval
-
-            message = json.dumps({'type': 'send', 'address': address,
-                                  'replyAddress': replyAddress, 'headers': headers, 'body': body, })
-            # print('sent'+message)
-
-            self.writable = True
-            self.sendFrame(message)
-
-            # replyHandler
-            if replyAddress != None and replyHandler != None:
-                self.ReplyHandler = {}
-                self.ReplyHandler['address'] = replyAddress
-                self.ReplyHandler['replyHandler'] = replyHandler
-            self.writable = False
-            # time.sleep(timeInterval)
-            i = 0.0
-            while timeInterval / self.TimeOut >= i:
-                time.sleep(self.TimeOut)
-                if self.ReplyHandler == None:
-                    break
-                if timeInterval / self.TimeOut == i and self.ReplyHandler != None:
-                    try:
-                        self.ReplyHandler['replyHandler'](
-                            self.this, 'Time Out Error', None)
-                        self.ReplyHandler = None
-                        break
-                    except:
-                        break
-                i += 1.0
-        else:
-            self.printErr(3, 'SEVERE', 'INVALID_STATE_ERR')
-
-    # address-string
-    # body - json object
-    #deliveryOption -object
-    def publish(self, address, body, deliveryOption=None):
-
-        if self.isConnected() is True:
-            if deliveryOption != None:
-                headers = deliveryOption.headers
-            else:
-                headers = None
-
-            message = json.dumps(
-                    {'type': 'publish', 'address': address, 'headers': headers, 'body': body, })
-            self.writable = True
-            self.sendFrame(message)
-            self.writable = False
-
-        else:
-            self.printErr(3, 'SEVERE', 'INVALID_STATE_ERR')
-
-    # address-string
-    #deliveryOption -object
-    #replyHandler -function
-    def registerHandler(self, address, handler):
-
-        if self.isConnected() is True:
-            message = None
-            if callable(handler) == True:
-                try:
-                    if (address not in self.Handlers.keys()) or (self.Handlers[address] == None):
-                        self.Handlers[address] = []
-                        message = json.dumps(
-                            {'type': 'register', 'address': address, })
-                        self.writable = True
-                        self.sendFrame(message)
-                        self.writable = False
-                        time.sleep(self.TimeOut)
-                except KeyError:
-                    self.Handlers[address] = []
-
-                try:
-                    self.Handlers[address].append(handler)
-                except Exception as e:
-                    self.printErr(
-                        4, 'SEVERE', 'Registration failed\n' + str(e))
-            else:
-                self.printErr(
-                    4, 'SEVERE', 'Registration failed. Function is not callable\n')
-        else:
-            self.printErr(3, 'SEVERE', 'INVALID_STATE_ERR')
-
-    # address-string
-    #deliveryOption -object
-    #replyHandler -function
-    def unregisterHandler(self, address):
-        if self.isConnected() is True:
-            message = None
-            try:
-                if self.Handlers[address] != None:
-                    if len(self.Handlers) == 1:
-                        message = json.dumps(
-                            {'type': 'unregister', 'address': address, })
-                        self.sendFrame(message)
-                    del self.Handlers[address]
-            except:
-                self.printErr(5, 'SEVERE', 'Unknown address:' + address)
-
-        else:
-            print('error occured: INVALID_STATE_ERR')
-
-# Errors -----------------------------------------------------------------
-    # 1 - connection errors
-    # 2 - unknown type of the received message
-    # 3 - invalid state errors
-    # 4 - registration failed error
-    # 5 - unknown address of un-registeration
-    def printErr(self, No, Category, error):
-        print(No)
-        print(Category)
-        print (error)
diff --git a/python/Vertx/__init__.py b/python/Vertx/__init__.py
deleted file mode 100644
index 9ca69a7..0000000
--- a/python/Vertx/__init__.py
+++ /dev/null
@@ -1,1 +0,0 @@
-__all__ =["Eventbus","DeliveryOption"]
\ No newline at end of file
diff --git a/python/build.sh b/python/build.sh
index c471658..424cdad 100755
--- a/python/build.sh
+++ b/python/build.sh
@@ -24,7 +24,7 @@ done
 echo -e "Build and test the project"
 
 echo -e "Run unit tests"
-python -m unittest -v test/Unit\ testing/*.py
+python -m unittest -v test/unittesting/*.py
 
 if [ $dist -eq 1 -o $publish -eq 1 ]; then
   echo -e "Package the project"
diff --git a/python/setup.py b/python/setup.py
index ba8c671..ae762b8 100644
--- a/python/setup.py
+++ b/python/setup.py
@@ -2,19 +2,19 @@
 import setuptools
 
 pwd = path.abspath(path.dirname(__file__))
-rdm_file = path.join(pwd, 'README.md')
-with open(rdm_file, encoding='utf-8') as f:
+rdm_file = path.join(pwd, "README.md")
+with open(rdm_file, encoding="utf-8") as f:
     readme = f.read()
 
 setuptools.setup(
-    name='vertx-eventbus-client',
-    version='1.0.0.dev0',
-    packages=['Vertx'],
-    author='vertx-dev',
-    author_email='vertx-dev@googlegroups.com',
-    url='https://github.com/vert-x3/vertx-eventbus-bridge-clients/tree/master/python',
-    license='Apache Software License 2.0',
-    description='Vertx EventBus Client',
+    name="vertx-eventbus-client",
+    version="1.0.0.dev0",
+    packages=["vertx"],
+    author="vertx-dev",
+    author_email="vertx-dev@googlegroups.com",
+    url="https://github.com/vert-x3/vertx-eventbus-bridge-clients/tree/master/python",
+    license="Apache Software License 2.0",
+    description="Vertx EventBus Client",
     long_description=readme,
-    long_description_content_type='text/markdown'
+    long_description_content_type="text/markdown",
 )
diff --git a/python/vertx/__init__.py b/python/vertx/__init__.py
new file mode 100644
index 0000000..33247a8
--- /dev/null
+++ b/python/vertx/__init__.py
@@ -0,0 +1,3 @@
+from .eventbus import EventBus
+
+__all__ = ["EventBus"]
diff --git a/python/vertx/eventbus.py b/python/vertx/eventbus.py
new file mode 100644
index 0000000..d8b5b3c
--- /dev/null
+++ b/python/vertx/eventbus.py
@@ -0,0 +1,428 @@
+#!/usr/bin/python
+# Authors:
+# 2016: Jayamine Alupotha https://github.com/jaymine
+# 2020: Wolfgang Fahl https://github.com/WolfgangFahl
+# 2021: Lin Gao https://github.com/gaol
+
+from enum import IntEnum
+import errno
+import logging
+import json
+import socket
+import struct
+from threading import Thread
+import time
+import uuid
+
+LOGGER = logging.getLogger(__name__)
+
+
+def create_message(
+    msg_type="ping", address=None, headers=None, body=None, reply_address=None
+):
+    msg = {"type": msg_type}
+    if "ping" != msg_type and address is None:
+        raise Exception("address of the message must be provided")
+    if address is not None:
+        msg["address"] = address
+    if reply_address is not None:
+        msg["replyAddress"] = reply_address
+    if headers is not None:
+        msg["headers"] = headers
+    if body is not None:
+        msg["body"] = body
+    return json.dumps(msg)
+
+
+def create_err_message(address, failure_code, message):
+    """
+    message type of the error message from client to bridge is always `send`
+    """
+    if address is None or failure_code is None or message is None:
+        raise Exception("All address, failure_code and message are required.")
+    return json.dumps(
+        {
+            "type": "send",
+            "address": address,
+            "failureCode": failure_code,
+            "message": message,
+        }
+    )
+
+
+class _State(IntEnum):
+    """
+    State of EventBus Client
+    """
+
+    NEW = 0  # when created or not connected / failed
+    CONNECTING = 1  # when the client is connecting to the bridge
+    CONNECTED = 2  # when the client gets connected to the bridge
+    CLOSING = 3  # when the client is closing the connection
+    CLOSED = 4  # when the client closed the connection
+    BROKEN = 5  # when the client connection is broken
+
+
+class EventBus:
+    """
+    Vert.x TCP EventBus Client for Python
+    """
+
+    def __init__(self, host="localhost", port=7000, options=None, err_handler=None):
+        """
+        EventBus Constructor
+
+        Args:
+            host(str): the host to connect to - default: 'localhost'
+            port(int): the port to use - default: 7000
+            options(dict): e.g. { ping_interval=5, timeout=60, debug=False, connect=False}
+
+        :raise:
+           :IOError: - the socket could not be opened
+           :Exception: - some other issue e.g. with starting the listening thread
+
+        """
+        self.socket = None
+        self.last_pong = None
+        self._state = _State.NEW
+        self.host = host
+        self.port = port
+        self.message_handlers = {}
+        self.options = options
+        self.timeout = 60  # socket timeout, in seconds
+        self.ping_interval = 5  # heart beat for ping/pong
+        self.reply_timeout = 60  # message reply timeout, in seconds
+        self._err_handler = err_handler
+        self.auto_connect = True
+        self.max_reconnect = 5
+        if self._err_handler is None:
+            self._err_handler = EventBus._default_err_handler
+        if options is not None:
+            if "timeout" in options:
+                self.timeout = int(options["timeout"])
+            if "reply_timeout" in options:
+                self.reply_timeout = int(options["reply_timeout"])
+            if "ping_interval" in options:
+                self.ping_interval = int(options["ping_interval"])
+            if "auto_connect" in options:
+                self.auto_connect = bool(options["auto_connect"])
+            if "max_reconnect" in options:
+                self.max_reconnect = int(options["max_reconnect"])
+            if "connect" in options and bool(options["connect"]):
+                self.connect()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self.close()
+
+    @staticmethod
+    def _default_err_handler(message):
+        LOGGER.error("Got Error Message: %s from server", message)
+
+    def connect(self):
+        if self._state == _State.CLOSED:
+            LOGGER.debug("Client has been closed")
+            return None
+        num_of_tries = self.max_reconnect if self.auto_connect else 1
+        for i in range(num_of_tries):
+            try:
+                if self._state != _State.CONNECTED:
+                    self._state = _State.CONNECTING
+                    self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+                    self.socket.settimeout(self.timeout)
+                    self.socket.connect((self.host, self.port))
+                    self._state = _State.CONNECTED
+                    receiving_thread = Thread(target=self._receive)
+                    receiving_thread.setDaemon(True)
+                    receiving_thread.start()
+                    ping_thread = Thread(target=self._ping)
+                    ping_thread.setDaemon(True)
+                    ping_thread.start()
+                    for address, message_handlers in self.message_handlers.items():
+                        if message_handlers.is_at_server():
+                            message = create_message("register", address)
+                            self._send_frame(message)
+                    break
+            except IOError as e:
+                LOGGER.warning(
+                    "Tried to connect %d times, try again.", i + 1, exc_info=e
+                )
+        else:
+            self._state = _State.CLOSED
+            raise Exception("Failed to connect after %d times try" % num_of_tries)
+
+    def _ping(self):
+        while self.is_connected():
+            try:
+                self.check_last_pong()
+                ping_message = create_message()
+                LOGGER.debug("send ping message")
+                self._send_frame(ping_message)
+                time.sleep(self.ping_interval)
+            except Exception as e:
+                LOGGER.debug("Error on sending ping message.", exc_info=e)
+
+    def check_last_pong(self):
+        # check last pong
+        if self.last_pong is not None:
+            if time.time() - self.last_pong > self.ping_interval * 2:
+                LOGGER.warning("ping/pong packet is slow")
+
+    def _receive_chunked(self, total_read=4096, step=2048):
+        bytes_recd = 0
+        chunks = []
+        while bytes_recd < total_read:
+            chunk = self.socket.recv(min(total_read - bytes_recd, step))
+            if chunk == b"":
+                return chunk
+            chunks.append(chunk)
+            bytes_recd = bytes_recd + len(chunk)
+        return b"".join(chunks)
+
+    def _receive(self):
+        """
+        This method gets running in receiving thread
+        """
+        while self.is_connected():
+            try:
+                len_str = self._receive_chunked(4, 4)
+                if len_str == b"":
+                    self._state = _State.CLOSED
+                    break
+                len1 = struct.unpack("!i", len_str)[0]
+                payload = self._receive_chunked(len1)
+                if payload == b"":
+                    self._state = _State.CLOSED
+                    break
+                json_message = payload.decode("utf-8")
+                message = json.loads(json_message)
+                if message["type"] == "message":  # message
+                    if "address" not in message:
+                        self._err_handler(message)
+                    else:
+                        if message["address"] in self.message_handlers:
+                            for message_handler in self.message_handlers[
+                                message["address"]
+                            ].all_message_handlers():
+                                message_handler.handle(message)
+                        else:
+                            LOGGER.warning(
+                                "No handler found on address %s", message["address"]
+                            )
+                elif message["type"] == "err":  # err
+                    self._err_handler(message)
+                elif message["type"] == "pong":  # ping/pong
+                    self.last_pong = time.time()
+                    LOGGER.debug("get pong response")
+                else:  # unknown message type
+                    self._err_handler(message)
+            except socket.timeout:
+                LOGGER.debug("timeout, try again")
+                continue
+            except Exception as e:
+                if self._state == _State.CLOSED:
+                    LOGGER.debug("client has been closed")
+                else:
+                    if e.args[0] == errno.ECONNRESET:
+                        self._state = _State.CLOSED
+                        self.message_handlers.clear()
+                        LOGGER.debug("connection reset by server")
+                    else:
+                        self._state = _State.BROKEN
+                        LOGGER.error("Connection was broken", exc_info=e)
+                break
+        if self.auto_connect and self._state != _State.CLOSED:
+            self.connect()
+
+    def handlers(self):
+        return self.message_handlers.copy()
+
+    def is_connected(self):
+        return self._state == _State.CONNECTED
+
+    def close(self):
+        if self._state != _State.CLOSED:
+            try:
+                self._state = _State.CLOSING
+                self.socket.close()
+                self._state = _State.CLOSED
+                self.message_handlers.clear()
+            except Exception as e:
+                LOGGER.error("Failed to close the socket", exc_info=e)
+
+    def _check_closed(self):
+        if not self.is_connected():
+            if self.auto_connect and self._state != _State.CLOSED:
+                self.connect()
+            else:
+                raise Exception("socket has been closed.")
+
+    # send, receive, register, unregister ------------------------------------
+
+    def _send_frame(self, message_s):
+        message = message_s.encode("utf-8")
+        frame = struct.pack("!I", len(message)) + message
+        self.socket.sendall(frame)
+
+    def send(
+        self, address, headers=None, body=None, reply_address=None, reply_handler=None
+    ):
+        self._check_closed()
+        the_reply_addr = reply_address
+        if reply_handler is not None:
+            the_reply_addr = reply_address or str(uuid.uuid1())
+            self._register_local(the_reply_addr, reply_handler, False)
+        message = create_message("send", address, headers, body, the_reply_addr)
+        self._send_frame(message)
+        if reply_handler is not None:
+            if self.message_handlers[the_reply_addr].check_time_out(
+                reply_handler, self.reply_timeout, 0.1
+            ):
+                self._err_handler("reply time out")
+            self.unregister_handler(the_reply_addr, reply_handler)
+
+    def publish(self, address, headers=None, body=None):
+        self._check_closed()
+        message = create_message("publish", address, headers, body)
+        self._send_frame(message)
+
+    def _register_local(self, address, handler, at_server=True):
+        message_handler = _MessageHandler(handler)
+        if address in self.message_handlers:
+            self.message_handlers[address].append_message_handler(
+                message_handler, at_server
+            )
+        else:
+            self.message_handlers[address] = _MessageHandlers(
+                message_handler, at_server
+            )
+
+    def _address_registered_at_server(self, address):
+        return (
+            address in self.message_handlers
+            and self.message_handlers[address].is_at_server()
+        )
+
+    def register_handler(self, address, handler):
+        """
+        Registers a handler on the address
+        
+        :param address: the address on which a handler gets registered
+        :param handler: the handler to register
+        """
+        if callable(handler):
+            if not self._address_registered_at_server(address):
+                try:
+                    self._check_closed()
+                    message = create_message("register", address)
+                    self._send_frame(message)
+                except Exception as e:
+                    LOGGER.error("Registration failed", exc_info=e)
+                    raise e
+            self._register_local(address, handler, True)
+        else:
+            raise Exception("Registration failed. Function is not callable")
+
+    def unregister_handler(self, address, handler=None):
+        """
+        Un-registers handlers with the address, if handler is not specified, all handlers with same address will be
+        cleared
+        
+        :param address: the address of the handlers
+        :param handler: the optional handler to be un-registered
+        """
+        if address in self.message_handlers:
+            message_handlers = self.message_handlers[address]
+            if handler is None:
+                message_handlers.clear()
+                del self.message_handlers[address]
+            else:
+                message_handlers.del_handler(handler)
+            if message_handlers.is_at_server() and message_handlers.is_empty():
+                try:
+                    self._check_closed()
+                    message = create_message("unregister", address)
+                    self._send_frame(message)
+                except Exception as e:
+                    LOGGER.error("Unregistering failed", exc_info=e)
+                    raise e
+
+
+class _MessageHandlers:
+    """
+    Handlers that get registered in client or/and in server
+    Only one handler with same address needs to get registered at server side, other handlers with same address
+    are in client side only, once message is back, all handlers with same address will be called in sequence.
+    """
+
+    def __init__(self, message_handler, at_server=True):
+        self._message_handlers = [message_handler]
+        self.at_server = at_server
+
+    def append_message_handler(self, message_handler, at_server=True):
+        """
+        Appends the handler if it is not in the list yet, return True if it gets appended, False otherwise
+        """
+        if at_server:
+            self.at_server = True
+        if message_handler not in self._message_handlers:
+            self._message_handlers.append(message_handler)
+
+    def del_handler(self, handler):
+        self._message_handlers = list(
+            filter(lambda mh: mh.handler() != handler, self._message_handlers)
+        )
+
+    def is_at_server(self):
+        return self.at_server
+
+    def clear(self):
+        self._message_handlers = []
+
+    def is_empty(self):
+        return len(self._message_handlers) == 0
+
+    def all_message_handlers(self):
+        return self._message_handlers
+
+    def check_time_out(self, the_handler, time_out, time_out_step):
+        """
+        check reply timeout for the reply handler using send method.
+        this method blocks until the reply handler gets called or time out.
+        the reply_handler will be removed after it gets called or time out.
+        return True if the_handler does not get called after time_out seconds
+        """
+        # check to see if the handle was called, otherwise, time-out exception
+        the_message_handler = next(
+            filter(lambda mh: mh.handler() == the_handler, self._message_handlers), None
+        )
+        if the_message_handler:
+            time_left = time_out
+            while time_left > 0 and not the_message_handler.handled():
+                time.sleep(time_out_step)
+                time_left = time_left - time_out_step
+            return time_left <= 0
+        else:
+            raise Exception("No registered handler found")
+
+
+class _MessageHandler:
+    """
+    Wrapper handler to handle message
+    """
+
+    def __init__(self, handler):
+        self._handler = handler
+        self._handled = False
+
+    def handler(self):
+        return self._handler
+
+    def handle(self, message):
+        self._handler(message)
+        self._handled = True
+
+    def handled(self):
+        return self._handled


Test patch:
diff --git a/python/test/Unit testing/test_deliveryOption.py b/python/test/Unit testing/test_deliveryOption.py
deleted file mode 100644
index ce83b87..0000000
--- a/python/test/Unit testing/test_deliveryOption.py
+++ /dev/null
@@ -1,34 +0,0 @@
-import Vertx.DeliveryOption as DeliveryOption
-import unittest
-import json
-
-
-class TestDeliveryOption(unittest.TestCase):
-    ('testing DeliveryOption')
-
-    def test_addHeader(self):
-        do = DeliveryOption.DeliveryOption()
-        do.addHeader('type', 'text')
-        header = {'type': 'text', }
-        self.assertEqual(do.headers, header)
-        do.addHeader('sessionID', '1234')
-        headers = {'type': 'text', 'sessionID': '1234', }
-        self.assertEqual(do.headers, headers)
-
-    def test_deleteHeader(self):
-        do = DeliveryOption.DeliveryOption()
-        do.addHeader('type', 'text')
-        do.addHeader('sessionID', '1234')
-        do.deleteHeader('type')
-        header = {'sessionID': '1234', }
-        self.assertEqual(do.headers, header)
-
-    def test_addReplyAddress(self):
-        do = DeliveryOption.DeliveryOption()
-        do.addReplyAddress('vert.x')
-        self.assertEqual(do.replyAddress, 'vert.x')
-        do.deleteReplyAddress()
-        self.assertEqual(do.replyAddress, None)
-
-if __name__ == '__main__':
-    unittest.main()
diff --git a/python/test/Unit testing/__init__.py b/python/test/systemtesting/__init__.py
similarity index 100%
rename from python/test/Unit testing/__init__.py
rename to python/test/systemtesting/__init__.py
diff --git a/python/test/System testing/server/pom.xml b/python/test/systemtesting/server/pom.xml
similarity index 100%
rename from python/test/System testing/server/pom.xml
rename to python/test/systemtesting/server/pom.xml
diff --git a/python/test/System testing/server/src/main/java/io/first/Server.java b/python/test/systemtesting/server/src/main/java/io/first/Server.java
similarity index 100%
rename from python/test/System testing/server/src/main/java/io/first/Server.java
rename to python/test/systemtesting/server/src/main/java/io/first/Server.java
diff --git a/python/test/System testing/test-client/test_publish.py b/python/test/systemtesting/test-client/test_publish.py
similarity index 100%
rename from python/test/System testing/test-client/test_publish.py
rename to python/test/systemtesting/test-client/test_publish.py
diff --git a/python/test/System testing/test-client/test_register.py b/python/test/systemtesting/test-client/test_register.py
similarity index 100%
rename from python/test/System testing/test-client/test_register.py
rename to python/test/systemtesting/test-client/test_register.py
diff --git a/python/test/System testing/test-client/test_send.py b/python/test/systemtesting/test-client/test_send.py
similarity index 100%
rename from python/test/System testing/test-client/test_send.py
rename to python/test/systemtesting/test-client/test_send.py
diff --git a/python/test/unittesting/__init__.py b/python/test/unittesting/__init__.py
new file mode 100644
index 0000000..e69de29