Fix default buffering strategy for stream deserialization to prevent discarding unpacking data. #321

yfakariya · yfakariya · commit a983c2399afc · 2019-07-20T01:12:22.000+09:00
For detailed information, see #321 comment.
diff --git a/src/MsgPack/PackerUnpackerStreamOptions.cs b/src/MsgPack/PackerUnpackerStreamOptions.cs
@@ -82,9 +82,19 @@ private static bool ShouldWrapStream( Stream stream )
 #else
 		internal
 #endif
-		static readonly PackerUnpackerStreamOptions SingletonForAsync =
+		static readonly PackerUnpackerStreamOptions SingletonForAsyncPacking =
+			// It is OK for serialize to do buffering because we explicitly call FlushAsync for it.
 			new PackerUnpackerStreamOptions { OwnsStream = true, WithBuffering = true };
 
+		#if UNITY && DEBUG
+		public
+#else
+		internal
+#endif
+		static readonly PackerUnpackerStreamOptions SingletonForAsyncUnpacking =
+			// Buffering causes data loss in deserialization because buffered bytes will be gone in a tail of Deserialize(Stream) call.
+			new PackerUnpackerStreamOptions { OwnsStream = true, WithBuffering = false };
+
 #if UNITY && DEBUG
 		public
 #else
diff --git a/src/MsgPack/Serialization/MessagePackSerializer`1.cs b/src/MsgPack/Serialization/MessagePackSerializer`1.cs
@@ -219,7 +219,7 @@ public Task PackAsync( Stream stream, T objectTree )
 		/// <seealso cref="P:Capabilities"/>
 		public async Task PackAsync( Stream stream, T objectTree, CancellationToken cancellationToken )
 		{
-			var packer = Packer.Create( stream, this.PackerCompatibilityOptions, PackerUnpackerStreamOptions.SingletonForAsync );
+			var packer = Packer.Create( stream, this.PackerCompatibilityOptions, PackerUnpackerStreamOptions.SingletonForAsyncPacking );
 			try
 			{
 				await this.PackToAsync( packer, objectTree, cancellationToken ).ConfigureAwait( false );
@@ -324,7 +324,7 @@ public Task<T> UnpackAsync( Stream stream )
 		public async Task<T> UnpackAsync( Stream stream, CancellationToken cancellationToken )
 		{
 			// Unpacker does not have finalizer, so just avoiding unpacker disposing prevents stream closing.
-			var unpacker = Unpacker.Create( stream, PackerUnpackerStreamOptions.SingletonForAsync, DefaultUnpackerOptions );
+			var unpacker = Unpacker.Create( stream, PackerUnpackerStreamOptions.SingletonForAsyncUnpacking, DefaultUnpackerOptions );
 			if ( !( await unpacker.ReadAsync( cancellationToken ).ConfigureAwait( false ) ) )
 			{
 				SerializationExceptions.ThrowUnexpectedEndOfStream( unpacker );
diff --git a/test/MsgPack.UnitTest/Serialization/RegressionTests.cs b/test/MsgPack.UnitTest/Serialization/RegressionTests.cs
@@ -547,5 +547,31 @@ public void TestIssue269()
 			var forString = Assert.Throws<InvalidOperationException>( () => target.AsString() );
 			Assert.That( forString.Message, Is.EqualTo( "Do not convert MsgPack.MessagePackExtendedTypeObject MessagePackObject to System.String." ) );
 		}
+
+#if FEATURE_TAP
+		[Test]
+		public async Task TestIssue321_AsyncBuffering()
+		{
+			var context = new SerializationContext();
+			var serializer = context.GetSerializer<string>();
+			using ( var stream = new MyMemoryStream() ) // could also be a NetworkStream or PipeStream
+			{
+				await serializer.PackAsync( stream, "hello" );
+				TestContext.WriteLine( $"1. Stream now in {stream.Position}" );
+				await serializer.PackAsync( stream, "world" );
+				TestContext.WriteLine( $"2. Stream now in {stream.Position}" );
+				TestContext.WriteLine( Binary.ToHexString( stream.ToArray() ) );
+				stream.Position = 0;
+				var result1 = await serializer.UnpackAsync( stream );
+				TestContext.WriteLine( $"3. Stream now in {stream.Position} -> {result1}" );
+				var result2 = await serializer.UnpackAsync( stream ); // throws MsgPack.InvalidMessagePackStreamException
+				TestContext.WriteLine( $"4. Stream now in {stream.Position} -> {result2}" );
+			}
+		}
+#endif // FEATURE_TAP
+
+		public class MyMemoryStream : MemoryStream
+		{
+		}
 	}
 }

Original file line number	Diff line number	Diff line change
`@@ -219,7 +219,7 @@ public Task PackAsync( Stream stream, T objectTree )`
`219`	`219`	`/// <seealso cref="P:Capabilities"/>`
`220`	`220`	`public async Task PackAsync( Stream stream, T objectTree, CancellationToken cancellationToken )`
`221`	`221`	`{`
`222`		`- var packer = Packer.Create( stream, this.PackerCompatibilityOptions, PackerUnpackerStreamOptions.SingletonForAsync );`
	`222`	`+ var packer = Packer.Create( stream, this.PackerCompatibilityOptions, PackerUnpackerStreamOptions.SingletonForAsyncPacking );`
`223`	`223`	`try`
`224`	`224`	`{`
`225`	`225`	`await this.PackToAsync( packer, objectTree, cancellationToken ).ConfigureAwait( false );`
`@@ -324,7 +324,7 @@ public Task<T> UnpackAsync( Stream stream )`
`324`	`324`	`public async Task<T> UnpackAsync( Stream stream, CancellationToken cancellationToken )`
`325`	`325`	`{`
`326`	`326`	`// Unpacker does not have finalizer, so just avoiding unpacker disposing prevents stream closing.`
`327`		`- var unpacker = Unpacker.Create( stream, PackerUnpackerStreamOptions.SingletonForAsync, DefaultUnpackerOptions );`
	`327`	`+ var unpacker = Unpacker.Create( stream, PackerUnpackerStreamOptions.SingletonForAsyncUnpacking, DefaultUnpackerOptions );`
`328`	`328`	`if ( !( await unpacker.ReadAsync( cancellationToken ).ConfigureAwait( false ) ) )`
`329`	`329`	`{`
`330`	`330`	`SerializationExceptions.ThrowUnexpectedEndOfStream( unpacker );`