Merge pull request #587 from AArnott/fix578

Fix hang in nested multiplexing stream

Author: AArnott

src/nerdbank-streams/src/MultiplexingStream.ts

@@ -37,8 +37,6 @@ export abstract class MultiplexingStream implements IDisposableObservable {
 	/** The default window size used for new channels that do not specify a value for ChannelOptions.ChannelReceivingWindowSize. */
 	readonly defaultChannelReceivingWindowSize: number
 
-	protected readonly formatter: MultiplexingStreamFormatter
-
 	protected get disposalToken() {
 		return this.disposalTokenSource.token
 	}
@@ -99,7 +97,7 @@ export abstract class MultiplexingStream implements IDisposableObservable {
 		const handshakeResult = await formatter.readHandshakeAsync(writeHandshakeData, cancellationToken)
 		formatter.isOdd = handshakeResult.isOdd
 
-		return new MultiplexingStreamClass(stream, handshakeResult.isOdd, options)
+		return new MultiplexingStreamClass(formatter, handshakeResult.isOdd, options)
 	}
 
 	/**
@@ -120,7 +118,7 @@ export abstract class MultiplexingStream implements IDisposableObservable {
 			throw new Error(`Protocol major version ${options.protocolMajorVersion} is not supported. Try CreateAsync instead.`)
 		}
 
-		return new MultiplexingStreamClass(stream, undefined, options)
+		return new MultiplexingStreamClass(formatter, undefined, options)
 	}
 
 	/**
@@ -176,22 +174,13 @@ export abstract class MultiplexingStream implements IDisposableObservable {
 
 	private disposalTokenSource = CancellationToken.create()
 
-	protected constructor(stream: NodeJS.ReadWriteStream, private readonly isOdd: boolean | undefined, options: MultiplexingStreamOptions) {
+	protected constructor(
+		protected readonly formatter: MultiplexingStreamFormatter,
+		private readonly isOdd: boolean | undefined,
+		options: MultiplexingStreamOptions
+	) {
 		this.defaultChannelReceivingWindowSize = options.defaultChannelReceivingWindowSize ?? MultiplexingStream.recommendedDefaultChannelReceivingWindowSize
 		this.protocolMajorVersion = options.protocolMajorVersion ?? 1
-		const formatter: MultiplexingStreamFormatter | undefined =
-			options.protocolMajorVersion === 1
-				? new MultiplexingStreamV1Formatter(stream)
-				: options.protocolMajorVersion === 2
-				? new MultiplexingStreamV2Formatter(stream)
-				: options.protocolMajorVersion === 3
-				? new MultiplexingStreamV3Formatter(stream)
-				: undefined
-		if (formatter === undefined) {
-			throw new Error(`Unsupported major protocol version: ${options.protocolMajorVersion}`)
-		}
-		formatter.isOdd = isOdd
-		this.formatter = formatter
 
 		if (options.seededChannels) {
 			for (let i = 0; i < options.seededChannels.length; i++) {
@@ -534,8 +523,8 @@ export class MultiplexingStreamClass extends MultiplexingStream {
 	protected lastOfferedChannelId: number
 	private readonly sendingSemaphore = new Semaphore(1)
 
-	constructor(stream: NodeJS.ReadWriteStream, isOdd: boolean | undefined, options: MultiplexingStreamOptions) {
-		super(stream, isOdd, options)
+	constructor(formatter: MultiplexingStreamFormatter, isOdd: boolean | undefined, options: MultiplexingStreamOptions) {
+		super(formatter, isOdd, options)
 
 		this.lastOfferedChannelId = isOdd ? -1 : 0 // the first channel created should be 1 or 2
 		this.lastOfferedChannelId += options.seededChannels?.length ?? 0

src/nerdbank-streams/src/tests/MultiplexingStream.spec.ts

@@ -115,7 +115,7 @@ import { nextTick } from 'process'
 
 			// Give time for the termination fram to arrive *before* we try to accept the channel.
 			for (let i = 0; i < 100; i++) {
-				await new Promise<void>(resolve => nextTick(() => resolve()))
+				await nextTickAsync()
 			}
 
 			// We expect this to timeout. But we need this for the test to fail if we have unobserved promise rejections.
@@ -323,6 +323,41 @@ import { nextTick } from 'process'
 				await Promise.all([mx1Local.offerChannelAsync(''), mx2Local.acceptChannelAsync('')])
 			})
 		}
+
+		it('nested stream does not pause', async () => {
+			const rpcChannels = await Promise.all([mx1.offerChannelAsync('test'), mx2.acceptChannelAsync('test')])
+
+			const inner1 = MultiplexingStream.Create(rpcChannels[0].stream, { protocolMajorVersion: 3 })
+			const inner2 = MultiplexingStream.Create(rpcChannels[1].stream, { protocolMajorVersion: 3 })
+			const innerRpcChannels = await Promise.all([inner1.offerChannelAsync('test'), inner2.acceptChannelAsync('test')])
+
+			const iterations = 32 // a high number to exceed high water mark levels in object streams
+
+			const fulfilled = new Promise<Buffer>(resolve => {
+				const chunks: Buffer[] = []
+				innerRpcChannels[1].stream.on('data', chunk => {
+					chunks.push(chunk)
+					if (chunks.length === iterations) {
+						resolve(Buffer.concat(chunks))
+					}
+				})
+			})
+
+			for (let i = 0; i < iterations; i++) {
+				innerRpcChannels[0].stream.write(Buffer.from([i]))
+			}
+
+			await fulfilled
+
+			innerRpcChannels[0].stream.end()
+			innerRpcChannels[1].stream.end()
+			inner1.dispose()
+			inner2.dispose()
+			await inner1.completion
+			await inner2.completion
+			rpcChannels[0].stream.end()
+			rpcChannels[1].stream.end()
+		})
 	})
 
 	async function expectThrow<T>(promise: Promise<T>): Promise<any> {
@@ -334,3 +369,7 @@ import { nextTick } from 'process'
 		}
 	}
 })
+
+function nextTickAsync() {
+	return new Promise<void>(resolve => nextTick(() => resolve()))
+}