Mercurial > code > home > repos > streamed-graph

comparison src/streamed_graph_client.ts @ 36:8b4dc9e87b56

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
reindent to 2-spaces with prettier
author drewp@bigasterisk.com
date Sat, 28 Dec 2019 02:24:55 -0800
parents 9ec3cbc8791a
children 163b4339804d
comparison
equal deleted inserted replaced
35:29d8ed02a275 36:8b4dc9e87b56
1 // from /my/site/homepage/www/rdf/streamed-graph.js 1 // from /my/site/homepage/www/rdf/streamed-graph.js
2 2
3 import { eachJsonLdQuad } from "./json_ld_quads"; 3 import { eachJsonLdQuad } from "./json_ld_quads";
4 import { N3Store } from 'n3'; 4 import { N3Store } from "n3";
5 import { Store } from 'n3'; 5 import { Store } from "n3";
6 6
7 export class StreamedGraphClient { 7 export class StreamedGraphClient {
8 // holds a n3 Store, which is synced to a server-side 8 // holds a n3 Store, which is synced to a server-side
9 // store that sends patches over SSE 9 // store that sends patches over SSE
10 10
11 onStatus: (msg: string) => void; 11 onStatus: (msg: string) => void;
12 onGraphChanged: () => void; 12 onGraphChanged: () => void;
13 store: N3Store; 13 store: N3Store;
14 events!: EventSource; 14 events!: EventSource;
15 constructor( 15 constructor(
16 eventsUrl: string, 16 eventsUrl: string,
17 onGraphChanged: () => void, 17 onGraphChanged: () => void,
18 onStatus: (status: string) => void, 18 onStatus: (status: string) => void,
19 prefixes: Array<Record<string, string>>, 19 prefixes: Array<Record<string, string>>,
20 staticGraphUrls: Array<string>) { 20 staticGraphUrls: Array<string>
21 console.log('new StreamedGraph', eventsUrl); 21 ) {
22 this.onStatus = onStatus; 22 console.log("new StreamedGraph", eventsUrl);
23 this.onGraphChanged = onGraphChanged; 23 this.onStatus = onStatus;
24 this.onStatus('startup...'); 24 this.onGraphChanged = onGraphChanged;
25 this.onStatus("startup...");
25 26
26 this.store = new Store(); 27 this.store = new Store();
27 28
28 // // Object.keys(prefixes).forEach((prefix) => { 29 // // Object.keys(prefixes).forEach((prefix) => {
29 // // this.store.setPrefix(prefix, prefixes[prefix]); 30 // // this.store.setPrefix(prefix, prefixes[prefix]);
30 // // }); 31 // // });
31 32
32 this.connect(eventsUrl); 33 this.connect(eventsUrl);
33 this.reconnectOnWake(); 34 this.reconnectOnWake();
34 35
35 // staticGraphUrls.forEach((url) => { 36 // staticGraphUrls.forEach((url) => {
36 // fetch(url).then((response) => response.text()) 37 // fetch(url).then((response) => response.text())
37 // .then((body) => { 38 // .then((body) => {
38 // // parse with n3, add to output 39 // // parse with n3, add to output
39 // }); 40 // });
40 // }); 41 // });
42 }
41 43
44 reconnectOnWake() {
45 // it's not this, which fires on every mouse-in on a browser window, and doesn't seem to work for screen-turned-back-on
46 //window.addEventListener('focus', function() { this.connect(eventsUrl); }.bind(this));
47 }
48
49 connect(eventsUrl: string) {
50 // need to exit here if this obj has been replaced
51
52 this.onStatus("start connect...");
53 this.close();
54 if (this.events && this.events.readyState != EventSource.CLOSED) {
55 this.onStatus("zombie");
56 throw new Error("zombie eventsource");
42 } 57 }
43 58
44 reconnectOnWake() { 59 this.events = new EventSource(eventsUrl);
45 // it's not this, which fires on every mouse-in on a browser window, and doesn't seem to work for screen-turned-back-on 60
46 //window.addEventListener('focus', function() { this.connect(eventsUrl); }.bind(this)); 61 this.events.addEventListener("error", ev => {
62 // todo: this is piling up tons of retries and eventually multiple connections
63 // this.testEventUrl(eventsUrl);
64 this.onStatus("connection lost- retrying");
65 setTimeout(() => {
66 requestAnimationFrame(() => {
67 this.connect(eventsUrl);
68 });
69 }, 3000);
70 });
71
72 this.events.addEventListener("fullGraph", async ev => {
73 this.onStatus("sync- full graph update");
74 await this.replaceFullGraph((ev as MessageEvent).data);
75 this.onStatus(`synced ${this.store.size}`);
76 this.onGraphChanged();
77 });
78
79 this.events.addEventListener("patch", async ev => {
80 this.onStatus("sync- updating");
81 await this.patchGraph((ev as MessageEvent).data);
82 this.onStatus(`synced ${this.store.size}`);
83 this.onGraphChanged();
84 });
85 this.onStatus("connecting...");
86 }
87
88 // these need some locks
89 async replaceFullGraph(jsonLdText: string) {
90 this.store = new Store();
91 await eachJsonLdQuad(
92 JSON.parse(jsonLdText),
93 this.store.addQuad.bind(this.store)
94 );
95 }
96
97 async patchGraph(patchJson: string) {
98 var patch = JSON.parse(patchJson).patch;
99
100 await eachJsonLdQuad(patch.deletes, this.store.removeQuad.bind(this.store));
101 await eachJsonLdQuad(patch.adds, this.store.addQuad.bind(this.store));
102 }
103
104 close() {
105 if (this.events) {
106 this.events.close();
47 } 107 }
108 }
48 109
49 connect(eventsUrl: string) { 110 async testEventUrl(eventsUrl: string): Promise<void> {
50 // need to exit here if this obj has been replaced 111 return new Promise<void>((resolve, reject) => {
51 112 this.onStatus("testing connection");
52 this.onStatus('start connect...'); 113 fetch(eventsUrl, {
53 this.close(); 114 method: "HEAD",
54 if (this.events && this.events.readyState != EventSource.CLOSED) { 115 credentials: "include"
55 this.onStatus('zombie'); 116 })
56 throw new Error("zombie eventsource"); 117 .then(value => {
57 } 118 if (value.status == 403) {
58 119 reject();
59 this.events = new EventSource(eventsUrl); 120 return;
60 121 }
61 this.events.addEventListener('error', (ev) => { 122 resolve();
62 // todo: this is piling up tons of retries and eventually multiple connections 123 })
63 // this.testEventUrl(eventsUrl); 124 .catch(err => {
64 this.onStatus('connection lost- retrying'); 125 reject();
65 setTimeout(() => {
66 requestAnimationFrame(() => {
67 this.connect(eventsUrl);
68 });
69 }, 3000);
70 }); 126 });
71 127 });
72 this.events.addEventListener('fullGraph', async (ev) => { 128 }
73 this.onStatus('sync- full graph update');
74 await this.replaceFullGraph((ev as MessageEvent).data);
75 this.onStatus(`synced ${this.store.size}`);
76 this.onGraphChanged();
77 });
78
79 this.events.addEventListener('patch', async (ev) => {
80 this.onStatus('sync- updating');
81 await this.patchGraph((ev as MessageEvent).data);
82 this.onStatus(`synced ${this.store.size}`);
83 this.onGraphChanged();
84 });
85 this.onStatus('connecting...');
86 }
87
88 // these need some locks
89 async replaceFullGraph(jsonLdText: string) {
90 this.store = new Store();
91 await eachJsonLdQuad(JSON.parse(jsonLdText),
92 this.store.addQuad.bind(this.store));
93 }
94
95 async patchGraph(patchJson: string) {
96 var patch = JSON.parse(patchJson).patch;
97
98 await eachJsonLdQuad(patch.deletes,
99 this.store.removeQuad.bind(this.store));
100 await eachJsonLdQuad(patch.adds,
101 this.store.addQuad.bind(this.store));
102 }
103
104 close() {
105 if (this.events) {
106 this.events.close();
107 }
108 }
109
110 async testEventUrl(eventsUrl: string): Promise<void> {
111 return new Promise<void>((resolve, reject) => {
112 this.onStatus('testing connection');
113 fetch(eventsUrl, {
114 method: "HEAD",
115 credentials: "include",
116 }).then((value) => {
117 if (value.status == 403) {
118 reject();
119 return;
120 }
121 resolve();
122 }).catch((err) => {
123 reject();
124 });
125 });
126 }
127
128 } 129 }