|
720
|
1 import logging
|
|
|
2 import time
|
|
|
3
|
|
|
4 from rx.subjects import BehaviorSubject
|
|
|
5 from twisted.internet import reactor
|
|
|
6 import treq
|
|
|
7
|
|
|
8 log = logging.getLogger('httpputoutputs')
|
|
|
9
|
|
|
10 class HttpPutOutput(object):
|
|
|
11 def __init__(self, url,
|
|
|
12 refreshSecs,#: BehaviorSubject,
|
|
|
13 mockOutput=False):
|
|
|
14 self.url = url
|
|
|
15 self.mockOutput = mockOutput
|
|
|
16 self.payload = None
|
|
|
17 self.foafAgent = None
|
|
|
18 self.nextCall = None
|
|
|
19 self.lastErr = None
|
|
|
20 self.numRequests = 0
|
|
|
21 self.refreshSecs = refreshSecs
|
|
|
22
|
|
|
23 def report(self):
|
|
|
24 return {
|
|
|
25 'url': self.url,
|
|
|
26 'urlAbbrev': self.url
|
|
|
27 .replace('http%3A%2F%2Fprojects.bigasterisk.com%2Froom%2F', ':')
|
|
|
28 .replace('http://projects.bigasterisk.com/room/', ':')
|
|
|
29 .replace('.vpn-home.bigasterisk.com', '.vpn-home'),
|
|
|
30 'payload': self.payload,
|
|
|
31 'numRequests': self.numRequests,
|
|
|
32 'lastChangeTime': round(self.lastChangeTime, 2),
|
|
|
33 'lastErr': str(self.lastErr) if self.lastErr is not None else None,
|
|
|
34 }
|
|
|
35
|
|
|
36 def setPayload(self, payload, foafAgent):
|
|
|
37 if self.numRequests > 0 and (self.payload == payload and
|
|
|
38 self.foafAgent == foafAgent):
|
|
|
39 return
|
|
|
40 self.payload = payload
|
|
|
41 self.foafAgent = foafAgent
|
|
|
42 self.lastChangeTime = time.time()
|
|
|
43 self.makeRequest()
|
|
|
44
|
|
|
45 def makeRequest(self):
|
|
|
46 if self.payload is None:
|
|
|
47 log.debug("PUT None to %s - waiting", self.url)
|
|
|
48 return
|
|
|
49 h = {}
|
|
|
50 if self.foafAgent:
|
|
|
51 h['x-foaf-agent'] = self.foafAgent
|
|
|
52 if self.nextCall and self.nextCall.active():
|
|
|
53 self.nextCall.cancel()
|
|
|
54 self.nextCall = None
|
|
|
55 self.lastErr = None
|
|
|
56 log.debug("PUT %s payload=%s agent=%s",
|
|
|
57 self.url, self.payload, self.foafAgent)
|
|
|
58 if not self.mockOutput:
|
|
|
59 self.currentRequest = treq.put(self.url, data=self.payload,
|
|
|
60 headers=h, timeout=3)
|
|
|
61 self.currentRequest.addCallback(self.onResponse).addErrback(
|
|
|
62 self.onError)
|
|
|
63 else:
|
|
|
64 reactor.callLater(.2, self.onResponse, None)
|
|
|
65
|
|
|
66 self.numRequests += 1
|
|
|
67
|
|
|
68 def currentRefreshSecs(self):
|
|
|
69 out = None
|
|
|
70 if 1:
|
|
|
71 # workaround
|
|
|
72 def secsFromLiteral(v):
|
|
|
73 if v[-1] != 's':
|
|
|
74 raise NotImplementedError(v)
|
|
|
75 return float(v[:-1])
|
|
|
76
|
|
|
77 out = secsFromLiteral(self.refreshSecs.value)
|
|
|
78 else:
|
|
|
79 # goal: caller should map secsFromLiteral on the
|
|
|
80 # observable, so we see a float
|
|
|
81 def recv(v):
|
|
|
82 log.info('recv %r', v)
|
|
|
83 import ipdb;ipdb.set_trace()
|
|
|
84 self.refreshSecs.subscribe(recv)
|
|
|
85 if out is None:
|
|
|
86 raise ValueError('refreshSecs had no value')
|
|
|
87 log.debug(' got refresh %r', out)
|
|
|
88 return out
|
|
|
89
|
|
|
90 def onResponse(self, resp):
|
|
|
91 log.debug(" PUT %s ok", self.url)
|
|
|
92 self.lastErr = None
|
|
|
93 self.currentRequest = None
|
|
|
94 self.nextCall = reactor.callLater(self.currentRefreshSecs(),
|
|
|
95 self.makeRequest)
|
|
|
96
|
|
|
97 def onError(self, err):
|
|
|
98 self.lastErr = err
|
|
|
99 log.debug(' PUT %s failed: %s', self.url, err)
|
|
|
100 self.currentRequest = None
|
|
|
101 self.nextCall = reactor.callLater(self.currentRefreshSecs(),
|
|
|
102 self.makeRequest)
|
|
|
103
|
|
|
104 class HttpPutOutputs(object):
|
|
|
105 """these grow forever"""
|
|
|
106 def __init__(self, mockOutput=False):
|
|
|
107 self.mockOutput = mockOutput
|
|
|
108 self.state = {} # url: HttpPutOutput
|
|
|
109
|
|
|
110 def put(self, url, payload, foafAgent, refreshSecs):
|
|
|
111 if url not in self.state:
|
|
|
112 self.state[url] = HttpPutOutput(url, mockOutput=self.mockOutput,
|
|
|
113 refreshSecs=refreshSecs)
|
|
|
114 self.state[url].setPayload(payload, foafAgent)
|
