Channel-Producer updates (interim). Committing to merge forthcoming changes to simplify from the System.Reactive.Linq oprerators pull request.

Author: dbarnett-netifi

RSocket.Core.Tests/ServerTests.cs

@@ -77,11 +77,17 @@ await yield.ReturnAsync((
 		//[TestMethod]
 		//public async Task ServerRequestChannelTest()
 		//{
-		//	Server.Streamer = ((ReadOnlySequence<byte> Data, ReadOnlySequence<byte> Metadata) request) => new System.Collections.Async.AsyncEnumerable<(ReadOnlySequence<byte> data, ReadOnlySequence<byte> metadata)>(async yield =>
+		//	Server.Channeler = ((ReadOnlySequence<byte> Data, ReadOnlySequence<byte> Metadata) request, IObservable<(ReadOnlySequence<byte> Data, ReadOnlySequence<byte> Metadata)> incoming) =>
 		//	{
-		//		foreach (var index in Enumerable.Range(0, 3))
-		//		{ await Task.CompletedTask; await yield.ReturnAsync((request.Data, request.Metadata)); }
-		//	}).ToAsyncEnumerable();
+		//		Action<(ReadOnlySequence<byte> Data, ReadOnlySequence<byte> Metadata)> onNext = value => { };
+		//		Action OnCompleted = () => { };
+		//		var enumerable = 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();
+		//		return (onNext, OnCompleted, enumerable);
+		//	};
 
 		//	var (data, metadata) = ("TEST DATA", "METADATA?_____");
 		//	var list = await StringClient.RequestStream(data, metadata).ToListAsync();
@@ -101,22 +107,4 @@ public void TestInitialize()
 			StringClient = new RSocketClient.ForStrings(Client);
 		}
 	}
-
-
-	public class Sample
-	{
-		static Random random = new Random(1234);
-		public int Id = random.Next(1000000);
-		public string Name = nameof(Sample) + random.Next(10000).ToString();
-		public DateTime Created = DateTime.Now;
-
-		public static implicit operator string(Sample value) => string.Join('|', value.Id, value.Name, value.Created);
-		public static implicit operator ReadOnlySequence<byte>(Sample value) => new ReadOnlySequence<byte>(Encoding.UTF8.GetBytes(string.Join('|', value.Id, value.Name, value.Created)));
-		public static implicit operator Sample(string value) { var values = value.Split('|'); return new Sample(values[0], values[1], values[2]); }
-		public static implicit operator Sample(ReadOnlySequence<byte> value) => Encoding.UTF8.GetString(value.ToArray());
-
-		public Sample() { }
-		public Sample(string id, string name, string created) { Id = int.Parse(id); Name = name; Created = DateTime.Parse(created); }
-		public ReadOnlySequence<byte> Bytes => this;
-	}
 }

RSocket.Core/IRSocketStream.cs

