Mercurial > code > home > repos > homeauto

comparison lib/patchablegraph/patchablegraph.py @ 704:74ad667f8c5b

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
reformat Ignore-this: e1fe14c4b973c3c1fc2714b988ecc7c8
author drewp@bigasterisk.com
date Mon, 03 Feb 2020 00:51:18 -0800
parents 83ccc9ba90ea
children b41247c7b080
comparison
equal deleted inserted replaced
703:83ccc9ba90ea 704:74ad667f8c5b
69 'adds': from_rdf(_graphFromQuads2(p.addQuads)), 69 'adds': from_rdf(_graphFromQuads2(p.addQuads)),
70 'deletes': from_rdf(_graphFromQuads2(p.delQuads)), 70 'deletes': from_rdf(_graphFromQuads2(p.delQuads)),
71 }}) 71 }})
72 patchAsJson = jsonFromPatch # deprecated name 72 patchAsJson = jsonFromPatch # deprecated name
73 73
74 74
75 def patchFromJson(j): 75 def patchFromJson(j):
76 body = json.loads(j)['patch'] 76 body = json.loads(j)['patch']
77 a = ConjunctiveGraph() 77 a = ConjunctiveGraph()
78 a.parse(StringInputSource(json.dumps(body['adds']).encode('utf8')), format='json-ld') 78 a.parse(StringInputSource(json.dumps(body['adds']).encode('utf8')), format='json-ld')
79 d = ConjunctiveGraph() 79 d = ConjunctiveGraph()
126 _observersAdded = scales.IntStat('observers/added') 126 _observersAdded = scales.IntStat('observers/added')
127 def addObserver(self, onPatch): 127 def addObserver(self, onPatch):
128 self._observers.append(onPatch) 128 self._observers.append(onPatch)
129 self._currentObservers = len(self._observers) 129 self._currentObservers = len(self._observers)
130 self._observersAdded += 1 130 self._observersAdded += 1
131 131
132 def removeObserver(self, onPatch): 132 def removeObserver(self, onPatch):
133 try: 133 try:
134 self._observers.remove(onPatch) 134 self._observers.remove(onPatch)
135 except ValueError: 135 except ValueError:
136 pass 136 pass
137 self._currentObservers = len(self._observers) 137 self._currentObservers = len(self._observers)
138 138
139 def setToGraph(self, newGraph): 139 def setToGraph(self, newGraph):
140 self.patch(Patch.fromDiff(self._graph, newGraph)) 140 self.patch(Patch.fromDiff(self._graph, newGraph))
141 141
142 _sendSimpleGraph = scales.PmfStat('serve/simpleGraph') 142 _sendSimpleGraph = scales.PmfStat('serve/simpleGraph')
143 _sendFullGraph = scales.PmfStat('serve/events/sendFull') 143 _sendFullGraph = scales.PmfStat('serve/events/sendFull')
144 _sendPatch = scales.PmfStat('serve/events/sendPatch') 144 _sendPatch = scales.PmfStat('serve/events/sendPatch')
145 145
146 class CycloneGraphHandler(PrettyErrorHandler, cyclone.web.RequestHandler): 146 class CycloneGraphHandler(PrettyErrorHandler, cyclone.web.RequestHandler):
147 def initialize(self, masterGraph): 147 def initialize(self, masterGraph):
148 self.masterGraph = masterGraph 148 self.masterGraph = masterGraph
149 149
150 def get(self): 150 def get(self):
151 with self.masterGraph._sendSimpleGraph.time(): 151 with self.masterGraph._sendSimpleGraph.time():
152 writeGraphResponse(self, self.masterGraph, 152 writeGraphResponse(self, self.masterGraph,
153 self.request.headers.get('accept')) 153 self.request.headers.get('accept'))
154 154
155 155
156 class CycloneGraphEventsHandler(cyclone.sse.SSEHandler): 156 class CycloneGraphEventsHandler(cyclone.sse.SSEHandler):
157 """ 157 """
158 One session with one client. 158 One session with one client.
159 159
160 returns current graph plus future patches to keep remote version 160 returns current graph plus future patches to keep remote version
161 in sync with ours. 161 in sync with ours.
162 162
163 intsead of turning off buffering all over, it may work for this 163 intsead of turning off buffering all over, it may work for this
164 response to send 'x-accel-buffering: no', per 164 response to send 'x-accel-buffering: no', per
178 def onPatch(self, patchJson): 178 def onPatch(self, patchJson):
179 with self.masterGraph._sendPatch.time(): 179 with self.masterGraph._sendPatch.time():
180 # throttle and combine patches here- ideally we could see how 180 # throttle and combine patches here- ideally we could see how
181 # long the latency to the client is to make a better rate choice 181 # long the latency to the client is to make a better rate choice
182 self.sendEvent(message=patchJson, event=b'patch') 182 self.sendEvent(message=patchJson, event=b'patch')
183 183
184 def unbind(self): 184 def unbind(self):
185 self.masterGraph.removeObserver(self.onPatch) 185 self.masterGraph.removeObserver(self.onPatch)
186