diff options
Diffstat (limited to 'Biz/Que/Client.py')
| -rwxr-xr-x | Biz/Que/Client.py | 93 |
1 files changed, 57 insertions, 36 deletions
diff --git a/Biz/Que/Client.py b/Biz/Que/Client.py index 1063eb8..58877bf 100755 --- a/Biz/Que/Client.py +++ b/Biz/Que/Client.py @@ -11,11 +11,15 @@ import logging import os import subprocess import sys +import textwrap import time import urllib.parse import urllib.request as request -MAX_TIMEOUT = 99999999 # basically never timeout +MAX_TIMEOUT = 9999999 +RETRIES = 10 +DELAY = 3 +BACKOFF = 1 def auth(args): @@ -33,8 +37,8 @@ def auth(args): def autodecode(bytestring): - """Attempt to decode bytes `bs` into common codecs, preferably utf-8. If - no decoding is available, just return the raw bytes. + """Attempt to decode bytes into common codecs, preferably utf-8. If no + decoding is available, just return the raw bytes. For all available codecs, see: <https://docs.python.org/3/library/codecs.html#standard-encodings> @@ -50,7 +54,7 @@ def autodecode(bytestring): return bytestring -def retry(exception, tries=4, delay=3, backoff=2): +def retry(exception, tries=RETRIES, delay=DELAY, backoff=BACKOFF): "Decorator for retrying an action." def decorator(func): @@ -73,20 +77,23 @@ def retry(exception, tries=4, delay=3, backoff=2): return decorator +@retry(urllib.error.URLError) +@retry(http.client.IncompleteRead) +@retry(http.client.RemoteDisconnected) def send(args): "Send a message to the que." logging.debug("send") key = auth(args) data = args.infile req = request.Request(f"{args.host}/{args.target}") - req.add_header("User-AgenT", "Que/Client") + req.add_header("User-Agent", "Que/Client") + req.add_header("Content-Type", "text/plain;charset=utf-8") if key: req.add_header("Authorization", key) if args.serve: logging.debug("serve") while not time.sleep(1): request.urlopen(req, data=data, timeout=MAX_TIMEOUT) - else: request.urlopen(req, data=data, timeout=MAX_TIMEOUT) @@ -96,75 +103,89 @@ def then(args, msg): if args.then: logging.debug("then") subprocess.run( - args.then.format(msg=msg, que=args.target), check=False, shell=True, + args.then.format(msg=msg, que=args.target), + check=False, + shell=True, ) -@retry(http.client.IncompleteRead, tries=10, delay=5, backoff=1) -@retry(http.client.RemoteDisconnected, tries=10, delay=2, backoff=2) +@retry(urllib.error.URLError) +@retry(http.client.IncompleteRead) +@retry(http.client.RemoteDisconnected) def recv(args): "Receive a message from the que." logging.debug("recv on: %s", args.target) - params = urllib.parse.urlencode({"poll": args.poll}) - req = request.Request(f"{args.host}/{args.target}?{params}") + if args.poll: + req = request.Request(f"{args.host}/{args.target}/stream") + else: + req = request.Request(f"{args.host}/{args.target}") req.add_header("User-Agent", "Que/Client") key = auth(args) if key: req.add_header("Authorization", key) with request.urlopen(req) as _req: if args.poll: - logging.debug("poll") + logging.debug("polling") while not time.sleep(1): - logging.debug("reading") - msg = autodecode(_req.readline()) - logging.debug("read") - print(msg, end="") - then(args, msg) + reply =_req.readline() + if reply: + msg = autodecode(reply) + logging.debug("read") + print(msg, end="") + then(args, msg) + else: + continue else: - msg = autodecode(_req.read()) + msg = autodecode(_req.readline()) print(msg) then(args, msg) def get_args(): "Command line parser" - cli = argparse.ArgumentParser(description=__doc__) + cli = argparse.ArgumentParser( + description=__doc__, + epilog=textwrap.dedent( + f"""Requests will retry up to {RETRIES} times, with {DELAY} seconds + between attempts.""" + ), + ) cli.add_argument("--debug", action="store_true", help="log to stderr") cli.add_argument( "--host", default="http://que.run", help="where que-server is running" ) cli.add_argument( - "--poll", default=False, action="store_true", help="stream data from the que" + "--poll", + default=False, + action="store_true", + help=textwrap.dedent( + """keep the connection open to stream data from the que. without + this flag, the program will exit after receiving a message""" + ), ) cli.add_argument( "--then", - help=" ".join( - [ - "when polling, run this shell command after each response,", - "presumably for side effects," - r"replacing '{que}' with the target and '{msg}' with the body of the response", - ] + help=textwrap.dedent( + """when polling, run this shell command after each response, + presumably for side effects, replacing '{que}' with the target and + '{msg}' with the body of the response""" ), ) cli.add_argument( "--serve", default=False, action="store_true", - help=" ".join( - [ - "when posting to the que, do so continuously in a loop.", - "this can be used for serving a webpage or other file continuously", - ] + help=textwrap.dedent( + """when posting to the que, do so continuously in a loop. this can + be used for serving a webpage or other file continuously""" ), ) - cli.add_argument( - "target", help="namespace and path of the que, like 'ns/path/subpath'" - ) + cli.add_argument("target", help="namespace and path of the que, like 'ns/path'") cli.add_argument( "infile", nargs="?", type=argparse.FileType("rb"), - help="data to put on the que. Use '-' for stdin, otherwise should be a readable file", + help="data to put on the que. use '-' for stdin, otherwise should be a readable file", ) return cli.parse_args() @@ -173,7 +194,7 @@ if __name__ == "__main__": ARGV = get_args() if ARGV.debug: logging.basicConfig( - format="%(asctime)s %(message)s", + format="%(asctime)s: %(levelname)s: %(message)s", level=logging.DEBUG, datefmt="%Y.%m.%d..%H.%M.%S", ) |