moved SSE concerns to apiMiddleware
Massimo Melina committed
Feb 4, 2022 at 10:27 UTC
d0cb8167454cd0d6edddb471f43f8b6bafc963ac
3 files changed
+80
-56
src/api.file_list.ts
+7
-7
@@ -1,10 +1,10 @@
1
import { getNodeName, vfs, VfsNode, walkNode } from './vfs'
2
-import createSSE from './sse'
2
import { ApiError, ApiHandler } from './apis'
3
import { stat } from 'fs/promises'
4
import { mapPlugins } from './plugins'
5
import { pattern2filter } from './misc'
6
import { FORBIDDEN } from './const'
7
+import EventEmitter from 'events'
8
9
export const file_list:ApiHandler = async ({ path, offset, limit, search, omit, sse }, ctx) => {
10
let node = await vfs.urlToNode(path || '/', ctx)
@@ -20,15 +20,15 @@ export const file_list:ApiHandler = async ({ path, offset, limit, search, omit,
20
limit = Number(limit)
21
const filter = pattern2filter(search)
22
const walker = walkNode(node, ctx, search ? Infinity : 0)
23
- const sseSrv = sse ? createSSE(ctx) : null
23
const onDirEntryHandlers = mapPlugins(plug => plug.onDirEntry)
24
+ const emitter = sse && new EventEmitter()
25
const res = produceEntries()
26
- return !sseSrv && { list: await res }
26
+ return emitter || { list: await res }
27
28
async function produceEntries() {
29
const list = []
30
for await (const sub of walker) {
31
- if (sseSrv?.stopped || ctx.aborted) break
31
+ if (ctx.aborted) break
32
if (!filter(getNodeName(sub)))
33
continue
34
const entry = await nodeToDirEntry(sub)
@@ -53,14 +53,14 @@ export const file_list:ApiHandler = async ({ path, offset, limit, search, omit,
53
entry.m = entry.c
54
delete entry.c
55
}
56
- if (sseSrv)
57
- sseSrv.send({ entry })
56
+ if (emitter)
57
+ emitter.emit('data', { entry })
58
else
59
list.push(entry)
60
if (limit && !--limit)
61
break
62
}
63
- sseSrv?.close()
63
+ emitter.emit('end')
64
return list
65
}
66
}
src/api.vfs.ts
+43
-48
@@ -1,10 +1,9 @@
1
import { getNodeName, nodeIsDirectory, vfs, VfsNode, VfsNodeType } from './vfs'
2
import _ from 'lodash'
3
import { stat } from 'fs/promises'
4
-import { ApiError, ApiHandlers } from './apis'
4
+import { apiEmitter, ApiError, ApiHandlers } from './apis'
5
import { dirname } from 'path'
6
import { saveConfigAsap } from './config'
7
-import createSSE from './sse'
7
import glob from 'fast-glob'
8
import { enforceFinal, isWindows } from './misc'
9
import { exec } from 'child_process'
@@ -95,56 +94,52 @@ const apis: ApiHandlers = {
94
return { path: process.cwd() }
95
},
96
98
- async ls({ path }, ctx) {
99
- (async () => {
100
- const sseSrv = createSSE(ctx)
101
- try {
102
- if (!path && isWindows()) {
103
- try {
104
- const { stdout } = await promisify(exec)('wmic logicaldisk get name')
105
- for (const drive of stdout.split('\n').slice(1).map(x => x.trim()).filter(Boolean))
106
- sseSrv.send({ entry: { n: drive } })
107
- }
108
- catch(error) {
109
- console.debug(error)
110
- }
111
- return
97
+ ls: apiEmitter(async ({ send, end, ctx, params:{ path } }) => {
98
+ try {
99
+ if (!path && isWindows()) {
100
+ try {
101
+ const { stdout } = await promisify(exec)('wmic logicaldisk get name')
102
+ for (const drive of stdout.split('\n').slice(1).map(x => x.trim()).filter(Boolean))
103
+ send({ entry: { n: drive } })
104
}
113
- const dirStream = glob.stream('*', {
114
- cwd: path,
115
- dot: true,
116
- onlyFiles: false,
117
- })
118
- const base = enforceFinal('/', path)
119
- for await (let path of dirStream) {
120
- if (ctx.req.aborted)
121
- return
122
- if (path instanceof Buffer)
123
- path = path.toString('utf8')
124
- try {
125
- const stats = await stat(base + path)
126
- const entry = {
127
- n: path,
128
- s: stats.size,
129
- c: stats.ctime,
130
- m: stats.mtime,
131
- k: stats.isDirectory() ? 'd' : undefined,
132
- }
133
- sseSrv.send({ entry })
134
- }
135
- catch {
136
- console.debug('ls: failed stat for ', path)
105
+ catch(error) {
106
+ console.debug(error)
107
+ }
108
+ return
109
+ }
110
+ const dirStream = glob.stream('*', {
111
+ cwd: path,
112
+ dot: true,
113
+ onlyFiles: false,
114
+ })
115
+ const base = enforceFinal('/', path)
116
+ for await (let path of dirStream) {
117
+ if (ctx.req.aborted)
118
+ return
119
+ if (path instanceof Buffer)
120
+ path = path.toString('utf8')
121
+ try {
122
+ const stats = await stat(base + path)
123
+ const entry = {
124
+ n: path,
125
+ s: stats.size,
126
+ c: stats.ctime,
127
+ m: stats.mtime,
128
+ k: stats.isDirectory() ? 'd' : undefined,
129
}
130
+ send({ entry })
131
+ }
132
+ catch {
133
+ console.debug('ls: failed stat for ', path)
134
}
139
- } catch (e) {
140
- if ((e as any).code !== 'ENOTDIR')
141
- throw e
142
- } finally {
143
- sseSrv.close()
135
}
145
- })().then()
146
- return false
147
- }
136
+ } catch (e) {
137
+ if ((e as any).code !== 'ENOTDIR')
138
+ throw e
139
+ } finally {
140
+ end()
141
+ }
142
+ })
143
144
}
145
src/apis.ts
+30
-1
@@ -1,15 +1,38 @@
1
import { IncomingMessage } from 'http'
2
import Koa from 'koa'
3
+import EventEmitter from 'events'
4
+import createSSE from './sse'
5
6
export class ApiError extends Error {
7
constructor(public status:number, message?:string | Error) {
8
super(typeof message === 'string' ? message : message?.message)
9
}
10
}
9
-type ApiHandlerResult = Record<string,any> | ApiError
11
+type ApiHandlerResult = Record<string,any> | ApiError | EventEmitter
12
export type ApiHandler = (params:any, ctx:Koa.Context) => ApiHandlerResult | Promise<ApiHandlerResult>
13
export type ApiHandlers = Record<string, ApiHandler>
14
15
+type ApiEmitter = (args:{ send: DataEmitter, end: EndEmitter, params:any, ctx: Koa.Context }) => void
16
+type DataEmitter = (data:any) => void
17
+type EndEmitter = () => void
18
+
19
+export function apiEmitter(cb: ApiEmitter) {
20
+ return (params:any, ctx: Koa.Context) => {
21
+ const em = new EventEmitter()
22
+ cb({
23
+ send(data) {
24
+ em.emit('data', data)
25
+ },
26
+ end() {
27
+ em.emit('end')
28
+ },
29
+ params,
30
+ ctx
31
+ })
32
+ return em
33
+ }
34
+}
35
+
36
export function apiMiddleware(apis: ApiHandlers) : Koa.Middleware {
37
return async (ctx) => {
38
const params = ctx.method === 'POST' ? await getJsonFromReq(ctx.req) : ctx.request.query
@@ -29,6 +52,12 @@ export function apiMiddleware(apis: ApiHandlers) : Koa.Middleware {
52
catch(e) {
53
ctx.throw(500, String(e))
54
}
55
+ if (res && res instanceof EventEmitter) {
56
+ const sse = createSSE(ctx)
57
+ res.on('data', data => sse.send(data))
58
+ res.on('end', () => sse.close())
59
+ return
60
+ }
61
if (!res) // this should happen only in case of SSE
62
return
63
if (res instanceof ApiError) {