optimized list protocol
Massimo Melina committed
Jul 12, 2023 at 00:29 UTC
042a3563e0191f8f8d12965ce43b87d3631d9b56
5 files changed
+119
-94
admin/src/api.ts
+50
-41
@@ -43,9 +43,9 @@ export function useApiList<T=any>(cmd:string|Falsy, params: Dict={}, { map=((x:a
43
const idGenerator = useRef(0)
44
useEffect(() => {
45
if (!cmd) return
46
- const buffer: T[] = []
46
+ const bufferAdd: T[] = []
47
const apply = _.debounce(() => {
48
- const chunk = buffer.splice(0, Infinity)
48
+ const chunk = bufferAdd.splice(0, Infinity)
49
if (chunk.length)
50
setList(list => [ ...list, ...chunk ])
51
}, 1000, { maxWait: 1000 })
@@ -65,59 +65,68 @@ export function useApiList<T=any>(cmd:string|Falsy, params: Dict={}, { map=((x:a
65
case 'closed':
66
return stop()
67
case 'msg':
68
- wantArray(data).forEach(entry => {
69
- if (entry === 'ready') {
68
+ const removeOnList: ReturnType<typeof _.matches>[] = []
69
+ const updateOnList: [object,object][] = []
70
+ wantArray(data).forEach(msg => {
71
+ if (!Array.isArray(msg))
72
+ return console.debug('illegal list packet', msg)
73
+ const [op, par] = msg
74
+ if (op === 'ready') {
75
apply.flush()
76
setInitializing(false)
77
return
78
}
74
- if (entry.error) {
75
- if (entry.error === 401)
76
- state.loginRequired = entry.any !== false || 403
79
+ if (op === 'error') {
80
+ if (par === 401)
81
+ state.loginRequired = msg[2].any !== false || 403
82
else
78
- setError(err2msg(entry.error))
83
+ setError(err2msg(par))
84
return
85
}
81
- if (entry.props)
82
- return setProps(entry.props)
83
- if (entry.add) {
84
- const rec = map(entry.add)
85
- rec.id ??= idGenerator.current = Math.max(idGenerator.current, Date.now()) + .001
86
- buffer.push(rec)
86
+ if (op === 'props')
87
+ return setProps(par)
88
+ if (op === 'add') {
89
+ const mappedPar = map(par)
90
+ mappedPar.id ??= idGenerator.current = Math.max(idGenerator.current, Date.now()) + .001
91
+ bufferAdd.push(mappedPar)
92
apply()
93
return
94
}
90
- if (entry.remove) {
91
- const matchOnList: ReturnType<typeof _.matches>[] = []
92
- // first remove from the buffer
93
- for (const key of entry.remove) {
94
- const match1 = _.matches(key)
95
- if (_.isEmpty(_.remove(buffer, match1)))
96
- matchOnList.push(match1)
97
- }
98
- // then work the hooked state
99
- if (_.isEmpty(matchOnList))
100
- return
101
- setList(list => {
102
- const filtered = list.filter(rec => !matchOnList.some(match1 => match1(rec)))
103
- return filtered.length < list.length ? filtered : list // avoid unnecessary changes
104
- })
95
+ if (op === 'remove') {
96
+ const match = _.matches(par)
97
+ if (_.isEmpty(_.remove(bufferAdd, match))) // first remove from the buffer
98
+ removeOnList.push(match)
99
return
100
}
107
- if (entry.update) {
108
- apply.flush() // avoid treating buffer
109
- setList(list => {
110
- const modified = [...list]
111
- for (const { search, change } of entry.update) {
112
- const idx = modified.findIndex(_.matches(search))
113
- if (idx >= 0)
114
- modified[idx] = { ...modified[idx], ...change }
115
- }
116
- return modified
117
- })
101
+ if (op === 'update') {
102
+ const change = msg[2]
103
+ const found = _.find(bufferAdd, par)
104
+ if (found)
105
+ return Object.assign(found, change)
106
+ updateOnList.push([par, change])
107
return
108
}
120
- console.debug('unknown api event', type, entry)
109
+ console.debug('unknown list api', op)
110
+ })
111
+ setList(list => {
112
+ let ret = list
113
+ let copy // optimization: remember if we already made a copy
114
+ if (removeOnList.length) {
115
+ copy = list.filter(rec => !removeOnList.some(match1 => match1(rec)))
116
+ if (copy.length < list.length) // avoid unnecessary render
117
+ ret = copy
118
+ }
119
+
120
+ if (updateOnList.length) {
121
+ for (const [search, change] of updateOnList) {
122
+ const foundAt = _.findIndex(ret, search)
123
+ if (foundAt < 0) continue
124
+ if (ret === list)
125
+ ret = copy ?? list.slice()
126
+ ret[foundAt] = { ...ret[foundAt], ...change }
127
+ }
128
+ }
129
+ return ret
130
})
131
if (src?.readyState === src?.CLOSED)
132
stop()
frontend/src/useFetchList.ts
+6
-6
@@ -78,7 +78,8 @@ export default function useFetchList() {
78
case 'msg':
79
state.loginRequired = false
80
for (const entry of data) {
81
- const { error } = entry
81
+ const [op, par] = entry
82
+ const error = op === 'error' && par
83
if (error === 405) { // "method not allowed" happens when we try to directly access an unauthorized file, and we get a login prompt, and then file_list the file (because we didn't know it was file or folder)
84
state.messageOnly = t('upload_starting', "Your download should now start")
85
window.location.reload() // reload will start the download, because now we got authenticated
@@ -95,15 +96,14 @@ export default function useFetchList() {
96
}
97
if (!desiredPath.endsWith('/')) // now we know it was a folder for sure
98
return navigate(desiredPath + '/')
98
- if (entry.props) {
99
- Object.assign(state, _.pick(entry.props, ['can_upload', 'can_delete', 'accept']))
99
+ if (op === 'props') {
100
+ Object.assign(state, _.pick(par, ['can_upload', 'can_delete', 'accept']))
101
continue
102
}
103
state.can_upload ??= false
104
state.can_delete ??= false
104
- const { add } = entry
105
- if (add)
106
- buffer.push(new DirEntry(add.n, add))
105
+ if (op === 'add')
106
+ buffer.push(new DirEntry(par.n, par))
107
}
108
if (src?.readyState === src?.CLOSED)
109
return state.stopSearch?.()
src/api.file_list.ts
+1
-1
@@ -46,7 +46,7 @@ export const file_list: ApiHandler = async ({ uri, offset, limit, search, c, sse
46
return { ...props, list: await asyncGeneratorToArray(produceEntries()) }
47
setTimeout(async () => {
48
if (can_upload || can_delete)
49
- list.custom({ props })
49
+ list.props(props)
50
for await (const entry of produceEntries())
51
list.add(entry)
52
list.close()
src/api.monitor.ts
+32
-33
@@ -2,7 +2,7 @@
2
3
import _ from 'lodash'
4
import { Connection, getConnections } from './connections'
5
-import { pendingPromise, wait } from './misc'
5
+import { pendingPromise, typedKeys, wait } from './misc'
6
import { ApiHandlers, SendListReadable } from './apiMiddleware'
7
import Koa from 'koa'
8
import { totalGot, totalInSpeed, totalOutSpeed, totalSent } from './throttler'
@@ -25,49 +25,44 @@ const apis: ApiHandlers = {
25
},
26
27
get_connections({}, ctx) {
28
- const list = new SendListReadable({ addAtStart: getConnections().map(c => serializeConnection(c)) })
28
+ const sent = Symbol('sent')
29
+ const list = new SendListReadable({
30
+ addAtStart: getConnections().map(c =>
31
+ !ignore(c) && (c[sent] = serializeConnection(c))).filter(Boolean),
32
+ onEnd() {
33
+ for (const c of getConnections())
34
+ delete c[sent]
35
+ }
36
+ })
37
type Change = Partial<Omit<Connection,'ip'>>
30
- const throttledUpdate = _.throttle(update, 1000/20) // try to avoid clogging with updates
31
- const state = Symbol('state') // undefined=added, Timeout=add-pending, false=removed
38
list.props({ you: ctx.ip })
39
return list.events(ctx, {
40
connection(conn: Connection) {
35
- conn[state] = setTimeout(() => add(conn), 100)
41
+ if (ignore(conn)) return
42
+ list.add(conn[sent] = serializeConnection(conn))
43
},
44
connectionClosed(conn: Connection) {
38
- if (cancel(conn)) return
45
+ if (ignore(conn)) return
46
list.remove(getConnAddress(conn))
40
- conn[state] = false
47
+ delete conn[sent]
48
},
49
connectionUpdated(conn: Connection, change: Change) {
43
- if (!change.ctx)
44
- return throttledUpdate(conn, change)
45
-
46
- Object.assign(change, fromCtx(change.ctx))
47
- change.ctx = undefined
48
- if (!add(conn))
49
- throttledUpdate(conn, change)
50
+ if (ignore(conn) || ignore(change as any) || !conn[sent]) return
51
+ if (change.ctx) {
52
+ Object.assign(change, fromCtx(change.ctx))
53
+ change.ctx = undefined
54
+ }
55
+ // avoid sending non-changes
56
+ const last = conn[sent]
57
+ for (const k of typedKeys(change))
58
+ if (change[k] === last[k])
59
+ delete change[k]
60
+ if (_.isEmpty(change)) return
61
+ Object.assign(last, change)
62
+ list.update(getConnAddress(conn), change)
63
},
64
})
65
53
- function add(conn: Connection) {
54
- if (!cancel(conn)) return
55
- list.add(serializeConnection(conn))
56
- return true
57
- }
58
-
59
- function cancel(conn: Connection) {
60
- if (!conn[state]) return
61
- clearTimeout(conn[state])
62
- conn[state] = undefined
63
- return true
64
- }
65
-
66
- function update(conn: Connection, change: Change) {
67
- if (conn[state] === false) return
68
- list.update(getConnAddress(conn), change)
69
- }
70
-
66
function serializeConnection(conn: Connection) {
67
const { socket, started, secure } = conn
68
return {
@@ -101,7 +96,7 @@ const apis: ApiHandlers = {
96
inSpeed: totalInSpeed,
97
got: totalGot,
98
sent: totalSent,
104
- connections: getConnections().length
99
+ connections: _.sumBy(getConnections(), x => ignore(x) ? 0 : 1),
100
}
101
await wait(1000)
102
}
@@ -110,6 +105,10 @@ const apis: ApiHandlers = {
105
106
export default apis
107
108
+function ignore(conn: Connection) {
109
+ return false //conn.socket && isLocalHost(conn)
110
+}
111
+
112
function getConnAddress(conn: Connection) {
113
return {
114
ip: conn.ip,
src/apiMiddleware.ts
+30
-13
@@ -82,18 +82,19 @@ export class SendListReadable<T> extends Readable {
82
protected lastError: string | number | undefined
83
protected buffer: any[] = []
84
protected processBuffer: _.DebouncedFunc<any>
85
- constructor({ addAtStart, doAtStart, bufferTime }:{ bufferTime?: number, addAtStart?: T[], doAtStart?: SendListFunc<T> }={}) {
85
+ constructor({ addAtStart, doAtStart, bufferTime, onEnd }:{ bufferTime?: number, addAtStart?: T[], doAtStart?: SendListFunc<T>, onEnd?: SendListFunc<T> }={}) {
86
super({ objectMode: true, read(){} })
87
if (!bufferTime)
88
- bufferTime = 100
88
+ bufferTime = 200
89
this.processBuffer = _.debounce(() => {
90
this.push(this.buffer)
91
this.buffer = []
92
}, bufferTime, { maxWait: bufferTime })
93
- this.on('end', () =>
94
- this.destroy())
95
- if (doAtStart)
96
- setTimeout(() => doAtStart(this)) // work later, when list object has been received by Koa
93
+ this.on('end', () => {
94
+ onEnd?.(this)
95
+ this.destroy()
96
+ })
97
+ setTimeout(() => doAtStart?.(this)) // work later, when list object has been received by Koa
98
if (addAtStart) {
99
for (const x of addAtStart)
100
this.add(x)
@@ -108,25 +109,41 @@ export class SendListReadable<T> extends Readable {
109
this.processBuffer()
110
}
111
add(rec: T | T[]) {
111
- this._push({ add: rec })
112
+ this._push(['add', rec])
113
}
113
- remove(key: Partial<T>) {
114
- this._push({ remove: [key] })
114
+ remove(search: Partial<T>) {
115
+ const match = _.matches(search)
116
+ const idx = _.findIndex(this.buffer, x => match(x[1]))
117
+ const found = this.buffer[idx]
118
+ const op = found?.[0]
119
+ if (op === 'remove') return
120
+ if (found) {
121
+ this.buffer.splice(idx, 1)
122
+ if (op === 'add') return
123
+ }
124
+ this._push(['remove', search])
125
}
126
update(search: Partial<T>, change: Partial<T>) {
117
- this._push({ update:[{ search, change }] })
127
+ if (_.isEmpty(change)) return
128
+ const match = _.matches(search)
129
+ const found = _.find(this.buffer, x => match(x[1]))
130
+ const op = found?.[0]
131
+ if (op === 'remove') return
132
+ if (op === 'add' || op === 'update')
133
+ return Object.assign(found[op === 'add' ? 1 : 2], change)
134
+ return this._push(['update', search, change])
135
}
136
ready() { // useful to indicate the end of an initial phase, but we leave open for updates
120
- this._push('ready')
137
+ this._push(['ready'])
138
}
139
custom(data: any) {
140
this._push(data)
141
}
142
props(props: object) {
126
- this._push({ props })
143
+ this._push(['props', props])
144
}
145
error(msg: NonNullable<typeof this.lastError>, close=false, props?: object) {
129
- this._push({ error: msg, ...props })
146
+ this._push(['error', msg, props])
147
this.lastError = msg
148
if (close)
149
this.close()