Merge pull request #60 from matburt/send_support

Add send support back in

Author: matburt

receptor/config.py

@@ -250,10 +250,10 @@ def __init__(self, args=None):
         # send options
         self.add_config_option(
             section='send',
-            key='socket_path',
-            default_value='/var/run/receptor_controller.sock',
-            value_type='path',
-            hint='Path to control socket for controller commands.',
+            key='peer',
+            default_value='',
+            value_type='str',
+            hint='The peer to relay the directive through'
         )
         self.add_config_option(
             section='send',
@@ -276,7 +276,7 @@ def __init__(self, args=None):
             long_option='send_payload',
             default_value='',
             value_type='str',
-            hint='Payload of the directive to send. Use - for stdin.',
+            hint='Payload of the directive to send. Use - for stdin or give the path to a file to transmit the file contents.',
         )
         # Component options. These are also only used in a config section
         # like auth, so they also set `subparse=False`.

receptor/entrypoints.py

@@ -1,10 +1,13 @@
 import logging
 import time
 import asyncio
+import sys
+import os
 
 from prometheus_client import start_http_server
 
 from .controller import Controller
+from .messages import Message
 
 logger = logging.getLogger(__name__)
 
@@ -84,4 +87,26 @@ async def send_pings():
 
 
 def run_as_send(config):
-    pass
+    async def send_entrypoint():
+        read_task = controller.loop.create_task(read_responses())
+        await controller.add_peer(config.send_peer)
+        start_wait = time.time()
+        while not controller.receptor.router.node_is_known(config.ping_recipient) and (time.time() - start_wait < 5):
+            await asyncio.sleep(0.1)
+        msg = Message(config.send_recipient, config.send_directive)
+        if config.send_payload == "-":
+            msg.buffer(sys.stdin)
+        elif os.path.exists(config.send_payload):
+            msg.file(config.send_payload)
+        else:
+            msg.data(config.send_payload)
+        await controller.send(msg)
+        await read_task
+
+    async def read_responses():
+        while True:
+            print("{}".format(await controller.recv()))
+
+    logger.info(f'Sending directive {config.send_directive} to {config.send_recipient} via {config.send_peer}')
+    controller = Controller(config)
+    controller.run(send_entrypoint)

receptor/messages/__init__.py

@@ -0,0 +1 @@
+from .envelope import Message  # noqa

receptor/messages/envelope.py

@@ -18,7 +18,7 @@
 
 class Message:
 
-    __slots__ = ("fd")
+    __slots__ = ("fd", "recipient", "directive")
 
     def __init__(self, recipient, directive):
         self.fd = io.BytesIO()