Improve CLI tools and documentation

Author: Marco Paolini

Dockerfile

@@ -13,5 +13,5 @@ RUN python3.5 -m pip install --no-cache-dir --no-index -f wheelhouse-app --no-de
 RUN adduser --disabled-password --disabled-login --home /app --system -q app
 WORKDIR /app
 USER app
-ENTRYPOINT ["python3.5", "-m", "pushpull.cli.server"]
+ENTRYPOINT ["pushpull-server"]
 EXPOSE 8080

README.md

@@ -27,15 +27,15 @@ Websocket to message broker gateway for servers
 
 Run the websocket server:
 
-    python -m pushpull.cli.server
+    pushpull-server
 
 Run the CLI websocket client:
 
-    python -m pushpull.cli.client challenge_websocket http://localhost:8080/sock
+    pushpull-client challenge_websocket http://localhost:8080/ mario
 
 Run the CLI rabbitmq client:
 
-    python -m pushpull.cli.client challenge_amqp
+    pushpull-client challenge_amqp amqp://localhost/ mario
 
 # Build docker image
 

pushpull/amqp/client.py

@@ -7,8 +7,8 @@
 logger = logging.getLogger(__name__)
 
 
-async def challenge(name, fd_in, fd_out):
-    async with Exchanger(name, Exchanger.ROLE_APP) as (amqp_sender, amqp_receiver):
+async def challenge(url, name, fd_in, fd_out):
+    async with Exchanger(name, Exchanger.ROLE_APP, url=url) as (amqp_sender, amqp_receiver):
         sender = send_from_fd_to_amqp(fd_in, amqp_sender)
         receiver = send_from_amqp_to_fd(amqp_receiver, fd_out)
         _, pending = await asyncio.wait([sender, receiver], return_when=asyncio.FIRST_COMPLETED)

pushpull/amqp/gateway/driver_aioamqp.py

@@ -4,34 +4,25 @@
 import aioamqp
 
 from ... import config
+from .driver_base import ExchangerBase
 
 logger = logging.getLogger(__name__)
 
 
-class Exchanger:
-
-    ROLE_WS = 1
-    ROLE_APP = 2
-
-    def __init__(self, name, role, client_id=0):
-        if role not in [self.ROLE_WS, self.ROLE_APP]:
-            raise ValueError('bad role {}'.format(role))
-        self.role = role
-        self.client_id = client_id
-        self.name = name
+class Exchanger(ExchangerBase):
 
     async def __aenter__(self):
         logger.debug('connecting with role {}'.format(self.role))
-        params = config.get_amqp_conn_params()
+        params = config.get_amqp_conn_params(self.url)
         params['login'] = params.pop('username')
         params['virtualhost'] = params.pop('virtual_host')
         self._transport, self._protocol = await aioamqp.connect(**params)
         # TODO: handle reconnect awaiting from self._conn
         self._chan = await self._protocol.channel()
-        app_exchange_name = 'pushpull.app'
-        app_routing_key = ''
-        ws_exchange_name = 'pushpull.ws'
-        ws_routing_key = 'pushpull.ws.{}'.format(self.name)
+        app_exchange_name = self.get_app_exchange_name()
+        app_routing_key = self.get_app_routing_key()
+        ws_exchange_name = self.get_ws_exchange_name()
+        ws_routing_key = self.get_ws_routing_key()
         await self._chan.exchange(app_exchange_name, 'fanout', durable=True)
         await self._chan.exchange(ws_exchange_name, 'direct', durable=True)
         if self.role == self.ROLE_WS:

pushpull/amqp/gateway/driver_asynqp.py

@@ -3,32 +3,23 @@
 
 import asynqp
 
+from .driver_base import ExchangerBase
 from ... import config
 
 logger = logging.getLogger(__name__)
 
 
