client

   1 #!/usr/bin/python3
   2
   3 from hippotat import *
   4
   5 client_cs = None
   6
   7 def set_client(ci,cs,pw):
   8   global client_cs
   9   global password
  10   assert(client_cs is None)
  11   client_cs = cs
  12   c.client = ci
  13   c.max_outstanding = cfg.getint(cs, 'max_requests_outstanding')
  14   c.target_outstanding = cfg.getint(cs, 'target_requests_outstanding')
  15   password = pw
  16
  17 def process_cfg():
  18   global url
  19   global max_requests_outstanding
  20
  21   process_cfg_common_always()
  22   process_cfg_server()
  23
  24   try:
  25     c.url = cfg.get('server','url')
  26   except NoOptionError:
  27     process_cfg_saddrs()
  28     sa = c.saddrs[0].url()
  29
  30   process_cfg_clients(set_client)
  31
  32   c.routes = cfg.get('virtual','routes')
  33   c.max_queue_time = cfg.getint(client_cs, 'max_queue_time')
  34   c.max_batch_up   = cfg.getint(client_cs, 'max_batch_up')
  35
  36   process_cfg_ipif(client_cs,
  37                    (('local', 'client'),
  38                     ('peer',  'server'),
  39                     ('rnets', 'routes')))
  40
  41 outstanding = 0
  42
  43 def start_client():
  44   global queue
  45   global agent
  46   queue = PacketQueue(c.max_queue_time)
  47   agent = twisted.web.client.Agent(reactor, connectTimeout = c.http_timeout)
  48
  49 def outbound(packet, saddr, daddr):
  50   #print('OUT ', saddr, daddr, repr(packet))
  51   queue.append(packet)
  52   check_outbound()
  53
  54 class ResponseConsumer(twisted.internet.protocol.Protocol):
  55   def __init__(self):
  56     self._ssd = SlipStreamDecoder(queue_inbound)
  57   def dataReceived(self, data):
  58     self._ssd.inputdata(mime_translate(data))
  59   def connectionMade(self): pass
  60   def connectionLost(self, reason):
  61     if isinstance(reason, twisted.internet.error.ConnectionDone):
  62       self._ssd.flush()
  63     else:
  64       print(reason, file=sys.stderr)
  65
  66 def req_ok(resp):
  67   resp.deliverBody(ResponseConsumer())
  68
  69 def req_err(err):
  70   print(err, >>sys.stderr)
  71
  72 def req_fin(*args):
  73   outstanding--
  74
  75 def check_outbound():
  76   while True:
  77     if                         outstanding >= c.max_outstanding   : break
  78     if not queue.nonempty() && outstanding >= c.target_outstanding: break
  79
  80     d = b''
  81     queue.process((lambda: len(d)),
  82                   (lambda s: d += s),
  83                   c.max_batch_up)
  84     assert(len(d))
  85
  86     crlf = b'\r\n'
  87     mime = (b'--b'                                      + crlf +
  88             b'Content-Disposition: form-data; name="m"' + crlf +
  89             password                                    + crlf +
  90             c.client                                    + crlf +
  91             c.target_outstanding                        + crlf +
  92             b'--b'                                      + crlf +
  93             b'Content-Disposition: form-data; name="d"' + crlf +
  94             mime_translate(d)                           + crlf +
  95             b'--b--'                                    + crlf)
  96
  97     hh = { 'User-Agent': ['hippotat'],
  98            'Content-Type': ['multipart/form-data; boundary="b"'] }
  99     req = agent.request(b'POST',
 100                         c.url,
 101                         twisted.web.client.Headers(hh))
 102     req.addTimeout(c.http_timeout,
 103     req.addCallbacks(req_ok, req_err)
 104     req.addBoth(req_fin)
 105     outstanding++
 106
 107 common_startup()
 108 process_cfg()
 109 start_client()
 110 start_ipif(c.ipif_command, outbound)
 111 common_run()