@@ -5,7 +5,7 @@
 namespace RSocket
 {
 	/// <summary>
-	/// A stream of items from an RSocket. This is simply an Observer of the protcol's tuples.
+	/// A stream of items from an RSocket. This is simply an Observer of the protocol's tuples.
 	/// </summary>
 	public interface IRSocketStream : IObserver<(ReadOnlySequence<byte> metadata, ReadOnlySequence<byte> data)>
 	{

RSocket.Core/RSocket.cs

@@ -184,8 +184,8 @@ async Task Respond(int stream)
 
 		void IRSocketProtocol.RequestStream(in RSocketProtocol.RequestStream message, ReadOnlySequence<byte> metadata, ReadOnlySequence<byte> data)
 		{
-			Respond(message.Stream).Start();
-			async Task Respond(int stream)
+			Stream(message.Stream).Start();
+			async Task Stream(int stream)
 			{
 				var source = Streamer((data, metadata));     //TODO Handle Errors.
 				var enumerator = source.GetAsyncEnumerator();
@@ -205,36 +205,55 @@ async Task Respond(int stream)
 		}
 
 
-		public Func<IAsyncEnumerable<(ReadOnlySequence<byte> Data, ReadOnlySequence<byte> Metadata)>, IAsyncEnumerable<(ReadOnlySequence<byte> data, ReadOnlySequence<byte> metadata)>> Channeler { get; set; } = request => throw new NotImplementedException();
+		//public void Channel<TSource, TResult>(IAsyncEnumerable<TSource> outgoing, 
+		//	Func<TSource, (ReadOnlySequence<byte> data, ReadOnlySequence<byte> metadata)> outgoingMapper,
+		//	Func<(ReadOnlySequence<byte> data, ReadOnlySequence<byte> metadata), TResult> incomingMapper)
+		//{
+		//	var receiver = new Receiver<TSource>(stream => Task.CompletedTask, sourceMapper);
 
-		void IRSocketProtocol.RequestChannel(in RSocketProtocol.RequestChannel message, ReadOnlySequence<byte> metadata, ReadOnlySequence<byte> data) 
-		{
-			throw new NotImplementedException();
-			//Respond(message.Stream).Start();
-
-			//new Receiver<bool>()
-
-			//new Receiver<bool>(stream => RequestFireAndForget(stream, data, metadata), _ => true).ExecuteAsync(result: true);
-			//var id = StreamDispatch(stream);
-
-
-			//async Task Respond(int stream)
-			//{
-			//	var source = Channeler((data, metadata));     //TODO Handle Errors.
-			//	var enumerator = source.GetAsyncEnumerator();
-			//	try
-			//	{
-			//		while (await enumerator.MoveNextAsync())
-			//		{
-			//			var (Data, Metadata) = enumerator.Current;
-			//			new RSocketProtocol.Payload(stream, Data, Metadata, next: true).Write(Transport.Output, Data, Metadata);
-			//			await Transport.Output.FlushAsync();
-			//		}
-			//		new RSocketProtocol.Payload(stream, complete: true).Write(Transport.Output);
-			//		await Transport.Output.FlushAsync();
-			//	}
-			//	finally { await enumerator.DisposeAsync(); }
-			//}
-		}
+		//	Channeler = request =>
+		//	(
+		//		receiver, 
+		//	);
+		//}
+
+
+		//public Func<(ReadOnlySequence<byte> Data, ReadOnlySequence<byte> Metadata),
+		//		(IObservable<(ReadOnlySequence<byte> data, ReadOnlySequence<byte> metadata)> Incoming,
+		//		IAsyncEnumerable<(ReadOnlySequence<byte> data, ReadOnlySequence<byte> metadata)> Outgoing)>
+		//	 Channeler { get; set; } = (request, incoming) => throw new NotImplementedException();
+
+		//void IRSocketProtocol.RequestChannel(in RSocketProtocol.RequestChannel message, ReadOnlySequence<byte> metadata, ReadOnlySequence<byte> data) 
+		//{
+		//	Channel(message.Stream).Start();
+
+		//	//new Receiver<bool>()
+
+		//	//new Receiver<bool>(stream => RequestFireAndForget(stream, data, metadata), _ => true).ExecuteAsync(result: true);
+		//	//var id = StreamDispatch(stream);
+
+		//	async Task Channel(int stream)
+		//	{
+		//		var (Incoming, Outoing) = Channeler((data, metadata));     //TODO Handle Errors.
+
+
+		//		using (observable.Subscribe())
+		//		{
+		//			var enumerator = source.GetAsyncEnumerator();
+		//			try
+		//			{
+		//				while (await enumerator.MoveNextAsync())
+		//				{
+		//					var (Data, Metadata) = enumerator.Current;
+		//					new RSocketProtocol.Payload(stream, Data, Metadata, next: true).Write(Transport.Output, Data, Metadata);
+		//					await Transport.Output.FlushAsync();
+		//				}
+		//				new RSocketProtocol.Payload(stream, complete: true).Write(Transport.Output);
+		//				await Transport.Output.FlushAsync();
+		//			}
+		//			finally { await enumerator.DisposeAsync(); }
+		//		}
+		//	}
+		//}
 	}
 }