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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
|
import Auth from './auth.js'
import { randomString } from './utils.js'
export default class {
constructor(graffitiURL="https://graffiti.csail.mit.edu") {
this.graffitiURL = graffitiURL
this.open = false
this.subscriptionData = {}
this.eventTarget = new EventTarget()
}
// CALL THIS BEFORE DOING ANYTHING ELSE
async initialize() {
// Perform authorization
this.authParams = await Auth.connect(this.graffitiURL)
// Rewrite the URL
this.wsURL = new URL(this.graffitiURL)
this.wsURL.host = "app." + this.wsURL.host
if (this.wsURL.protocol == 'https:') {
this.wsURL.protocol = 'wss:'
} else {
this.wsURL.protocol = 'ws:'
}
if (this.authParams.token) {
this.wsURL.searchParams.set("token", this.authParams.token)
}
// And commence connection
this.connect()
}
connect() {
this.ws = new WebSocket(this.wsURL)
this.ws.onmessage = this.onMessage.bind(this)
this.ws.onclose = this.onClose.bind(this)
this.ws.onopen = this.onOpen.bind(this)
}
// authorization functions
logIn() { Auth.logIn(this.graffitiURL) }
logOut() { Auth.logOut() }
get myID() { return this.authParams.myID }
get loggedIn() { return this.authParams.loggedIn }
async onClose() {
this.open = false
console.error("lost connection to graffiti server, attemping reconnect soon...")
await new Promise(resolve => setTimeout(resolve, 2000))
this.connect()
}
async request(msg) {
// Create a random message ID
const messageID = randomString()
// Create a listener for the reply
const dataPromise = new Promise(resolve => {
this.eventTarget.addEventListener(messageID, (e) => {
resolve(e.data)
})
})
// Wait for the socket to open
if (!this.open) {
await new Promise(resolve => {
this.eventTarget.addEventListener("graffitiOpen", () => resolve() )
})
}
// Send the request
msg.messageID = messageID
this.ws.send(JSON.stringify(msg))
// Await the reply
const data = await dataPromise
delete data.messageID
if (data.type == 'error' ) {
throw data
} else {
return data
}
}
onMessage(event) {
const data = JSON.parse(event.data)
if ('messageID' in data) {
// It's a reply
// Forward it back to the sender
const messageEvent = new Event(data.messageID)
messageEvent.data = data
this.eventTarget.dispatchEvent(messageEvent)
} else if (['updates', 'removes'].includes(data.type)) {
// Subscription data
if (data.queryID in this.subscriptionData) {
const sd = this.subscriptionData[data.queryID]
// For each data point, either add or remove it
for (const r of data.results) {
if (data.type == 'updates') {
sd.updateCallback(r)
} else {
sd.removeCallback(r)
}
}
// And update this query's notion of "now"
if (data.complete) {
if (data.historical) {
sd.historyComplete = true
}
if (sd.historyComplete) {
sd.since = data.now
}
}
}
}
}
async update(object, query) {
const data = await this.request({
type: "update",
object, query
})
return data.objectID
}
async remove(objectID) {
await this.request({
type: "remove",
objectID
})
}
async subscribe(
query,
updateCallback,
removeCallback,
since=null,
queryID=null) {
// Create a random query ID
if (!queryID) queryID = randomString()
// Send the request
await this.request({
type: "subscribe",
queryID, query, since
})
// Store the subscription in case of disconnections
this.subscriptionData[queryID] = {
query, since, updateCallback, removeCallback,
historyComplete: false
}
return queryID
}
async unsubscribe(queryID) {
// Remove allocated space
delete this.subscriptionData[queryID]
// And unsubscribe
const data = await this.request({
type: "unsubscribe",
queryID
})
}
async onOpen() {
console.log("connected to the graffiti socket")
this.open = true
this.eventTarget.dispatchEvent(new Event("graffitiOpen"))
// Resubscribe to hanging queries
for (const queryID in this.subscriptionData) {
const sd = this.subscriptionData[queryID]
await this.subscribe(
sd.query,
sd.updateCallback,
sd.removeCallback,
sd.since,
queryID)
}
}
// Adds required fields to an object.
// You should probably call this before 'update'
completeObject(object) {
// Add by/to fields
object._by = this.myID
if ('_to' in object && !Array.isArray(object._to)) {
throw new Error("_to must be an array")
}
// Pre-generate the object's ID if it does not already exist
if (!object._id) object._id = randomString()
}
// Utility function to get a universally unique string
// that represents a particular object
objectUUID(object) {
if (!object._id || !object._by) {
throw {
type: 'error',
content: 'the object you are trying to identify does not have an ID or owner',
object
}
}
return object._id + object._by
}
}
|
