Mercurial > code > home > repos > homeauto
annotate lib/patchablegraph/patchsource.py @ 514:495f573af4f4
make patchablegraph release
Ignore-this: f55c9a56b052797ff23a80630714b51a
| author | drewp@bigasterisk.com |
|---|---|
| date | Mon, 22 Apr 2019 23:29:19 -0700 |
| parents | lib/patchsource.py@b5abd4fc65a4 |
| children | 2d13254fdd44 |
| rev | line source |
|---|---|
| 514 | 1 import logging |
| 302 | 2 import traceback |
| 3 from rdflib import ConjunctiveGraph | |
| 4 from rdflib.parser import StringInputSource | |
| 514 | 5 from twisted.internet import reactor, defer |
| 302 | 6 |
| 7 from patchablegraph import patchFromJson | |
|
351
7716b1810d6c
reasoning & collector move into docker images
drewp@bigasterisk.com
parents:
312
diff
changeset
|
8 from rdfdb.patch import Patch |
| 514 | 9 from twisted_sse_demo.eventsource import EventSource |
| 302 | 10 |
|
312
170dc9b1e789
fix input graph web display by dirtying combinedGraph better.
drewp@bigasterisk.com
parents:
306
diff
changeset
|
11 log = logging.getLogger('fetch') |
| 302 | 12 |
| 13 class PatchSource(object): | |
| 14 """wrap EventSource so it emits Patch objects and has an explicit stop method.""" | |
|
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
15 def __init__(self, url, agent): |
|
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
16 self.url = str(url) |
| 302 | 17 |
| 18 # add callbacks to these to learn if we failed to connect | |
| 19 # (approximately) or if the ccnnection was unexpectedly lost | |
| 20 self.connectionFailed = defer.Deferred() | |
| 21 self.connectionLost = defer.Deferred() | |
| 22 | |
| 23 self._listeners = set() | |
| 24 log.info('start read from %s', url) | |
|
312
170dc9b1e789
fix input graph web display by dirtying combinedGraph better.
drewp@bigasterisk.com
parents:
306
diff
changeset
|
25 # note: fullGraphReceived isn't guaranteed- the stream could |
|
170dc9b1e789
fix input graph web display by dirtying combinedGraph better.
drewp@bigasterisk.com
parents:
306
diff
changeset
|
26 # start with patches |
| 302 | 27 self._fullGraphReceived = False |
|
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
28 self._eventSource = EventSource(url.toPython().encode('utf8'), |
|
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
29 userAgent=agent) |
| 302 | 30 |
|
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
31 self._eventSource.addEventListener(b'fullGraph', self._onFullGraph) |
|
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
32 self._eventSource.addEventListener(b'patch', self._onPatch) |
| 302 | 33 self._eventSource.onerror(self._onError) |
|
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
34 self._eventSource.onConnectionLost = self._onDisconnect |
| 306 | 35 |
|
439
124c921ad52d
stats->state to make room for greplin stats
drewp@bigasterisk.com
parents:
429
diff
changeset
|
36 def state(self): |
| 306 | 37 return { |
| 38 'url': self.url, | |
| 39 'fullGraphReceived': self._fullGraphReceived, | |
| 40 } | |
| 302 | 41 |
| 42 def addPatchListener(self, func): | |
| 43 """ | |
| 44 func(patch, fullGraph=[true if the patch is the initial fullgraph]) | |
| 45 """ | |
| 46 self._listeners.add(func) | |
| 47 | |
| 48 def stop(self): | |
| 49 log.info('stop read from %s', self.url) | |
| 50 try: | |
| 51 self._eventSource.protocol.stopProducing() # needed? | |
| 52 except AttributeError: | |
| 53 pass | |
| 54 self._eventSource = None | |
| 55 | |
|
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
56 def _onDisconnect(self, reason): |
|
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
57 log.debug('PatchSource._onDisconnect from %s (%s)', self.url, reason) |
| 302 | 58 # skip this if we're doing a stop? |
| 59 self.connectionLost.callback(None) | |
| 60 | |
| 61 def _onError(self, msg): | |
| 62 log.debug('PatchSource._onError from %s %r', self.url, msg) | |
| 63 if not self._fullGraphReceived: | |
| 64 self.connectionFailed.callback(msg) | |
| 65 else: | |
| 66 self.connectionLost.callback(msg) | |
| 67 | |
| 68 def _onFullGraph(self, message): | |
| 69 try: | |
| 70 g = ConjunctiveGraph() | |
| 71 g.parse(StringInputSource(message), format='json-ld') | |
| 72 p = Patch(addGraph=g) | |
| 73 self._sendPatch(p, fullGraph=True) | |
|
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
74 except Exception: |
| 302 | 75 log.error(traceback.format_exc()) |
| 76 raise | |
| 77 self._fullGraphReceived = True | |
| 78 | |
| 79 def _onPatch(self, message): | |
| 80 try: | |
| 81 p = patchFromJson(message) | |
| 82 self._sendPatch(p, fullGraph=False) | |
| 83 except: | |
| 84 log.error(traceback.format_exc()) | |
| 85 raise | |
| 86 | |
| 87 def _sendPatch(self, p, fullGraph): | |
|
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
88 log.debug('PatchSource %s received patch %s (fullGraph=%s)', |
|
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
89 self.url, p.shortSummary(), fullGraph) |
| 302 | 90 for lis in self._listeners: |
| 91 lis(p, fullGraph=fullGraph) | |
| 92 | |
| 93 def __del__(self): | |
| 94 if self._eventSource: | |
|
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
95 raise ValueError("PatchSource wasn't stopped before del") |
| 302 | 96 |
| 97 class ReconnectingPatchSource(object): | |
| 98 """ | |
| 99 PatchSource api, but auto-reconnects internally and takes listener | |
| 100 at init time to not miss any patches. You'll get another | |
| 101 fullGraph=True patch if we have to reconnect. | |
| 102 | |
| 103 todo: generate connection stmts in here | |
| 104 """ | |
|
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
105 def __init__(self, url, listener, reconnectSecs=60, agent='unset'): |
|
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
106 # type: (str, Any, Any, str) |
| 302 | 107 self.url = url |
| 108 self._stopped = False | |
| 109 self._listener = listener | |
|
429
fcd2c026f51e
exportStats for sending scales data to influxdb
drewp@bigasterisk.com
parents:
351
diff
changeset
|
110 self.reconnectSecs = reconnectSecs |
|
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
111 self.agent = agent |
| 302 | 112 self._reconnect() |
| 113 | |
| 114 def _reconnect(self): | |
| 115 if self._stopped: | |
| 116 return | |
|
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
117 self._ps = PatchSource(self.url, agent=self.agent) |
| 302 | 118 self._ps.addPatchListener(self._onPatch) |
| 119 self._ps.connectionFailed.addCallback(self._onConnectionFailed) | |
| 120 self._ps.connectionLost.addCallback(self._onConnectionLost) | |
| 121 | |
| 122 def _onPatch(self, p, fullGraph): | |
| 123 self._listener(p, fullGraph=fullGraph) | |
| 306 | 124 |
|
439
124c921ad52d
stats->state to make room for greplin stats
drewp@bigasterisk.com
parents:
429
diff
changeset
|
125 def state(self): |
| 306 | 126 return { |
|
439
124c921ad52d
stats->state to make room for greplin stats
drewp@bigasterisk.com
parents:
429
diff
changeset
|
127 'reconnectedPatchSource': self._ps.state(), |
| 306 | 128 } |
| 302 | 129 |
| 130 def stop(self): | |
| 131 self._stopped = True | |
| 132 self._ps.stop() | |
| 133 | |
| 134 def _onConnectionFailed(self, arg): | |
|
429
fcd2c026f51e
exportStats for sending scales data to influxdb
drewp@bigasterisk.com
parents:
351
diff
changeset
|
135 reactor.callLater(self.reconnectSecs, self._reconnect) |
| 302 | 136 |
| 137 def _onConnectionLost(self, arg): | |
|
429
fcd2c026f51e
exportStats for sending scales data to influxdb
drewp@bigasterisk.com
parents:
351
diff
changeset
|
138 reactor.callLater(self.reconnectSecs, self._reconnect) |
|
482
b5abd4fc65a4
UA support, some rewrites from twisted_sse_demo work
drewp@bigasterisk.com
parents:
447
diff
changeset
|
139 |