-class Exchanger:
-
-    ROLE_WS = 1
-    ROLE_APP = 2
-
-    def __init__(self, name, role, client_id=0):
-        if role not in [self.ROLE_WS, self.ROLE_APP]:
-            raise ValueError('bad role {}'.format(role))
-        self.role = role
-        self.client_id = client_id
-        self.name = name
+class Exchanger(ExchangerBase):
 
     async def __aenter__(self):
         logger.debug('connecting with role {}'.format(self.role))
-        params = config.get_amqp_conn_params()
+        params = config.get_amqp_conn_params(self.url)
         self._conn = await asynqp.connect(**params)
         self._chan = await self._conn.open_channel()
-        app_exchange_name = 'pushpull.app'
-        app_routing_key = ''
-        ws_exchange_name = 'pushpull.ws'
-        ws_routing_key = 'pushpull.ws.{}'.format(self.name)
+        app_exchange_name = self.get_app_exchange_name()
+        app_routing_key = self.get_app_routing_key()
+        ws_exchange_name = self.get_ws_exchange_name()
+        ws_routing_key = self.get_ws_routing_key()
         app_exchange = await self._chan.declare_exchange(app_exchange_name, 'fanout')
         ws_exchange = await self._chan.declare_exchange(ws_exchange_name, 'direct')
         if self.role == self.ROLE_WS:

pushpull/amqp/gateway/driver_base.py

@@ -0,0 +1,24 @@
+class ExchangerBase:
+
+    ROLE_WS = 1
+    ROLE_APP = 2
+
+    def __init__(self, name, role, client_id=0, url=None):
+        if role not in [self.ROLE_WS, self.ROLE_APP]:
+            raise ValueError('bad role {}'.format(role))
+        self.role = role
+        self.client_id = client_id
+        self.name = name
+        self.url = url
+
+    def get_app_exchange_name(self):
+        return 'pushpull.app'
+
+    def get_app_routing_key(self):
+        return ''
+
+    def get_ws_exchange_name(self):
+        return 'pushpull.ws'
+
+    def get_ws_routing_key(self):
+        return 'pushpull.ws.{}'.format(self.name)

pushpull/cli/client.py

@@ -24,10 +24,11 @@ def challenge_websocket(url, name):
 
 
 @click.command()
+@click.argument('url')
 @click.argument('name')
-def challenge_amqp(name):
+def challenge_amqp(url, name):
     logging.basicConfig(level=logging.DEBUG)
-    click.echo(asyncio.get_event_loop().run_until_complete(amqp_client.challenge(name, sys.stdin, sys.stdout)))
+    click.echo(asyncio.get_event_loop().run_until_complete(amqp_client.challenge(url, name, sys.stdin, sys.stdout)))
 client.add_command(challenge_amqp)
 
 

pushpull/config.py

@@ -33,9 +33,10 @@ def get_url_path():
     return path
 
 
-def get_amqp_conn_params():
-    url_string = BROKER_URL or 'amqp://guest:guest@localhost:5672/'
-    url = urllib.parse.urlparse(url_string)
+def get_amqp_conn_params(url=None):
+    if url is None:
+        url = BROKER_URL or 'amqp://guest:guest@localhost:5672/'
+    url = urllib.parse.urlparse(url)
     return {
         'host': url.hostname or 'localhost',
         'port': url.port,

pushpull/websocket/server.py

@@ -13,7 +13,9 @@ def serve(argv):
     cors.add(
         route,
         {
-            origin: aiohttp_cors.ResourceOptions(allow_credentials=True)
+            origin: aiohttp_cors.ResourceOptions(
+                allow_credentials=True,
+            )
             for origin in config.get_cors_allowed_origins()
         }
     )

setup.py

@@ -21,5 +21,11 @@
     url='https://github.com/elastic-coders/pushpull/',
     license='MIT',
     include_package_data=True,
-    install_requires=reqs
+    install_requires=reqs,
+    entry_points={
+        'console_scripts': [
+            'pushpull-server = pushpull.cli.server:serve',
+            'pushpull-client = pushpull.cli.client:client',
+        ]
+    }
 )