better code: support streams for apis, to be used where is more fit
Massimo Melina committed
Mar 17, 2022 at 13:52 UTC
1280cc5154da420db642674e27a509acd893e6af
5 files changed
+52
-60
server/src/adminApis.ts
+19
-15
@@ -7,11 +7,12 @@ import { BUILD_TIMESTAMP, FORBIDDEN, HFS_STARTED, VERSION } from './const'
7
import vfsApis from './api.vfs'
8
import accountsApis from './api.accounts'
9
import { Connection, getConnections } from './connections'
10
-import { generatorAsCallback, onOffMap, pendingPromise } from './misc'
10
+import { onOffMap, pendingPromise } from './misc'
11
import _ from 'lodash'
12
import events from './events'
13
import { getFromAccount } from './perm'
14
import Koa from 'koa'
15
+import { Readable } from 'stream'
16
17
export const adminApis: ApiHandlers = {
18
@@ -62,21 +63,24 @@ export const adminApis: ApiHandlers = {
63
return { result: Boolean(c) }
64
},
65
65
- async *get_connections({}, ctx) {
66
+ get_connections({}, ctx) {
67
+ const ret = new Readable({ objectMode: true, read(){} }) // we don't care what you ask/read, we just push and hope for the best
68
+ // start with existing connections
69
for (const conn of getConnections())
67
- yield { add: serializeConnection(conn) }
68
- yield* generatorAsCallback(wrapper =>
69
- ctx.res.once('close', // as connection is closed, call the callback returned by onOffMap that uninstalls the listener
70
- onOffMap(events, {
71
- connection: conn => wrapper.callback({ add: serializeConnection(conn) }),
72
- connectionClosed(conn: Connection) {
73
- wrapper.callback({ remove: [ serializeConnection(conn, true) ] })
74
- },
75
- connectionUpdated(conn: Connection, change: Partial<Connection>) {
76
- wrapper.callback({ update: [{ search: serializeConnection(conn, true), change }] })
77
- },
78
- })
79
- ) )
70
+ ret.push({ add: serializeConnection(conn) })
71
+ // then send updates
72
+ const off = onOffMap(events, {
73
+ connection: conn => ret.push({ add: serializeConnection(conn) }),
74
+ connectionClosed(conn: Connection) {
75
+ ret.push({ remove: [ serializeConnection(conn, true) ] })
76
+ },
77
+ connectionUpdated(conn: Connection, change: Partial<Connection>) {
78
+ ret.push({ update: [{ search: serializeConnection(conn, true), change }] })
79
+ },
80
+ })
81
+ // we never close this stream ourselves, just when connection is closed we have to take care of listeners
82
+ ctx.res.once('close', off)
83
+ return ret
84
85
function serializeConnection(conn: Connection, minimal?:true) {
86
const { socket, started, secure, got, path } = conn
server/src/api.file_list.ts
+4
-2
@@ -4,7 +4,7 @@ import { cantReadStatusCode, getNodeName, hasPermission, urlToNode, VfsNode, wal
4
import { ApiError, ApiHandler } from './apiMiddleware'
5
import { stat } from 'fs/promises'
6
import { mapPlugins } from './plugins'
7
-import { asyncGeneratorToArray, dirTraversal, filterMapGenerator, pattern2filter } from './misc'
7
+import { asyncGeneratorToArray, asyncGeneratorToReadable, dirTraversal, filterMapGenerator, pattern2filter } from './misc'
8
9
export const file_list:ApiHandler = async ({ path, offset, limit, search, omit, sse }, ctx) => {
10
let node = await urlToNode(path || '/', ctx)
@@ -21,7 +21,9 @@ export const file_list:ApiHandler = async ({ path, offset, limit, search, omit,
21
const filter = pattern2filter(search)
22
const walker = walkNode(node, ctx, search ? Infinity : 0)
23
const onDirEntryHandlers = mapPlugins(plug => plug.onDirEntry)
24
- return sse ? filterMapGenerator(produceEntries(), async (entry) => ({ entry })) // wrap entry in an object
24
+ return sse ? asyncGeneratorToReadable(
25
+ filterMapGenerator(produceEntries(), async (entry) => ({ entry })) // wrap entry in an object
26
+ )
27
: { list: await asyncGeneratorToArray(produceEntries()) }
28
29
async function* produceEntries() {
server/src/apiMiddleware.ts
+9
-14
@@ -3,13 +3,15 @@
3
import { IncomingMessage } from 'http'
4
import Koa from 'koa'
5
import createSSE from './sse'
6
+import { Readable } from 'stream'
7
+import { asyncGeneratorToReadable } from './misc'
8
9
export class ApiError extends Error {
10
constructor(public status:number, message?:string | Error) {
11
super(typeof message === 'string' ? message : message?.message)
12
}
13
}
12
-type ApiHandlerResult = Record<string,any> | ApiError | AsyncGenerator<any>
14
+type ApiHandlerResult = Record<string,any> | ApiError | Readable | AsyncGenerator<any>
15
export type ApiHandler = (params:any, ctx:Koa.Context) => ApiHandlerResult | Promise<ApiHandlerResult>
16
export type ApiHandlers = Record<string, ApiHandler>
17
@@ -23,24 +25,17 @@ export function apiMiddleware(apis: ApiHandlers) : Koa.Middleware {
25
}
26
const csrf = ctx.cookies.get('csrf')
27
// we don't rely on SameSite cookie option because it's https-only
26
- const res = csrf && csrf !== params.csrf ? new ApiError(401, 'csrf')
28
+ let res = csrf && csrf !== params.csrf ? new ApiError(401, 'csrf')
29
: await apis[ctx.path](params || {}, ctx)
28
- // if it returns an AsyncIterator we'll go SSE-mode
29
- if (isAsyncGenerator(res)) {
30
- const sse = createSSE(ctx) // initiate SSE and return, then we'll continue sending values asynchronously
31
- setTimeout(async ()=> {
32
- const iterable = { [Symbol.asyncIterator]: () => res }
33
- for await (const value of iterable)
34
- sse.send(value)
35
- sse.close()
36
- })
37
- return
38
- }
30
+ if (isAsyncGenerator(res))
31
+ res = asyncGeneratorToReadable(res)
32
+ if (res instanceof Readable) // Readable, we'll go SSE-mode
33
+ return res.pipe(createSSE(ctx))
34
if (res instanceof ApiError) {
35
ctx.body = res.message
36
return ctx.status = res.status
37
}
43
- if (res instanceof Error) {
38
+ if (res instanceof Error) { // generic exception
39
ctx.body = String(res)
40
return ctx.status = 400
41
}
server/src/misc.ts
+10
-16
@@ -5,6 +5,7 @@ import fs from 'fs/promises'
5
import { basename, dirname } from 'path'
6
import { watch } from 'fs'
7
import _ from 'lodash'
8
+import { Readable } from 'stream'
9
10
export type Callback<IN=void, OUT=void> = (x:IN) => OUT
11
export type Dict<T = any> = Record<string, T>
@@ -72,22 +73,15 @@ export async function asyncGeneratorToArray<T>(generator: AsyncIterable<T>): Pro
73
return ret
74
}
75
75
-// let you use work with a callback when a generator is required
76
-export function generatorAsCallback<T>(caller: Callback<{ callback:Callback<T> }>) {
77
- let p = pendingPromise()
78
- const ref = { callback: p.resolve }
79
- caller(ref)
80
- return {
81
- [Symbol.asyncIterator]: () =>
82
- ({
83
- async next() {
84
- const value = await p
85
- p = pendingPromise()
86
- ref.callback = p.resolve
87
- return { value }
88
- }
89
- })
90
- }
76
+export function asyncGeneratorToReadable<T>(generator: AsyncIterable<T>) {
77
+ const iterator = generator[Symbol.asyncIterator]()
78
+ return new Readable({
79
+ objectMode: true,
80
+ read() {
81
+ iterator.next().then(it =>
82
+ this.push(it.done ? null : it.value))
83
+ }
84
+ })
85
}
86
87
export function getOrSet<T>(o: Record<string,T>, k:string, creator:()=>T): T {
server/src/sse.ts
+10
-13
@@ -1,7 +1,7 @@
1
// This file is part of HFS - Copyright 2021-2022, Massimo Melina <a@rejetto.com> - License https://www.gnu.org/licenses/gpl-3.0.txt
2
3
import Koa from 'koa'
4
-import { PassThrough } from 'stream'
4
+import { Transform } from 'stream'
5
6
export default function createSSE(ctx: Koa.Context) {
7
const { socket } = ctx.req
@@ -15,18 +15,15 @@ export default function createSSE(ctx: Koa.Context) {
15
'X-Accel-Buffering': 'no', // avoid buffering when reverse-proxied through nginx
16
})
17
ctx.status = 200
18
- const stream = ctx.body = new PassThrough()
19
- const ret = {
20
- stream,
21
- stopped: false,
22
- send(data:any){
23
- stream.write(`data: ${JSON.stringify(data)}\n\n`)
18
+ return ctx.body = new Transform({
19
+ objectMode: true,
20
+ transform(chunk, encoding, cb) {
21
+ this.push(`data: ${JSON.stringify(chunk)}\n\n`)
22
+ cb()
23
},
25
- close() {
26
- stream.end('data:\n\n')
24
+ flush(cb) {
25
+ this.push('data:\n\n')
26
+ cb()
27
}
28
- }
29
- stream.on('close', ()=>
30
- ret.stopped = true)
31
- return ret
28
+ })
29
}