Skip to content

Commit 1be85c2

Browse files
committed
fix: Client.stream writableNeedDrain
Fixes: #441 Refs: nodejs/node#35348 Refs: nodejs/node#35341
1 parent 37ecb75 commit 1be85c2

File tree

4 files changed

+94
-2
lines changed

4 files changed

+94
-2
lines changed

lib/client-stream.js

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,12 @@ class StreamHandler extends AsyncResource {
109109
})
110110

111111
this.res = res
112+
113+
const needDrain = res.writableNeedDrain !== undefined
114+
? res.writableNeedDrain
115+
: res._writableState && res._writableState.needDrain
116+
117+
return needDrain !== true
112118
}
113119

114120
onData (chunk) {

lib/core/client.js

Lines changed: 43 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -374,19 +374,37 @@ class Parser extends HTTPParser {
374374
this.headers = null
375375
this.shouldKeepAlive = false
376376
this.request = null
377+
this.paused = false
378+
this.queue = []
377379

378380
this._resume = () => {
379-
// TODO: Resume parser.
381+
this.paused = false
382+
383+
while (this.queue.length) {
384+
const [fn, ...args] = this.queue.shift()
385+
386+
fn.apply(this, args)
387+
388+
if (this.paused) {
389+
return
390+
}
391+
}
392+
380393
socketResume(socket)
381394
}
382395

383396
this._pause = () => {
384-
// TODO: Pause parser.
397+
this.paused = true
385398
socketPause(socket)
386399
}
387400
}
388401

389402
[HTTPParser.kOnHeaders] (rawHeaders) {
403+
if (this.paused) {
404+
this.queue.push([this[HTTPParser.kOnHeaders], rawHeaders])
405+
return
406+
}
407+
390408
if (this.headers) {
391409
Array.prototype.push.apply(this.headers, rawHeaders)
392410
} else {
@@ -395,6 +413,11 @@ class Parser extends HTTPParser {
395413
}
396414

397415
[HTTPParser.kOnExecute] (ret) {
416+
if (this.paused) {
417+
this.queue.push([this[HTTPParser.kOnExecute], ret])
418+
return
419+
}
420+
398421
const { upgrade, socket } = this
399422

400423
if (!Number.isFinite(ret)) {
@@ -465,6 +488,12 @@ class Parser extends HTTPParser {
465488

466489
[HTTPParser.kOnHeadersComplete] (versionMajor, versionMinor, rawHeaders, method,
467490
url, statusCode, statusMessage, upgrade, shouldKeepAlive) {
491+
if (this.paused) {
492+
this.queue.push([this[HTTPParser.kOnHeadersComplete], versionMajor, versionMinor, rawHeaders, method,
493+
url, statusCode, statusMessage, upgrade, shouldKeepAlive])
494+
return
495+
}
496+
468497
const { client, socket } = this
469498

470499
const request = client[kQueue][client[kRunningIdx]]
@@ -560,6 +589,11 @@ class Parser extends HTTPParser {
560589
}
561590

562591
[HTTPParser.kOnBody] (chunk, offset, length) {
592+
if (this.paused) {
593+
this.queue.push([this[HTTPParser.kOnBody], chunk, offset, length])
594+
return
595+
}
596+
563597
const { socket, statusCode, request } = this
564598

565599
if (socket.destroyed) {
@@ -578,6 +612,11 @@ class Parser extends HTTPParser {
578612
}
579613

580614
[HTTPParser.kOnMessageComplete] () {
615+
if (this.paused) {
616+
this.queue.push([this[HTTPParser.kOnMessageComplete]])
617+
return
618+
}
619+
581620
const { client, socket, statusCode, headers, upgrade, request, trailers } = this
582621

583622
if (socket.destroyed) {
@@ -785,6 +824,7 @@ function connect (client) {
785824
}
786825

787826
function socketPause (socket) {
827+
// TODO: Pause parser.
788828
if (socket._handle && socket._handle.reading) {
789829
socket._handle.reading = false
790830
const err = socket._handle.readStop()
@@ -795,6 +835,7 @@ function socketPause (socket) {
795835
}
796836

797837
function socketResume (socket) {
838+
// TODO: Resume parser.
798839
if (socket._handle && !socket._handle.reading) {
799840
socket._handle.reading = true
800841
const err = socket._handle.readStart()

lib/core/request.js

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -162,6 +162,7 @@ class Request {
162162

163163
onBody (chunk, offset, length) {
164164
assert(!this.aborted)
165+
assert(!this[kPaused])
165166

166167
if (this[kTimeout] && this[kTimeout].refresh) {
167168
this[kTimeout].refresh()

test/client-stream.js

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -657,3 +657,47 @@ test('stream body destroyed on invalid callback', (t) => {
657657
}
658658
})
659659
})
660+
661+
test('stream needDrain', (t) => {
662+
t.plan(1)
663+
664+
const server = createServer((req, res) => {
665+
res.end(Buffer.alloc(4096))
666+
})
667+
t.tearDown(server.close.bind(server))
668+
669+
server.listen(0, async () => {
670+
const client = new Client(`http://localhost:${server.address().port}`)
671+
t.tearDown(() => {
672+
console.error(3)
673+
client.destroy()
674+
})
675+
676+
const dst = new PassThrough()
677+
dst.pause()
678+
679+
while (dst.write(Buffer.alloc(4096))) {
680+
681+
}
682+
683+
const orgWrite = dst.write
684+
dst.write = () => t.fail()
685+
const p = client.stream({
686+
path: '/',
687+
method: 'GET'
688+
}, () => {
689+
return dst
690+
})
691+
692+
setTimeout(() => {
693+
dst.write = (...args) => {
694+
orgWrite.call(dst, ...args)
695+
}
696+
dst.resume()
697+
}, 1e3)
698+
699+
await p
700+
701+
t.pass()
702+
})
703+
})

0 commit comments

Comments
 (0)