Port Update message writer and reader to ignore Message info from DI (#…

…3058) to ODL 7x (#3072)

src/Microsoft.OData.Core/ODataMessageReader.cs

@@ -935,24 +935,18 @@ private ODataMessageInfo GetOrCreateMessageInfo(Stream messageStream, bool isAsy
         {
             if (this.messageInfo == null)
             {
-                if (this.container == null)
+                this.messageInfo = new ODataMessageInfo
                 {
-                    this.messageInfo = new ODataMessageInfo();
-                }
-                else
-                {
-                    this.messageInfo = this.container.GetRequiredService<ODataMessageInfo>();
-                }
-
-                this.messageInfo.Encoding = this.encoding;
-                this.messageInfo.IsResponse = this.readingResponse;
-                this.messageInfo.IsAsync = isAsync;
-                this.messageInfo.MediaType = this.contentType;
-                this.messageInfo.Model = this.model;
-                this.messageInfo.PayloadUriConverter = this.payloadUriConverter;
-                this.messageInfo.Container = this.container;
-                this.messageInfo.MessageStream = messageStream;
-                this.messageInfo.PayloadKind = this.readerPayloadKind;
+                    Encoding = this.encoding,
+                    IsResponse = this.readingResponse,
+                    IsAsync = isAsync,
+                    MediaType = this.contentType,
+                    Model = this.model,
+                    PayloadUriConverter = this.payloadUriConverter,
+                    Container = this.container,
+                    MessageStream = messageStream,
+                    PayloadKind = this.readerPayloadKind
+                };
             }
 
             return this.messageInfo;

src/Microsoft.OData.Core/ODataMessageWriter.cs

@@ -1257,23 +1257,17 @@ private ODataMessageInfo GetOrCreateMessageInfo(Stream messageStream, bool isAsy
         {
             if (this.messageInfo == null)
             {
-                if (this.container == null)
+                this.messageInfo = new ODataMessageInfo
                 {
-                    this.messageInfo = new ODataMessageInfo();
-                }
-                else
-                {
-                    this.messageInfo = this.container.GetRequiredService<ODataMessageInfo>();
-                }
-
-                this.messageInfo.Encoding = this.encoding;
-                this.messageInfo.IsResponse = this.writingResponse;
-                this.messageInfo.IsAsync = isAsync;
-                this.messageInfo.MediaType = this.mediaType;
-                this.messageInfo.Model = this.model;
-                this.messageInfo.PayloadUriConverter = this.payloadUriConverter;
-                this.messageInfo.Container = this.container;
-                this.messageInfo.MessageStream = messageStream;
+                    Encoding = this.encoding,
+                    IsResponse = this.writingResponse,
+                    IsAsync = isAsync,
+                    MediaType = this.mediaType,
+                    Model = this.model,
+                    PayloadUriConverter = this.payloadUriConverter,
+                    Container = this.container,
+                    MessageStream = messageStream
+                };
             }
 
             return this.messageInfo;

test/FunctionalTests/Microsoft.OData.Core.Tests/MessageWritterConcurrencyTests.cs

@@ -0,0 +1,94 @@
+//---------------------------------------------------------------------
+// <copyright file="MessageWriterConcurrencyTests.cs" company="Microsoft">
+//      Copyright (C) Microsoft Corporation. All rights reserved. See License.txt in the project root for license information.
+// </copyright>
+//---------------------------------------------------------------------
+
+using System.IO;
+using System.Threading.Tasks;
+using System;
+using Xunit;
+using System.Linq;
+using Microsoft.OData.Tests;
+using Microsoft.Test.OData.DependencyInjection;
+
+namespace Microsoft.OData.Core.Tests
+{
+    public class MessageWriterConcurrencyTests
+    {
+        /// <summary>
+        /// Verifies that concurrent message writer does not interleave execution and isolates the underlying streams.
+        /// </summary>
+        /// <returns>A task for the asynchronous test</returns>
+        [Fact]
+        public async Task VerifyConcurrentResultsAreIsolatedAsync()
+        {
+            TestContainerBuilder containerBuilder = new TestContainerBuilder();
+            containerBuilder.AddDefaultODataServices();
+            IServiceProvider serviceProvider = containerBuilder.BuildContainer();
+
+            string content1 = string.Concat(Enumerable.Repeat('A', 1000_000));
+            string content2 = string.Concat(Enumerable.Repeat('B', 1000_000));
+
+            await TaskUtils.CompletedTask;
+            for (int i = 0; i < 1000; i++)
+            {
+                string[] values = await Task.WhenAll(WritePayloadAsync(content1, serviceProvider), WritePayloadAsync(content2, serviceProvider));
+                Assert.Equal(content1.Length, values[0].Length);
+                Assert.Equal(content2.Length, values[1].Length);
+
+                Assert.Equal(content1, values[0]);
+                Assert.Equal(content2, values[1]);
+            }
+        }
+
+
+        /// <summary>
+        /// A helper function that writes to a stream using the message writer and returns the content that is present in the stream.
+        /// </summary>
+        /// <param name="content">String content to write.</param>
+        /// <param name="serviceProvider">A service provider with the default configurations.</param>
+        /// <returns>A task that resolves to the string present in the output stream.</returns>
+        private static async Task<string> WritePayloadAsync(string content, IServiceProvider serviceProvider)
+        {
+            using (Stream outputStream = new MemoryStream())
+            {
+
+                var message = new InMemoryMessage
+                {
+                    Stream = outputStream,
+                    Container = serviceProvider
+                };
+
+                var responseMessage = new ODataResponseMessage(message, writing: true, enableMessageStreamDisposal: true, maxMessageSize: -1);
+#if NETCOREAPP3_1_OR_GREATER
+
+                await using (ODataMessageWriter writer = new ODataMessageWriter(responseMessage))
+                {
+#else
+                using (ODataMessageWriter writer = new ODataMessageWriter(responseMessage))
+                {
+#endif
+                    await Task.Yield();
+
+                    await writer.WriteValueAsync(content);
+
+                    outputStream.Position = 0;
+                    StreamReader reader = new StreamReader(outputStream);
+
+
+                    await Task.Yield();
+
+                    string response = await reader.ReadToEndAsync();
+#if NETCOREAPP3_1_OR_GREATER
+
+                    await writer.DisposeAsync();
+#else
+                    writer.Dispose();
+#endif
+                    return response;
+                }
+            }
+        }
+    }
+}