@@ -306,8 +306,23 @@ function onStreamClose(code) { |
306 | 306 |
|
307 | 307 | if (state.fd !== undefined) |
308 | 308 | tryClose(state.fd); |
309 | | - stream.push(null); |
310 | | - stream[kMaybeDestroy](null, code); |
| 309 | + |
| 310 | + // Defer destroy we actually emit end. |
| 311 | + if (stream._readableState.endEmitted || code !== NGHTTP2_NO_ERROR) { |
| 312 | + // If errored or ended, we can destroy immediately. |
| 313 | + stream[kMaybeDestroy](null, code); |
| 314 | + } else { |
| 315 | + // Wait for end to destroy. |
| 316 | + stream.on('end', stream[kMaybeDestroy]); |
| 317 | + // Push a null so the stream can end whenever the client consumes |
| 318 | + // it completely. |
| 319 | + stream.push(null); |
| 320 | + |
| 321 | + // Same as net. |
| 322 | + if (stream._readableState.length === 0) { |
| 323 | + stream.read(0); |
| 324 | + } |
| 325 | + } |
311 | 326 | } |
312 | 327 |
|
313 | 328 | // Receives a chunk of data for a given stream and forwards it on |
@@ -325,11 +340,19 @@ function onStreamRead(nread, buf) { |
325 | 340 | } |
326 | 341 | return; |
327 | 342 | } |
| 343 | + |
328 | 344 | // Last chunk was received. End the readable side. |
329 | 345 | debug(`Http2Stream ${stream[kID]} [Http2Session ` + |
330 | 346 | `${sessionName(stream[kSession][kType])}]: ending readable.`); |
331 | | - stream.push(null); |
332 | | - stream[kMaybeDestroy](); |
| 347 | + |
| 348 | + // defer this until we actually emit end |
| 349 | + if (stream._readableState.endEmitted) { |
| 350 | + stream[kMaybeDestroy](); |
| 351 | + } else { |
| 352 | + stream.on('end', stream[kMaybeDestroy]); |
| 353 | + stream.push(null); |
| 354 | + stream.read(0); |
| 355 | + } |
333 | 356 | } |
334 | 357 |
|
335 | 358 | // Called when the remote peer settings have been updated. |
@@ -1826,21 +1849,25 @@ class Http2Stream extends Duplex { |
1826 | 1849 | session[kMaybeDestroy](); |
1827 | 1850 | process.nextTick(emit, this, 'close', code); |
1828 | 1851 | callback(err); |
1829 | | - } |
1830 | 1852 |
|
| 1853 | + } |
1831 | 1854 | // The Http2Stream can be destroyed if it has closed and if the readable |
1832 | 1855 | // side has received the final chunk. |
1833 | 1856 | [kMaybeDestroy](error, code = NGHTTP2_NO_ERROR) { |
1834 | | - if (error == null) { |
1835 | | - if (code === NGHTTP2_NO_ERROR && |
1836 | | - (!this._readableState.ended || |
1837 | | - !this._writableState.ended || |
1838 | | - this._writableState.pendingcb > 0 || |
1839 | | - !this.closed)) { |
1840 | | - return; |
1841 | | - } |
| 1857 | + if (error || code !== NGHTTP2_NO_ERROR) { |
| 1858 | + this.destroy(error); |
| 1859 | + return; |
| 1860 | + } |
| 1861 | + |
| 1862 | + // TODO(mcollina): remove usage of _*State properties |
| 1863 | + if (this._readableState.ended && |
| 1864 | + this._writableState.ended && |
| 1865 | + this._writableState.pendingcb === 0 && |
| 1866 | + this.closed) { |
| 1867 | + this.destroy(); |
| 1868 | + // This should return, but eslint complains. |
| 1869 | + // return |
1842 | 1870 | } |
1843 | | - this.destroy(error); |
1844 | 1871 | } |
1845 | 1872 | } |
1846 | 1873 |
|
|