-
-
- ''')
-
-
-SEND_FULL_GRAPH = Summary('send_full_graph', 'fullGraph SSE events')
-SEND_PATCH = Summary('send_patch', 'patch SSE events')
-
-
-class CycloneGraphEventsHandler(cyclone.sse.SSEHandler):
- """
- One session with one client.
-
- returns current graph plus future patches to keep remote version
- in sync with ours.
-
- intsead of turning off buffering all over, it may work for this
- response to send 'x-accel-buffering: no', per
- http://nginx.org/en/docs/http/ngx_http_proxy_module.html#proxy_buffering
- """
-
- def __init__(self, application, request, masterGraph):
- cyclone.sse.SSEHandler.__init__(self, application, request)
- self.masterGraph = masterGraph
-
- def bind(self):
- with SEND_FULL_GRAPH.time():
- graphJson = self.masterGraph.asJsonLd()
- log.debug("send fullGraph event: %s", graphJson)
- self.sendEvent(message=graphJson, event=b'fullGraph')
- self.masterGraph.addObserver(self.onPatch)
-
- def onPatch(self, patchJson):
- with SEND_PATCH.time():
- # throttle and combine patches here- ideally we could see how
- # long the latency to the client is to make a better rate choice
- self.sendEvent(message=patchJson, event=b'patch')
-
- def unbind(self):
- self.masterGraph.removeObserver(self.onPatch)
-
- def flush(self):
- if getattr(self.settings, 'allowOrigin'):
- allow = self.settings.allowOrigin
- if allow != b'*':
- raise NotImplementedError()
- self.set_header(b"Access-Control-Allow-Origin", allow)
- return CycloneGraphEventsHandler.flush(self)
\ No newline at end of file
diff -r 1eac25669333 -r e11d407c46f8 patchablegraph_handler.py
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/patchablegraph_handler.py Sat Apr 23 23:58:41 2022 -0700
@@ -0,0 +1,135 @@
+import html
+import logging
+from typing import Callable, Tuple
+
+from prometheus_client import Summary
+from rdflib.namespace import NamespaceManager
+from sse_starlette import ServerSentEvent
+from sse_starlette.sse import EventSourceResponse
+from starlette.requests import Request
+from starlette.responses import Response
+
+from patchablegraph import PatchableGraph
+
+log = logging.getLogger('patchablegraph')
+
+SEND_SIMPLE_GRAPH = Summary('send_simple_graph', 'calls to _writeGraphResponse')
+SEND_FULL_GRAPH = Summary('send_full_graph', 'fullGraph SSE events')
+SEND_PATCH = Summary('send_patch', 'patch SSE events')
+
+
+def StaticGraph(masterGraph: PatchableGraph) -> Callable[[Request], Response]:
+ """
+ e.g.
+ Route('/graph/environment', StaticGraph(masterGraph)),
+ """
+
+ @SEND_SIMPLE_GRAPH.time()
+ def handle(request: Request) -> Response:
+ ctype, content = _writeGraphResponse(masterGraph, request.headers.get('accept', default=''))
+ r = Response(content=content)
+ r.media_type = ctype
+ return r
+
+ return handle
+
+
+def _writeGraphResponse(masterGraph: PatchableGraph, accept: str) -> Tuple[str, bytes]:
+ if accept == 'application/nquads':
+ return 'application/nquads', masterGraph.serialize(format='nquads')
+ elif accept == 'application/ld+json':
+ return 'application/ld+json', masterGraph.serialize(format='json-ld', indent=2)
+ else:
+ if accept.startswith('text/html'):
+ return _writeGraphForBrowser(masterGraph)
+ return 'application/x-trig', masterGraph.serialize(format='trig')
+
+
+def _writeGraphForBrowser(masterGraph: PatchableGraph) -> Tuple[str, bytes]:
+ # We think this is a browser, so respond with a live graph view
+ # (todo)
+ out = (b'''
+
''')
+
+ ns = NamespaceManager(masterGraph._graph)
+ # maybe these could be on the PatchableGraph instance
+ ns.bind('ex', 'http://example.com/')
+ ns.bind('', 'http://projects.bigasterisk.com/room/')
+ ns.bind("rdf", "http://www.w3.org/1999/02/22-rdf-syntax-ns#")
+ ns.bind("xsd", "http://www.w3.org/2001/XMLSchema#")
+
+ for s, p, o, g in sorted(masterGraph._graph.quads()):
+ g = g.identifier
+ nquadLine = f'{s.n3(ns)} {p.n3(ns)} {o.n3(ns)} {g.n3(ns)} .\n'
+ out += html.escape(nquadLine).encode('utf8')
+
+ out += b'''
+
+
+
+ '''
+ return 'text/html', out
+
+
+def GraphEvents(masterGraph: PatchableGraph):
+ """
+ e.g.
+ Route('/graph/environment/events', GraphEvents(masterGraph)),
+ """
+
+ async def generateEvents():
+ events = masterGraph.subscribeToPatches()
+ while True: # we'll get cancelled by EventSourceResponse when the conn drops
+ etype, data = await events.get()
+ # Are there more to get? We might throttle and combine patches here- ideally we could see how
+ # long the latency to the client is to make a better rate choice
+ metric = {'graph': SEND_FULL_GRAPH, 'patch': SEND_PATCH}[etype]
+ with metric.time():
+ yield ServerSentEvent(event=etype, data=data)
+
+ async def handle(request: Request):
+ """
+ One session with one client.
+
+ returns current graph plus future patches to keep remote version
+ in sync with ours.
+
+ instead of turning off buffering all over, it may work for this
+ response to send 'x-accel-buffering: no', per
+ http://nginx.org/en/docs/http/ngx_http_proxy_module.html#proxy_buffering
+ """
+ return EventSourceResponse(generateEvents())
+
+ return handle
diff -r 1eac25669333 -r e11d407c46f8 setup.py
--- a/setup.py Sat Apr 23 23:48:55 2022 -0700
+++ b/setup.py Sat Apr 23 23:58:41 2022 -0700
@@ -2,19 +2,17 @@
setup(
name='patchablegraph',
- version='0.19.0',
+ version='1.0.0',
packages=['patchablegraph'],
package_dir={'patchablegraph': ''},
install_requires=[
- 'cyclone',
- 'twisted',
'rdflib >= 6.0.1',
'rdfdb >= 0.8.0',
'prometheus_client',
- 'cycloneerr',
- 'twisted_sse >= 0.3.0',
+ 'sse_starlette>=0.10.3',
+ 'starlette>=0.19.1'
],
- url='https://projects.bigasterisk.com/patchablegraph/patchablegraph-0.19.0.tar.gz',
+ url='https://projects.bigasterisk.com/patchablegraph/patchablegraph-1.0.0.tar.gz',
author='Drew Perttula',
author_email='drewp@bigasterisk.com',
)