X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ian/git?p=hippotat.git;a=blobdiff_plain;f=client;h=c9682bd944e135b4a57f0bdb8c375f5bdf2188fa;hp=b9f8ce516298eae80a57a8ec17b0f55145c5c435;hb=68afd97b014f134e9681ab90b374a6b44e6eb2b3;hpb=c55f394eb74d593bd5535ba3dc639473db69784d diff --git a/client b/client index b9f8ce5..c9682bd 100755 --- a/client +++ b/client @@ -2,22 +2,138 @@ from hippotat import * -defcfg = ''' -[DEFAULT] -max_requests_outstanding = 4 +import twisted.web +import twisted.web.client -[virtual] -mtu = 1500 -# [host] } maybe computed from `network' (see server defaults) +client_cs = None -[server] -# url } maybe computed from `addrs' and `port' (see server defaults) +def set_client(ci,cs,pw): + global client_cs + global password + assert(client_cs is None) + client_cs = cs + c.client = ci + c.max_outstanding = cfg.getint(cs, 'max_requests_outstanding') + c.target_outstanding = cfg.getint(cs, 'target_requests_outstanding') + password = pw -# [] -# password = -''' +def process_cfg(): + global url + global max_requests_outstanding -def startup(): - common_startup(defcfg) + process_cfg_common_always() + process_cfg_server() -startup() + try: + c.url = cfg.get('server','url') + except NoOptionError: + process_cfg_saddrs() + c.url = c.saddrs[0].url() + + process_cfg_clients(set_client) + + c.routes = cfg.get('virtual','routes') + c.max_queue_time = cfg.getint(client_cs, 'max_queue_time') + c.max_batch_up = cfg.getint(client_cs, 'max_batch_up') + c.http_timeout = cfg.getint(client_cs, 'http_timeout') + c.http_retry = cfg.getint(client_cs, 'http_retry') + + process_cfg_ipif(client_cs, + (('local', 'client'), + ('peer', 'server'), + ('rnets', 'routes'))) + +outstanding = 0 + +def start_client(): + global queue + global agent + queue = PacketQueue('up', c.max_queue_time) + agent = twisted.web.client.Agent(reactor, connectTimeout = c.http_timeout) + +def outbound(packet, saddr, daddr): + #print('OUT ', saddr, daddr, repr(packet)) + queue.append(packet) + check_outbound() + +class ResponseConsumer(twisted.internet.protocol.Protocol): + def __init__(self): + self._ssd = SlipStreamDecoder(queue_inbound) + def dataReceived(self, data): + try: self._ssd.inputdata(mime_translate(data)) + except Exception as e: asyncfailure(e) + def connectionMade(self): pass + def connectionLost(self, reason): + if isinstance(reason, twisted.internet.error.ConnectionDone): + try: self._ssd.flush() + except Exception as e: asyncfailure(e) + else: + asyncfailure(reason) + +def req_ok(resp): + resp.deliverBody(ResponseConsumer()) + req_fin() + +def req_err(err): + print(err, file=sys.stderr) + reactor.callLater(c.http_retry, req_fin) + +def req_fin(*args): + global outstanding + outstanding -= 1 + check_outbound() + +def asyncfailure(reason): + global outstanding + outstanding += 1 + req_err(reason) + +def check_outbound(): + global outstanding + + while True: + if outstanding >= c.max_outstanding : break + if not queue.nonempty() and outstanding >= c.target_outstanding: break + + d = b'' + def moredata(s): nonlocal d; d += s + queue.process((lambda: len(d)), + moredata, + c.max_batch_up) + + crlf = b'\r\n' + lf = b'\n' + mime = (b'--b' + crlf + + b'Content-Disposition: form-data; name="m"' + crlf + crlf + + password + crlf + + str(c.client) .encode('ascii') + crlf + + str(c.target_outstanding) .encode('ascii') + crlf + + (( + b'--b' + crlf + + b'Content-Disposition: form-data; name="d"' + crlf + crlf + + mime_translate(d) + crlf + ) if len(d) else b'') + + b'--b--' + crlf) + + df = open('data.dump.dbg', mode='wb') + df.write(mime) + df.close() + # POST -use -c 'multipart/form-data; boundary="b"' http://localhost:8099/