support resume on zip files https://github.com/rejetto/hfs/issues/63
Massimo Melina committed
Jun 25, 2022 at 16:57 UTC
c605eedbb8f9163be43e7f92b2687d7bfe9adfd9
4 files changed
+122
-74
server/src/QuickZipStream.ts
+82
-44
@@ -3,16 +3,19 @@
3
import { Readable } from 'stream'
4
// @ts-ignore
5
import { crc32 as crc32lib } from 'buffer-crc32'
6
+import assert from 'assert'
7
8
const ZIP64_LIMIT = 2**31 -1
9
9
-const crc32provider = import('@node-rs/crc32').then(lib => lib.crc32, () => {
10
+let crc32function: (input: string | Buffer, initialState?: number | undefined | null) => number
11
+import('@node-rs/crc32').then(lib => crc32function = lib.crc32, () => {
12
console.log('using generic lib for crc32')
11
- return crc32lib.unsigned
13
+ return crc32function = crc32lib.unsigned
14
})
15
16
interface ZipSource {
17
path: string
18
+ sourcePath?: string
19
getData: () => Readable // deferred stream, so that we don't keep many open files because of calculateSize()
20
size: number
21
ts: Date
@@ -24,32 +27,57 @@ export class QuickZipStream extends Readable {
27
private finished = false
28
private readonly centralDir: ({ size:number, crc:number, ts:Date, pathAsBuffer:Buffer, offset:number, version:number, extAttr: number })[] = []
29
private dataWritten = 0
27
- private prewalk?: ZipSource[]
30
+ private consumedCalculating: ZipSource[] = []
31
+ private skip: number = 0
32
+ private limit?: number
33
34
constructor(private readonly walker: AsyncIterableIterator<ZipSource>) {
35
super({})
36
}
37
33
- _push(chunk:any) {
38
+ earlyClose() {
39
+ this.finished = true
40
+ this.push(null)
41
+ }
42
+
43
+ applyRange(start: number, end: number) {
44
+ if (end < start)
45
+ return this.earlyClose()
46
+ this.skip = start
47
+ this.limit = end - start + 1
48
+ }
49
+
50
+ _push(chunk: number[] | Buffer) {
51
if (Array.isArray(chunk))
52
chunk = buffer(chunk)
36
- this.push(chunk)
53
this.dataWritten += chunk.length
54
+ if (this.skip) {
55
+ if (this.skip >= chunk.length)
56
+ return this.skip -= chunk.length
57
+ chunk = chunk.subarray(this.skip)
58
+ this.skip = 0
59
+ }
60
+ const lastBit = this.limit! < chunk.length
61
+ if (lastBit)
62
+ chunk = chunk.subarray(0, this.limit)
63
+
64
+ this.push(chunk)
65
+ if (lastBit)
66
+ this.earlyClose()
67
}
68
69
async calculateSize(howLong:number = 1000) {
41
- this.prewalk = []
70
const endBy = Date.now() + howLong
71
while (1) {
72
if (Date.now() >= endBy)
73
return NaN
74
const { value } = await this.walker.next()
75
if (!value) break
48
- this.prewalk.push(value) // we keep same shape of the generator, so
76
+ this.consumedCalculating.push(value) // we keep same shape of the generator, so
77
}
78
let offset = 0
79
let centralDirSize = 0
52
- for (const file of this.prewalk) {
80
+ for (const file of this.consumedCalculating) {
81
const pathSize = Buffer.from(file.path, 'utf8').length
82
const extraLength = (file.size > ZIP64_LIMIT ? 2 : 0) + (offset > ZIP64_LIMIT ? 1 : 0)
83
const extraDataSize = extraLength && (2+2 + extraLength*8)
@@ -63,17 +91,14 @@ export class QuickZipStream extends Readable {
91
return offset + centralDirSize
92
}
93
66
- async _read(): Promise<void> {
94
+ async _read() {
95
if (this.workingFile || this.finished || this.destroyed) return
68
- const file = this.prewalk?.shift() || (await this.walker.next()).value as ZipSource
96
+ const file = this.consumedCalculating.shift() || (await this.walker.next()).value as ZipSource
97
if (!file)
98
return this.closeArchive()
99
++this.numberOfFiles
72
- let { path, getData, size, ts, mode } = file
73
- const data = getData()
100
+ let { path, sourcePath, getData, size, ts, mode } = file
101
const pathAsBuffer = Buffer.from(path, 'utf8')
75
- const crc32 = await crc32provider
76
- let crc: number | undefined = undefined
102
const offset = this.dataWritten
103
let version = 20
104
this._push([
@@ -89,22 +114,39 @@ export class QuickZipStream extends Readable {
114
2, 0, // extra length
115
])
116
this._push(pathAsBuffer)
117
+ if (this.finished) return
118
93
- let total = 0
119
+ const cache = sourcePath ? crcCache[sourcePath] : undefined
120
+ const cacheHit = Number(cache?.ts) === Number(ts)
121
+ let crc = cacheHit ? cache!.crc : crc32function('')
122
+ const extAttr = !mode ? 0 : (mode | 0x8000) * 0x10000 // it's like <<16 but doesn't overflow so easily
123
+ const centralDirEntry = { size, crc, pathAsBuffer, ts, offset, version, extAttr }
124
+ if (this.skip >= size && cacheHit) {
125
+ this.skip -= size
126
+ this.dataWritten += size
127
+ this.centralDir.push(centralDirEntry)
128
+ setTimeout(() => this.push('')) // this "signal" works only after _read() is done
129
+ return
130
+ }
131
+ const data = getData()
132
data.on('error', (err) => console.error(err))
133
+ data.on('end', ()=>{
134
+ this.workingFile = false
135
+ centralDirEntry.crc = crc
136
+ if (sourcePath)
137
+ crcCache[sourcePath] = { ts, crc }
138
+ this.centralDir.push(centralDirEntry)
139
+ this.push('') // continue piping
140
+ })
141
+ this.workingFile = true
142
data.on('data', chunk => {
143
if (this.destroyed)
144
return data.destroy()
145
this._push(chunk)
99
- crc = crc32(chunk, crc)
100
- total += chunk.length
101
- })
102
- this.workingFile = true
103
- data.on('end', ()=>{
104
- this.workingFile = false
105
- const extAttr = !mode ? 0 : (mode | 0x8000) * 0x10000 // it's like <<16 but doesn't overflow so easily
106
- this.centralDir.push({ size, crc:crc!, pathAsBuffer, ts, offset, version, extAttr })
107
- this.push('') // continue piping
146
+ if (!cacheHit)
147
+ crc = crc32function(chunk, crc)
148
+ if (this.finished)
149
+ return data.destroy()
150
})
151
}
152
@@ -183,35 +225,28 @@ export class QuickZipStream extends Readable {
225
}
226
}
227
186
-function buffer(parts: any[]) {
187
- const pairs = []
228
+function buffer(pairs: number[]) {
229
+ assert(pairs.length % 2 === 0)
230
let total = 0
189
- while (parts.length) {
190
- const size = parts.shift()
191
- if (typeof size === 'string') {
192
- pairs.push([String, size])
193
- total += size.length
194
- }
195
- else {
196
- pairs.push([size, parts.shift()])
197
- total += size
198
- }
199
- }
231
+ for (let i=0; i < pairs.length; i+=2)
232
+ total += pairs[i]
233
const ret = Buffer.alloc(total, 0)
234
let offset = 0
202
- for (const [size, data] of pairs) {
235
+ let i = 0
236
+ while (i < pairs.length) {
237
+ const size = pairs[i++]
238
+ const data = pairs[i++]
239
if (size === 1)
204
- offset = ret.writeUInt8(data, offset)
240
+ ret.writeUInt8(data, offset)
241
else if (size === 2)
206
- offset = ret.writeUInt16LE(data, offset)
242
+ ret.writeUInt16LE(data, offset)
243
else if (size === 4)
208
- offset = ret.writeUInt32LE(data, offset)
244
+ ret.writeUInt32LE(data, offset)
245
else if (size === 8)
210
- offset = ret.writeBigUInt64LE(BigInt(data), offset)
211
- else if (size === String)
212
- offset = ret.write(data,'ascii')
246
+ ret.writeBigUInt64LE(BigInt(data), offset)
247
else
248
throw 'unsupported'
249
+ offset += size
250
}
251
return ret
252
}
@@ -224,3 +259,6 @@ function ts2buf(ts:Date) {
259
2, date,
260
]
261
}
262
+
263
+interface CrcCacheEntry { ts: Date, crc: number }
264
+const crcCache: Record<string, CrcCacheEntry> = {}
server/src/serveFile.ts
+30
-26
@@ -45,8 +45,6 @@ export function serveFile(source:string, mime?:string, modifier?:(s:string)=>str
45
return async (ctx) => {
46
if (!source)
47
return
48
- const { range } = ctx.request.header
49
- ctx.set('Accept-Ranges', 'bytes')
48
const fn = path.basename(source)
49
mime = mime ?? _.find(mimeCfg.get(), (v,k) => k>'' && isMatch(fn, k)) // isMatch throws on an empty string
50
if (mime === MIME_AUTO)
@@ -73,33 +71,39 @@ export function serveFile(source:string, mime?:string, modifier?:(s:string)=>str
71
updateConnection(conn, { ctx }) // fileSource is affecting connection's outputted data, so we request an update
72
if (modifier)
73
return ctx.body = modifier(String(await fs.readFile(source)))
76
- if (!range) {
77
- ctx.body = createReadStream(source)
78
- ctx.response.length = stats.size
79
- return
80
- }
81
- const ranges = range.split('=')[1]
82
- if (ranges.includes(','))
83
- return ctx.throw(400, 'multi-range not supported')
84
- let bytes = ranges?.split('-')
85
- if (!bytes?.length)
86
- return ctx.throw(400, 'bad range')
87
- const max = stats.size - 1
88
- let start = Number(bytes[0]) || 0
89
- let end = Number(bytes[1]) || max
90
- if (end > max || start > max) {
91
- ctx.status = 416
92
- ctx.set('Content-Range', `bytes ${stats.size}`)
93
- ctx.body = 'Requested Range Not Satisfiable'
94
- return
95
- }
96
- ctx.status = 206
97
- ctx.set('Content-Range', `bytes ${start}-${end}/${stats.size}`)
98
- ctx.body = createReadStream(source, { start, end })
99
- ctx.response.length = end - start + 1
74
+ const range = getRange(ctx, stats.size)
75
+ ctx.body = createReadStream(source, range)
76
}
77
catch {
78
return ctx.status = 404
79
}
80
}
81
}
82
+
83
+export function getRange(ctx: Koa.Context, totalSize: number) {
84
+ ctx.set('Accept-Ranges', 'bytes')
85
+ const { range } = ctx.request.header
86
+ if (!range) {
87
+ ctx.response.length = totalSize
88
+ return
89
+ }
90
+ const ranges = range.split('=')[1]
91
+ if (ranges.includes(','))
92
+ return ctx.throw(400, 'multi-range not supported')
93
+ let bytes = ranges?.split('-')
94
+ if (!bytes?.length)
95
+ return ctx.throw(400, 'bad range')
96
+ const max = totalSize - 1
97
+ const start = bytes[0] ? Number(bytes[0]) : Math.max(0, totalSize-Number(bytes[1])) // a negative start is relative to the end
98
+ const end = bytes[0] ? Number(bytes[1] || max) : max // NaN in case we are asked for last N bytes without knowing max
99
+ if (isNaN(end) || end > max || start > max) {
100
+ ctx.status = 416
101
+ ctx.set('Content-Range', `bytes ${totalSize}`)
102
+ ctx.body = 'Requested Range Not Satisfiable'
103
+ return
104
+ }
105
+ ctx.status = 206
106
+ ctx.set('Content-Range', `bytes ${start}-${end}/${isNaN(totalSize) ? '*' : totalSize}`)
107
+ ctx.response.length = end - start + 1
108
+ return { start, end }
109
+}
server/src/zip.ts
+6
-1
@@ -2,13 +2,14 @@
2
3
import { getNodeName, hasPermission, nodeIsDirectory, urlToNode, VfsNode, walkNode } from './vfs'
4
import Koa from 'koa'
5
-import { filterMapGenerator, pattern2filter, prefix } from './misc'
5
+import { filterMapGenerator, pattern2filter } from './misc'
6
import { QuickZipStream } from './QuickZipStream'
7
import { createReadStream } from 'fs'
8
import fs from 'fs/promises'
9
import { defineConfig } from './config'
10
import { dirname } from 'path'
11
import { updateConnection } from './connections'
12
+import { getRange } from './serveFile'
13
14
export async function zipStreamFromFolder(node: VfsNode, ctx: Koa.Context) {
15
ctx.status = 200
@@ -46,6 +47,7 @@ export async function zipStreamFromFolder(node: VfsNode, ctx: Koa.Context) {
47
size: st.size,
48
ts: st.mtime || st.ctime,
49
mode: st.mode,
50
+ sourcePath: source,
51
getData: () => createReadStream(source)
52
}
53
}
@@ -54,6 +56,9 @@ export async function zipStreamFromFolder(node: VfsNode, ctx: Koa.Context) {
56
const zip = new QuickZipStream(mappedWalker)
57
const time = 1000 * zipSeconds.get()
58
ctx.response.length = await zip.calculateSize(time)
59
+ const range = getRange(ctx, ctx.response.length)
60
+ if (range)
61
+ zip.applyRange(range.start, range.end)
62
ctx.body = zip
63
ctx.req.on('close', ()=> zip.destroy())
64
ctx.state.archive = 'zip'
tests/test.ts
+4
-3
@@ -26,7 +26,7 @@ describe('basics', () => {
26
it('search', reqList('f1', { inList:['f2/'], outList:['page'] }, { search:'2' }))
27
it('search root', reqList('/', { inList:['cantReadPage/'], outList:['cantReadPage/page/'] }, { search:'page' }))
28
it('download', req('/f1/f2/alfa.txt', { re:/abcd/, mime:'text/plain' }))
29
- it('partial download', req('/f1/f2/alfa.txt', /a[^d]+$/, { // only "abc" is expected
29
+ it('download.partial', req('/f1/f2/alfa.txt', /a[^d]+$/, { // only "abc" is expected
30
headers: { Range: 'bytes=0-2' }
31
}))
32
it('bad range', req('/f1/f2/alfa.txt', 416, {
@@ -66,6 +66,7 @@ describe('basics', () => {
66
it('protectFromAbove.list', reqList('/protectFromAbove/child/', { outList:['alfa.txt'] }))
67
68
it('zip.head', req('/f1/?get=zip', { empty:true, length:13010 }, { method:'HEAD' }) )
69
+ it('zip.partial', req('/f1/f2/?get=zip', { re:/^6/, length:10 }, { headers: { Range: 'bytes=-10' } }) )
70
it('zip.alfa is forbidden', req('/protectFromAbove/child/?get=zip&list=alfa.txt*renamed', { empty: true, length:118 }, { method:'HEAD' }))
71
it('login', reqApi('login', { username, password }, 406)) // by default, we don't support clear-text login
72
@@ -105,12 +106,12 @@ function req(methodUrl: string, test:Tester, requestOptions?:any) {
106
const method = methodUrl.slice(0,i) || requestOptions?.data && 'POST' || 'GET'
107
const url = BASE_URL+methodUrl.slice(i)
108
client.request({ method, url, ...requestOptions })
108
- .then(fun, fun)
109
+ .then(process, process)
110
.catch(err => {
111
done(err)
112
})
113
113
- function fun(res:any) {
114
+ function process(res:any) {
115
//console.debug('sent', requestOptions, 'got', res instanceof Error ? String(res) : [res.status])
116
if (test && test instanceof RegExp)
117
test = { re:test }