blob: aa890bb741cf7b8abf348677b5f7bf1c945f2f62 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
import { fetchApi } from "../utils"
export const CMD_RESET = 'reset'
export default class WebsocketBackend {
constructor(store) {
this.activeFetches = {}
this.store = store
this.connect()
}
connect() {
this.socket = new WebSocket(location.origin.replace('http', 'ws') + '/updates')
this.socket.addEventListener('open', () => this.onOpen())
this.socket.addEventListener('close', () => this.onClose())
this.socket.addEventListener('message', msg => this.onMessage(JSON.parse(msg.data)))
this.socket.addEventListener('error', error => this.onError(error))
}
onOpen() {
this.fetchData("settings")
this.fetchData("flows")
this.fetchData("events")
}
fetchData(resource) {
let queue = []
this.activeFetches[resource] = queue
fetchApi(`/${resource}`)
.then(res => res.json())
.then(json => {
// Make sure that we are not superseded yet by the server sending a RESET.
if (this.activeFetches[resource] === queue)
this.receive(resource, json)
})
}
onMessage(msg) {
if (msg.cmd === CMD_RESET) {
return this.fetchData(msg.resource)
}
if (msg.resource in this.activeFetches) {
this.activeFetches[msg.resource].push(msg)
} else {
let type = `${msg.resource}_${msg.cmd}`.toUpperCase()
this.store.dispatch({ type, ...msg })
}
}
receive(resource, msg) {
let type = `${resource}_RECEIVE`.toUpperCase()
this.store.dispatch({ type, [resource]: msg })
let queue = this.activeFetches[resource]
delete this.activeFetches[resource]
queue.forEach(msg => this.onMessage(msg))
}
onClose() {
// FIXME
console.error("onClose", arguments)
}
onError() {
// FIXME
console.error("onError", arguments)
}
}
|