chiark / gitweb /
no dupe warning on cancel
[hippotat.git] / client
1 #!/usr/bin/python3
2
3 from hippotat import *
4
5 import twisted.web
6 import twisted.web.client
7
8 import io
9
10 client_cs = None
11
12 def set_client(ci,cs,pw):
13   global client_cs
14   global password
15   assert(client_cs is None)
16   client_cs = cs
17   c.client = ci
18   c.max_outstanding = cfg.getint(cs, 'max_requests_outstanding')
19   c.target_outstanding = cfg.getint(cs, 'target_requests_outstanding')
20   password = pw
21
22 def process_cfg():
23   global url
24   global max_requests_outstanding
25
26   process_cfg_common_always()
27   process_cfg_server()
28
29   try:
30     c.url = cfg.get('server','url')
31   except NoOptionError:
32     process_cfg_saddrs()
33     c.url = c.saddrs[0].url()
34
35   process_cfg_clients(set_client)
36
37   c.routes = cfg.get('virtual','routes')
38   c.max_queue_time = cfg.getint(client_cs, 'max_queue_time')
39   c.max_batch_up   = cfg.getint(client_cs, 'max_batch_up')
40   c.http_retry     = cfg.getint(client_cs, 'http_retry')
41   c.http_timeout = (cfg.getint(client_cs, 'http_timeout') +
42                     cfg.getint(client_cs, 'http_timeout_grace'))
43
44   process_cfg_ipif(client_cs,
45                    (('local', 'client'),
46                     ('peer',  'server'),
47                     ('rnets', 'routes')))
48
49 outstanding = { }
50
51 def log_outstanding():
52   log_debug(DBG.CTRL_DUMP, 'OS %s' % outstanding)
53
54 def start_client():
55   global queue
56   global agent
57   queue = PacketQueue('up', c.max_queue_time)
58   agent = twisted.web.client.Agent(reactor, connectTimeout = c.http_timeout)
59
60 def outbound(packet, saddr, daddr):
61   #print('OUT ', saddr, daddr, repr(packet))
62   queue.append(packet)
63   check_outbound()
64
65 class GeneralResponseConsumer(twisted.internet.protocol.Protocol):
66   def __init__(self, req, desc):
67     self._req = req
68     self._desc = desc
69
70   def _log(self, dflag, msg, **kwargs):
71     log_debug(dflag, '%s: %s' % (self._desc, msg), idof=self._req, **kwargs)
72
73   def connectionMade(self):
74     self._log(DBG.HTTP_CTRL, 'connectionMade')
75
76 class ResponseConsumer(GeneralResponseConsumer):
77   def __init__(self, req):
78     super().__init__(req, 'RC')
79     ssddesc = '[%s] %s' % (id(req), self._desc)
80     self._ssd = SlipStreamDecoder(ssddesc, queue_inbound)
81     self._log(DBG.HTTP_CTRL, '__init__')
82
83   def dataReceived(self, data):
84     self._log(DBG.HTTP, 'dataReceived', d=data)
85     try:
86       self._ssd.inputdata(data)
87     except Exception as e:
88       self._handleexception()
89
90   def connectionLost(self, reason):
91     self._log(DBG.HTTP_CTRL, 'connectionLost ' + str(reason))
92     if not reason.check(twisted.web.client.ResponseDone):
93       self.latefailure()
94       return
95     try:
96       self._log(DBG.HTTP, 'ResponseDone')
97       self._ssd.flush()
98       req_fin(self._req)
99     except Exception as e:
100       self._handleexception()
101
102   def _handleexception(self):
103     self._latefailure(traceback.format_exc())
104
105   def _latefailure(self, reason):
106     self._log(DBG.HTTP_CTRL, '_latefailure ' + str(reason))
107     req_err(self._req, reason)
108
109 class ErrorResponseConsumer(twisted.internet.protocol.Protocol):
110   def __init__(self, req, resp):
111     super().__init__(req, 'ERROR-RC')
112     self._resp = resp
113     self._m = b''
114     try:
115       self._phrase = resp.phrase.decode('utf-8')
116     except Exception:
117       self._phrase = repr(resp.phrase)
118     self._log(DBG.HTTP_CTRL, '__init__ %d %s' % (resp.code, self._phrase))
119
120   def dataReceived(self, data):
121     self._log(DBG.HTTP_CTRL, 'dataReceived ' + repr(data))
122     self._m += data
123
124   def connectionLost(self, reason):
125     try:
126       mbody = self._m.decode('utf-8')
127     except Exception:
128       mbody = repr(self._m)
129     if not reason.check(twisted.web.client.ResponseDone):
130       mbody += ' || ' + str(reason)
131     req_err(self._req,
132             "FAILED %d %s | %s"
133             % (self._resp.code, self._phrase, mbody))
134
135 def req_ok(req, resp):
136   log_debug(DBG.HTTP_CTRL,
137             'req_ok %d %s %s' % (resp.code, repr(resp.phrase), str(resp)),
138             idof=req)
139   if resp.code == 200:
140     rc = ResponseConsumer(req)
141   else:
142     rc = ErrorResponseConsumer(req, resp)
143
144   resp.deliverBody(rc)
145   # now rc is responsible for calling req_fin
146
147 def req_err(req, err):
148   # called when the Deferred fails, or (if it completes),
149   # later, by ResponsConsumer or ErrorResponsConsumer
150   try:
151     log_debug(DBG.HTTP_CTRL, 'req_err ' + str(err), idof=req)
152     if isinstance(err, twisted.python.failure.Failure):
153       err = err.getTraceback()
154     print('[%#x] %s' % (id(req), err), file=sys.stderr)
155     if not isinstance(outstanding[req], int):
156       raise RuntimeError('[%#x] previously %s' % (id(req), outstanding[req]))
157     outstanding[req] = err
158     log_outstanding()
159     reactor.callLater(c.http_retry, partial(req_fin, req))
160   except Exception as e:
161     crash(traceback.format_exc() + '\n----- handling -----\n' + err)
162
163 def req_fin(req):
164   del outstanding[req]
165   log_debug(DBG.HTTP_CTRL, 'req_fin OS=%d' % len(outstanding), idof=req)
166   check_outbound()
167
168 class Errb:
169   def __init__(self, req):
170     self._req = req
171   def call(self, err):
172     req_err(self._req, err)
173
174 def check_outbound():
175   global outstanding
176
177   while True:
178     if                          len(outstanding) >= c.max_outstanding   : break
179     if not queue.nonempty() and len(outstanding) >= c.target_outstanding: break
180
181     d = b''
182     def moredata(s): nonlocal d; d += s
183     queue.process((lambda: len(d)),
184                   moredata,
185                   c.max_batch_up)
186     
187     d = mime_translate(d)
188
189     crlf = b'\r\n'
190     lf   =   b'\n'
191     mime = (b'--b'                                        + crlf +
192             b'Content-Type: text/plain; charset="utf-8"'  + crlf +
193             b'Content-Disposition: form-data; name="m"'   + crlf + crlf +
194             str(c.client)             .encode('ascii')    + crlf +
195             password                                      + crlf +
196             str(c.target_outstanding) .encode('ascii')    + crlf +
197             str(c.http_timeout)       .encode('ascii')    + crlf +
198           ((
199             b'--b'                                        + crlf +
200             b'Content-Type: application/octet-stream'     + crlf +
201             b'Content-Disposition: form-data; name="d"'   + crlf + crlf +
202             d                                             + crlf
203            ) if len(d) else b'')                               +
204             b'--b--'                                      + crlf)
205
206     #df = open('data.dump.dbg', mode='wb')
207     #df.write(mime)
208     #df.close()
209     # POST -use -c 'multipart/form-data; boundary="b"' http://localhost:8099/ <data.dump.dbg
210
211     log_debug(DBG.HTTP_FULL, 'requesting: ' + str(mime))
212
213     hh = { 'User-Agent': ['hippotat'],
214            'Content-Type': ['multipart/form-data; boundary="b"'],
215            'Content-Length': [str(len(mime))] }
216
217     bytesreader = io.BytesIO(mime)
218     producer = twisted.web.client.FileBodyProducer(bytesreader)
219
220     req = agent.request(b'POST',
221                         c.url,
222                         twisted.web.client.Headers(hh),
223                         producer)
224
225     outstanding[req] = len(d)
226     log_debug(DBG.HTTP_CTRL, 'request OS=%d' % len(outstanding), idof=req, d=d)
227     req.addTimeout(c.http_timeout, reactor)
228     req.addCallback(partial(req_ok, req))
229     req.addErrback(partial(req_err, req))
230
231   log_outstanding()
232
233 common_startup()
234 process_cfg()
235 start_client()
236 start_ipif(c.ipif_command, outbound)
237 check_outbound()
238 common_run()