Merging AsyncEnumerable native library support and refactorings.

Author: dbarnett-netifi

.editorconfig

@@ -0,0 +1,10 @@
+# editorconfig.org
+root = true
+
+[*]
+indent_style = tab
+indent_size = tab
+end_of_line = lf
+charset = utf-8
+trim_trailing_whitespace = true
+insert_final_newline = true

RSocket.Core.Tests/ClientServerTests.cs

@@ -5,14 +5,17 @@
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using RSocket.Transports;
 
+using IRSocketStream = System.IObserver<(System.Buffers.ReadOnlySequence<byte> metadata, System.Buffers.ReadOnlySequence<byte> data)>;
+
 namespace RSocket.Tests
 {
 	[TestClass]
 	public class ClientServerTests
 	{
 
 		[TestClass]
-		public class ClientTests
+        [Ignore]
+        public class ClientTests
 		{
 			Lazy<RSocketClient> _Client;
 			RSocketClient Client => _Client.Value;

RSocket.Core.Tests/Extensions.cs

@@ -13,11 +13,11 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="akarnokd.async-enumerable-dotnet" Version="0.0.2" />
-    <PackageReference Include="AsyncEnumerator" Version="2.2.2" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.9.0" />
     <PackageReference Include="MSTest.TestAdapter" Version="1.4.0" />
     <PackageReference Include="MSTest.TestFramework" Version="1.4.0" />
+    <PackageReference Include="System.Linq.Async" Version="4.0.0-preview.1.build.745" />
+    <PackageReference Include="System.Reactive" Version="4.1.3" />
   </ItemGroup>
 
   <ItemGroup>

RSocket.Core.Tests/RSocket.Core.Tests.csproj

@@ -5,7 +5,8 @@
 using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using RSocket.Transports;
-using async_enumerable_dotnet;
+using System.Linq;
+using System.Reactive.Linq;
 
 namespace RSocket.Tests
 {
@@ -21,11 +22,11 @@ public class ServerAlternateProducerTests
 		[TestMethod]
 		public async Task ServerRequestStreamTest()
 		{
-			Server.Streamer = ((ReadOnlySequence<byte> Data, ReadOnlySequence<byte> Metadata) request) =>
-				AsyncEnumerable.Interval(TimeSpan.FromMilliseconds(10))
-					.Take(3)
-					.Map(i => (request.Data, request.Metadata))
-					.ToAsyncEnumerable();
+            Server.Streamer = ((ReadOnlySequence<byte> Data, ReadOnlySequence<byte> Metadata) request) =>
+                Observable.Interval(TimeSpan.FromMilliseconds(10))
+                    .Take(3)
+                    .Select(i => (request.Data, request.Metadata))
+                    .ToAsyncEnumerable();
 
 			var (data, metadata) = ("TEST DATA", "METADATA?_____");
 

RSocket.Core.Tests/ServerAlternateProducerTests.cs

@@ -36,11 +36,9 @@ public async Task ServerRequestResponseTest()
 		[TestMethod]
 		public async Task ServerRequestStreamTest()
 		{
-			Server.Streamer = ((ReadOnlySequence<byte> Data, ReadOnlySequence<byte> Metadata) request) => new System.Collections.Async.AsyncEnumerable<(ReadOnlySequence<byte> data, ReadOnlySequence<byte> metadata)>(async yield =>
-			{
-				foreach (var index in Enumerable.Range(0, 3))
-				{ await Task.CompletedTask; await yield.ReturnAsync((request.Data, request.Metadata)); }
-			}).ToAsyncEnumerable();
+            Server.Streamer = ((ReadOnlySequence<byte> Data, ReadOnlySequence<byte> Metadata) request) =>
+                AsyncEnumerable.Range(0, 3)
+                    .Select(i => (request.Data, request.Metadata));
 
 			var (data, metadata) = ("TEST DATA", "METADATA?_____");
 			var list = await StringClient.RequestStream(data, metadata).ToListAsync();
@@ -52,15 +50,11 @@ public async Task ServerRequestStreamTest()
 		public async Task ServerRequestStreamBinaryDetailsTest()
 		{
 			var count = 20;
-			Server.Streamer = ((ReadOnlySequence<byte> Data, ReadOnlySequence<byte> Metadata) request) => new System.Collections.Async.AsyncEnumerable<(ReadOnlySequence<byte> data, ReadOnlySequence<byte> metadata)>(async yield =>
-			{
-				foreach (byte index in Enumerable.Range(0, count))
-				{
-					await yield.ReturnAsync((
-						new ReadOnlySequence<byte>(request.Data.ToArray().Skip(index).Take(1).ToArray()),
-						new ReadOnlySequence<byte>(request.Metadata.ToArray().Skip(index).Take(1).ToArray())));
-				}
-			}).ToAsyncEnumerable();
+            Server.Streamer = ((ReadOnlySequence<byte> Data, ReadOnlySequence<byte> Metadata) request) =>
+                AsyncEnumerable.Range(0, count)
+                    .Select(i => (
+                        new ReadOnlySequence<byte>(request.Data.ToArray().Skip(i).Take(1).ToArray()),
+                        new ReadOnlySequence<byte>(request.Metadata.ToArray().Skip(i).Take(1).ToArray())));
 
 			var (requestData, requestMetadata) = (Enumerable.Range(1, count).Select(i => (byte)i).ToArray(), Enumerable.Range(100, count).Select(i => (byte)i).ToArray());
 			var list = await Client.RequestStream(result => (Data: result.data.ToArray(), Metadata: result.metadata.ToArray()), new ReadOnlySequence<byte>(requestData), new ReadOnlySequence<byte>(requestMetadata)).ToListAsync();

RSocket.Core.Tests/ServerTests.cs

@@ -9,6 +9,8 @@
 using System.Collections.Concurrent;
 using RSocket.Transports;
 
+using IRSocketStream = System.IObserver<(System.Buffers.ReadOnlySequence<byte> metadata, System.Buffers.ReadOnlySequence<byte> data)>;
+
 namespace RSocket.Tests
 {
 	public class TestServer : RSocket

RSocket.Core.Tests/TestServer.cs

@@ -19,6 +19,7 @@
     <PackageReference Include="Microsoft.Extensions.Logging.Debug" Version="2.2.0" />
     <PackageReference Include="System.IO.Pipelines" Version="4.5.3" />
     <PackageReference Include="System.Linq.Async" Version="4.0.0-preview.1.build.745" />
+    <PackageReference Include="System.Reactive" Version="4.1.3" />
   </ItemGroup>
 
 </Project>