fix: file list looping forever if empty
Massimo Melina committed
Feb 8, 2022 at 19:43 UTC
ba4705c600881de07681597de508882ed6f23429
6 files changed
+82
-89
src/adminApis.ts
+18
-21
@@ -1,4 +1,4 @@
1
-import { apiEmitter, ApiHandlers } from './apis'
1
+import { ApiHandlers } from './apis'
2
import { getWholeConfig, setConfig } from './config'
3
import { getStatus } from './listen'
4
import { app, HFS_STARTED } from './index'
@@ -6,7 +6,7 @@ import { Server } from 'http'
6
import vfsApis from './api.vfs'
7
import accountsApis from './api.accounts'
8
import { Connection, getConnections } from './connections'
9
-import { onOffMap, pendingPromise } from './misc'
9
+import { generatorAsCallback, onOffMap, pendingPromise } from './misc'
10
11
export const adminApis: ApiHandlers = {
12
@@ -49,23 +49,21 @@ export const adminApis: ApiHandlers = {
49
return { result: Boolean(c) }
50
},
51
52
- get_connections: apiEmitter(async ({ send, onClose }) => {
53
- getConnections().forEach(add)
54
- onClose(
55
- onOffMap(app, {
56
- connection: add,
57
- connectionClosed(conn: Connection) {
58
- send({ remove: [ serializeConnection(conn, true) ] })
59
- },
60
- connectionUpdated(conn: Connection, change: Partial<Connection>) {
61
- send({ update: [{ search: serializeConnection(conn, true), change }] })
62
- },
63
- })
64
- )
65
-
66
- function add(conn: Connection) {
67
- send({ add: serializeConnection(conn) })
68
- }
52
+ async *get_connections({}, ctx) {
53
+ for (const conn of getConnections())
54
+ yield { add: serializeConnection(conn) }
55
+ yield* generatorAsCallback(wrapper =>
56
+ ctx.res.once('close', // as connection is closed, call the callback returned by onOffMap that uninstalls the listener
57
+ onOffMap(app, {
58
+ connection: conn => wrapper.callback({ add: serializeConnection(conn) }),
59
+ connectionClosed(conn: Connection) {
60
+ wrapper.callback({ remove: [ serializeConnection(conn, true) ] })
61
+ },
62
+ connectionUpdated(conn: Connection, change: Partial<Connection>) {
63
+ wrapper.callback({ update: [{ search: serializeConnection(conn, true), change }] })
64
+ },
65
+ })
66
+ ) )
67
68
function serializeConnection(conn: Connection, minimal?:true) {
69
const { socket, started, secure, got } = conn
@@ -79,7 +77,6 @@ export const adminApis: ApiHandlers = {
77
}
78
79
}
82
-
83
- })
80
+ }
81
82
}
src/api.file_list.ts
+5
-13
@@ -2,9 +2,8 @@ import { getNodeName, vfs, VfsNode, walkNode } from './vfs'
2
import { ApiError, ApiHandler } from './apis'
3
import { stat } from 'fs/promises'
4
import { mapPlugins } from './plugins'
5
-import { dirTraversal, pattern2filter } from './misc'
5
+import { asyncGeneratorToArray, dirTraversal, filterMapGenerator, pattern2filter } from './misc'
6
import { FORBIDDEN } from './const'
7
-import EventEmitter from 'events'
7
8
export const file_list:ApiHandler = async ({ path, offset, limit, search, omit, sse }, ctx) => {
9
let node = await vfs.urlToNode(path || '/', ctx)
@@ -21,12 +20,10 @@ export const file_list:ApiHandler = async ({ path, offset, limit, search, omit,
20
const filter = pattern2filter(search)
21
const walker = walkNode(node, ctx, search ? Infinity : 0)
22
const onDirEntryHandlers = mapPlugins(plug => plug.onDirEntry)
24
- const emitter = sse && new EventEmitter()
25
- const res = produceEntries()
26
- return emitter || { list: await res }
23
+ return sse ? filterMapGenerator(produceEntries(), async (entry) => ({ entry })) // wrap entry in an object
24
+ : { list: await asyncGeneratorToArray(produceEntries()) }
25
28
- async function produceEntries() {
29
- const list = []
26
+ async function* produceEntries() {
27
for await (const sub of walker) {
28
if (ctx.aborted) break
29
if (!filter(getNodeName(sub)))
@@ -53,15 +50,10 @@ export const file_list:ApiHandler = async ({ path, offset, limit, search, omit,
50
entry.m = entry.c
51
delete entry.c
52
}
56
- if (emitter)
57
- emitter.emit('data', { entry })
58
- else
59
- list.push(entry)
53
+ yield entry
54
if (limit && !--limit)
55
break
56
}
63
- emitter?.emit('end')
64
- return list
57
}
58
}
59
src/api.vfs.ts
+15
-17
@@ -1,7 +1,7 @@
1
import { getNodeName, nodeIsDirectory, vfs, VfsNode, VfsNodeType } from './vfs'
2
import _ from 'lodash'
3
import { stat } from 'fs/promises'
4
-import { apiEmitter, ApiError, ApiHandlers } from './apis'
4
+import { ApiError, ApiHandlers } from './apis'
5
import { dirname } from 'path'
6
import { saveConfigAsap } from './config'
7
import glob from 'fast-glob'
@@ -94,18 +94,18 @@ const apis: ApiHandlers = {
94
return { path: process.cwd() }
95
},
96
97
- ls: apiEmitter(async ({ send, end, ctx, params:{ path } }) => {
98
- try {
99
- if (!path && isWindows()) {
100
- try {
101
- for (const n of await getDrives())
102
- send({ add: { n } })
103
- }
104
- catch(error) {
105
- console.debug(error)
106
- }
107
- return
97
+ async *ls({ path }, ctx) {
98
+ if (!path && isWindows()) {
99
+ try {
100
+ for (const n of await getDrives())
101
+ yield { add: { n } }
102
+ }
103
+ catch(error) {
104
+ console.debug(error)
105
}
106
+ return
107
+ }
108
+ try {
109
const dirStream = glob.stream('*', {
110
cwd: path,
111
dot: true,
@@ -119,7 +119,7 @@ const apis: ApiHandlers = {
119
path = path.toString('utf8')
120
try {
121
const stats = await stat(base + path)
122
- send({
122
+ yield {
123
add: {
124
n: path,
125
s: stats.size,
@@ -127,7 +127,7 @@ const apis: ApiHandlers = {
127
m: stats.mtime,
128
k: stats.isDirectory() ? 'd' : undefined,
129
}
130
- })
130
+ }
131
}
132
catch {
133
console.debug('ls: failed stat for ', path)
@@ -136,10 +136,8 @@ const apis: ApiHandlers = {
136
} catch (e) {
137
if ((e as any).code !== 'ENOTDIR')
138
throw e
139
- } finally {
140
- end()
139
}
142
- })
140
+ }
141
142
}
143
src/apis.ts
+14
-33
@@ -1,42 +1,16 @@
1
import { IncomingMessage } from 'http'
2
import Koa from 'koa'
3
-import EventEmitter from 'events'
3
import createSSE from './sse'
5
-import { Callback, wait } from './misc'
4
5
export class ApiError extends Error {
6
constructor(public status:number, message?:string | Error) {
7
super(typeof message === 'string' ? message : message?.message)
8
}
9
}
12
-type ApiHandlerResult = Record<string,any> | ApiError | EventEmitter
10
+type ApiHandlerResult = Record<string,any> | ApiError | AsyncGenerator<any>
11
export type ApiHandler = (params:any, ctx:Koa.Context) => ApiHandlerResult | Promise<ApiHandlerResult>
12
export type ApiHandlers = Record<string, ApiHandler>
13
16
-type ApiEmitter = (args:{ send: DataEmitter, end: Callback, onClose: Callback<Callback>, params:any, ctx: Koa.Context }) => void
17
-type DataEmitter = (data:any) => void
18
-
19
-export function apiEmitter(cb: ApiEmitter) {
20
- return (params:any, ctx: Koa.Context) => {
21
- const em = new EventEmitter()
22
- let ready = wait(0) // wait for the sse to be created
23
- cb({
24
- send(data) {
25
- ready.then(() => em.emit('data', data))
26
- },
27
- end() {
28
- ready.then(() => em.emit('end'))
29
- },
30
- onClose(cb) {
31
- ctx.res.once('close', cb)
32
- },
33
- params,
34
- ctx
35
- })
36
- return em
37
- }
38
-}
39
-
14
export function apiMiddleware(apis: ApiHandlers) : Koa.Middleware {
15
return async (ctx) => {
16
const params = ctx.method === 'POST' ? await getJsonFromReq(ctx.req) : ctx.request.query
@@ -49,14 +23,17 @@ export function apiMiddleware(apis: ApiHandlers) : Koa.Middleware {
23
// we don't rely on SameSite cookie option because it's https-only
24
const res = csrf && csrf !== params.csrf ? new ApiError(401, 'csrf')
25
: await apis[ctx.path](params || {}, ctx)
52
- if (res && res instanceof EventEmitter) {
53
- const sse = createSSE(ctx)
54
- res.on('data', data => sse.send(data))
55
- res.on('end', () => sse.close())
26
+ // if it returns an AsyncIterator we'll go SSE-mode
27
+ if (isAsyncGenerator(res)) {
28
+ const sse = createSSE(ctx) // initiate SSE and return, then we'll continue sending values asynchronously
29
+ setTimeout(async ()=> {
30
+ const iterable = { [Symbol.asyncIterator]: () => res }
31
+ for await (const value of iterable)
32
+ sse.send(value)
33
+ sse.close()
34
+ })
35
return
36
}
58
- if (!res) // this should happen only in case of SSE
59
- return
37
if (res instanceof ApiError) {
38
ctx.body = res.message
39
return ctx.status = res.status
@@ -69,6 +46,10 @@ export function apiMiddleware(apis: ApiHandlers) : Koa.Middleware {
46
}
47
}
48
49
+function isAsyncGenerator(x: any): x is AsyncGenerator {
50
+ return typeof (x as AsyncGenerator)?.next === 'function'
51
+}
52
+
53
async function getJsonFromReq(req: IncomingMessage): Promise<any> {
54
return new Promise((resolve, reject) => {
55
let data = ''
src/frontEndApis.ts
+2
-2
@@ -1,8 +1,8 @@
1
import { ApiHandlers } from './apis'
2
-import * as api_file_list from './api.file_list'
2
+import { file_list } from './api.file_list'
3
import * as api_auth from './api.auth'
4
5
export const frontEndApis: ApiHandlers = {
6
- ...api_file_list,
6
+ file_list,
7
...api_auth,
8
}
src/misc.ts
+28
-3
@@ -58,6 +58,31 @@ export async function* filterMapGenerator<IN,OUT>(generator: AsyncIterableIterat
58
}
59
}
60
61
+export async function asyncGeneratorToArray<T>(generator: AsyncIterable<T>): Promise<T[]> {
62
+ const ret: T[] = []
63
+ for await(const x of generator)
64
+ ret.push(x)
65
+ return ret
66
+}
67
+
68
+// let you use work with a callback when a generator is required
69
+export function generatorAsCallback<T>(caller: Callback<{ callback:Callback<T> }>) {
70
+ let p = pendingPromise()
71
+ const ref = { callback: p.resolve }
72
+ caller(ref)
73
+ return {
74
+ [Symbol.asyncIterator]: () =>
75
+ ({
76
+ async next() {
77
+ const value = await p
78
+ p = pendingPromise()
79
+ ref.callback = p.resolve
80
+ return { value }
81
+ }
82
+ })
83
+ }
84
+}
85
+
86
export function getOrSet<T>(o:any, k:string, creator:()=>T): T {
87
return k in o ? o[k]
88
: (o[k] = creator())
@@ -125,9 +150,9 @@ export function pendingPromise<T>() {
150
return Object.assign(ret, takeOut) as PendingPromise<T>
151
}
152
128
-// returns an 'uninstall' callback for the handlers you just installed. Pass a map {event:handler}
129
-export function onOffMap(em: EventEmitter, events: Record<string, (...args: any[]) => void>) {
130
- events = { ...events } // avoid later modifications
153
+// install multiple handlers and returns a handy 'uninstall' function which requires no parameter. Pass a map {event:handler}
154
+export function onOffMap(em: EventEmitter, events: { [eventName:string]: (...args: any[]) => void }) {
155
+ events = { ...events } // avoid later modifications, as we need this later for uninstallation
156
for (const k in events)
157
em.on(k, events[k])
158
return () => {