Update message writer and reader to ignore Message info from DI (#3058)

Update message writer and reader to ignore Message info from DI as it is overwritten.

buildandtest.yml

@@ -20,11 +20,6 @@ steps:
   inputs:
     version: '8.x'
 
-# .NET Core App 2.1 is required to run the ESRP code signing task
-- task: UseDotNet@2
-  inputs:
-    version: '2.1.x'
-
 - task: DotNetCoreCLI@2
   displayName: 'Build'  
   inputs:
@@ -48,4 +43,4 @@ steps:
     command: 'test'
     arguments: '--configuration $(buildConfiguration) --collect "Code coverage"'
     projects: |      
-     $(Build.SourcesDirectory)\test\EndToEndTests\Tests\Client\Microsoft.OData.Client.E2E.Tests\Microsoft.OData.Client.E2E.Tests.csproj
+     $(Build.SourcesDirectory)\test\EndToEndTests\Tests\Client\Microsoft.OData.Client.E2E.Tests\Microsoft.OData.Client.E2E.Tests.csproj

nightly.yml

@@ -11,7 +11,10 @@
           BuildDropPath: '$(Build.ArtifactStagingDirectory)\$(nugetArtifactsDir)\sbom'
 
     - template: credscan.yml
-
+    # .NET Core App 2.1 is required to run the ESRP code signing task
+    - task: UseDotNet@2
+      inputs:
+        version: '2.1.x'
     # CodeSign
     - task: EsrpCodeSigning@1
       displayName: 'ESRP CodeSign - OData'

src/Microsoft.OData.Core/ODataMessageReader.cs

@@ -904,24 +904,18 @@ private ODataMessageInfo GetOrCreateMessageInfo(Stream messageStream, bool isAsy
         {
             if (this.messageInfo == null)
             {
-                if (this.serviceProvider == null)
+                this.messageInfo = new ODataMessageInfo
                 {
-                    this.messageInfo = new ODataMessageInfo();
-                }
-                else
-                {
-                    this.messageInfo = this.serviceProvider.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.ServiceProvider = this.serviceProvider;
-                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,
+                    ServiceProvider = this.serviceProvider,
+                    MessageStream = messageStream,
+                    PayloadKind = this.readerPayloadKind
+                };
             }
 
             return this.messageInfo;

src/Microsoft.OData.Core/ODataMessageWriter.cs

@@ -19,9 +19,9 @@ namespace Microsoft.OData
     using Microsoft.OData.Metadata;
     #endregion Namespaces
 
-/// <summary>
-/// Writer class used to write all OData payloads (entries, resource sets, metadata documents, service documents, etc.).
-/// </summary>
+    /// <summary>
+    /// Writer class used to write all OData payloads (entries, resource sets, metadata documents, service documents, etc.).
+    /// </summary>
 #if NETCOREAPP
     public sealed class ODataMessageWriter : IDisposable, IAsyncDisposable
 #else
@@ -1111,7 +1111,7 @@ private ODataPayloadKind VerifyCanWriteValue(object value)
             // We cannot use ODataRawValueUtils.TryConvertPrimitiveToString for all cases since binary values are
             // converted into unencoded byte streams in the raw format
             // (as opposed to base64 encoded byte streams in the ODataRawValueUtils); see OIPI 2.2.6.4.1.
-            return value is byte[] ? ODataPayloadKind.BinaryValue : ODataPayloadKind.Value;
+            return value is byte[]? ODataPayloadKind.BinaryValue : ODataPayloadKind.Value;
         }
 
         /// <summary>
@@ -1227,23 +1227,18 @@ private ODataMessageInfo GetOrCreateMessageInfo(Stream messageStream, bool isAsy
         {
             if (this.messageInfo == null)
             {
-                if (this.serviceProvider == null)
-                {
-                    this.messageInfo = new ODataMessageInfo();
-                }
-                else
-                {
-                    this.messageInfo = this.serviceProvider.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.ServiceProvider = this.serviceProvider;
-                this.messageInfo.MessageStream = messageStream;
+                this.messageInfo = new ODataMessageInfo
+                {
+                    Encoding = this.encoding,
+                    IsResponse = this.writingResponse,
+                    IsAsync = isAsync,
+                    MediaType = this.mediaType,
+                    Model = this.model,
+                    PayloadUriConverter = this.payloadUriConverter,
+                    ServiceProvider = this.serviceProvider,
+                    MessageStream = messageStream
+                };
             }
 
             return this.messageInfo;
@@ -1321,7 +1316,7 @@ private async Task<TResult> WriteToOutputAsync<TResult>(ODataPayloadKind payload
             this.outputContext = await this.format.CreateOutputContextAsync(
                 this.GetOrCreateMessageInfo(messageStream, true),
                 this.settings).ConfigureAwait(false);
-            
+
             return await writeFunc(this.outputContext).ConfigureAwait(false);
         }
     }

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

@@ -0,0 +1,77 @@
+//---------------------------------------------------------------------
+// <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 Microsoft.Extensions.DependencyInjection;
+using System.Linq;
+using Microsoft.OData.Tests;
+
+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()
+        {
+            ServiceCollection services = new();
+            services.AddDefaultODataServices();
+            ServiceProvider serviceProvider = services.BuildServiceProvider();
+
+            var content1 = string.Concat(Enumerable.Repeat('A', 1000_000));
+            var content2 = string.Concat(Enumerable.Repeat('B', 1000_000));
+            for (int i = 0; i < 1000; i++)
+            {
+                var 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)
+        {
+            await using Stream outputStream = new MemoryStream();
+
+            var message = new InMemoryMessage
+            {
+                Stream = outputStream,
+                ServiceProvider = serviceProvider
+            };
+
+            var responseMessage = new ODataResponseMessage(message, writing: true, enableMessageStreamDisposal: true, maxMessageSize: -1);
+            await using ODataMessageWriter writer = new ODataMessageWriter(responseMessage);
+
+            await Task.Yield();
+
+            await writer.WriteValueAsync(content);
+
+            outputStream.Position = 0;
+            using var reader = new StreamReader(outputStream);
+
+            await Task.Yield();
+
+            string written = await reader.ReadToEndAsync();
+            await writer.DisposeAsync();
+            return written;
+        }
+    }
+}