Implementing WriteFlush() in the main RSocket class. Adding helper method for the Setup() call in the main RSocket class (with protocol support in the native-typed version for metadata&data - must be old code). Augmenting the Client class to take a connection policy object (which was already implemented) and removed all dependency on the Protocol layer (which was an abstraction violation anyway).

Author: dbarnett-netifi

RSocket.Core/RSocket.cs

@@ -61,16 +61,8 @@ public RSocket(IRSocketTransport transport, RSocketOptions options = default)
 		/// <param name="cancel">Cancellation for the handler. Requesting cancellation will stop message handling.</param>
 		/// <returns>The handler task.</returns>
 		public Task Connect(CancellationToken cancel = default) => RSocketProtocol.Handler(this, Transport.Input, cancel);
+		public Task Setup(TimeSpan keepalive, TimeSpan lifetime, string metadataMimeType = null, string dataMimeType = null, ReadOnlySequence<byte> data = default, ReadOnlySequence<byte> metadata = default) => new RSocketProtocol.Setup(keepalive, lifetime, metadataMimeType: metadataMimeType, dataMimeType: dataMimeType, data: data, metadata: metadata).WriteFlush(Transport.Output, data: data, metadata: metadata);
 
-		//public async Task<RSocket> ConnectAsync()
-		//{
-		//	await Transport.StartAsync();
-		//	var server = RSocketProtocol.Handler(this, Transport.Input, CancellationToken.None, name: nameof(RSocketClient));
-		//	////TODO Move defaults to policy object
-		//	new RSocketProtocol.Setup(keepalive: TimeSpan.FromSeconds(60), lifetime: TimeSpan.FromSeconds(180), metadataMimeType: "binary", dataMimeType: "binary").Write(Transport.Output);
-		//	await Transport.Output.FlushAsync();
-		//	return this;
-		//}
 
 		//TODO SPEC: A requester MUST not send PAYLOAD frames after the REQUEST_CHANNEL frame until the responder sends a REQUEST_N frame granting credits for number of PAYLOADs able to be sent.
 
@@ -82,8 +74,7 @@ public IAsyncEnumerable<T> RequestChannel<TSource, T>(IAsyncEnumerable<TSource>
 		public async Task<IRSocketChannel> RequestChannel(IRSocketStream stream, ReadOnlySequence<byte> data, ReadOnlySequence<byte> metadata = default, int initial = RSocketOptions.INITIALDEFAULT)
 		{
 			var id = StreamDispatch(stream);
-			new RSocketProtocol.RequestChannel(id, data, metadata, initialRequest: Options.GetInitialRequestSize(initial)).Write(Transport.Output, data, metadata);
-			await Transport.Output.FlushAsync();
+			await new RSocketProtocol.RequestChannel(id, data, metadata, initialRequest: Options.GetInitialRequestSize(initial)).WriteFlush(Transport.Output, data, metadata);
 			var channel = new ChannelHandler(this, id);
 			return channel;
 		}
@@ -98,9 +89,7 @@ protected class ChannelHandler : IRSocketChannel       //TODO hmmm...
 			public Task Send((ReadOnlySequence<byte> metadata, ReadOnlySequence<byte> data) value)
 			{
 				if (!Socket.Dispatcher.ContainsKey(Stream)) { throw new InvalidOperationException("Channel is closed"); }
-				new RSocketProtocol.Payload(Stream, value.data, value.metadata, next: true).Write(Socket.Transport.Output, value.data, value.metadata);
-				var result = Socket.Transport.Output.FlushAsync();
-				return result.IsCompleted ? Task.CompletedTask : result.AsTask();
+				return new RSocketProtocol.Payload(Stream, value.data, value.metadata, next: true).WriteFlush(Socket.Transport.Output, value.data, value.metadata);
 			}
 		}
 
@@ -113,9 +102,7 @@ public Task RequestStream(IRSocketStream stream, ReadOnlySequence<byte> data, Re
 		{
 			if (initial <= INITIALDEFAULT) { initial = Options.InitialRequestSize; }
 			var id = StreamDispatch(stream);
-			new RSocketProtocol.RequestStream(id, data, metadata, initialRequest: Options.GetInitialRequestSize(initial)).Write(Transport.Output, data, metadata);
-			var result = Transport.Output.FlushAsync();
-			return result.IsCompleted ? Task.CompletedTask : result.AsTask();
+			return new RSocketProtocol.RequestStream(id, data, metadata, initialRequest: Options.GetInitialRequestSize(initial)).WriteFlush(Transport.Output, data, metadata);
 		}
 
 		public Task<T> RequestResponse<T>(Func<(ReadOnlySequence<byte> data, ReadOnlySequence<byte> metadata), T> resultmapper,
@@ -125,9 +112,7 @@ public Task<T> RequestResponse<T>(Func<(ReadOnlySequence<byte> data, ReadOnlySeq
 		public Task RequestResponse(IRSocketStream stream, ReadOnlySequence<byte> data, ReadOnlySequence<byte> metadata = default)
 		{
 			var id = StreamDispatch(stream);
-			new RSocketProtocol.RequestResponse(id, data, metadata).Write(Transport.Output, data, metadata);
-			var result = Transport.Output.FlushAsync();
-			return result.IsCompleted ? Task.CompletedTask : result.AsTask();
+			return new RSocketProtocol.RequestResponse(id, data, metadata).WriteFlush(Transport.Output, data, metadata);
 		}
 
 
@@ -138,9 +123,7 @@ public Task RequestFireAndForget(
 		public Task RequestFireAndForget(IRSocketStream stream, ReadOnlySequence<byte> data, ReadOnlySequence<byte> metadata = default)
 		{
 			var id = StreamDispatch(stream);
-			new RSocketProtocol.RequestFireAndForget(id, data, metadata).Write(Transport.Output, data, metadata);
-			var result = Transport.Output.FlushAsync();
-			return result.IsCompleted ? Task.CompletedTask : result.AsTask();
+			return new RSocketProtocol.RequestFireAndForget(id, data, metadata).WriteFlush(Transport.Output, data, metadata);
 		}
 
 

RSocket.Core/RSocketClient.cs

@@ -20,16 +20,18 @@ public class RSocketClient : RSocket
 
 		public RSocketClient(IRSocketTransport transport, RSocketOptions options = default) : base(transport, options) { }
 
-		public async Task ConnectAsync()
+		public Task ConnectAsync(RSocketOptions options = default, byte[] data = default, byte[] metadata = default) => ConnectAsync(options, data: data == default ? default : new ReadOnlySequence<byte>(data), metadata: metadata == default ? default : new ReadOnlySequence<byte>(metadata));
+
+		public async Task ConnectAsync(RSocketOptions options = default, ReadOnlySequence<byte> metadata = default, ReadOnlySequence<byte> data = default)
 		{
+			options = options ?? RSocketOptions.Default;
 			await Transport.StartAsync();
 			Handler = Connect(CancellationToken.None);
-			////TODO Move defaults to policy object, also, maybe not inline, like prefer a method.
-			new RSocketProtocol.Setup(keepalive: TimeSpan.FromSeconds(60), lifetime: TimeSpan.FromSeconds(180), metadataMimeType: "binary", dataMimeType: "binary").Write(Transport.Output);
-			await Transport.Output.FlushAsync();
+			await Setup(options.KeepAlive, options.Lifetime, options.MetadataMimeType, options.DataMimeType, data: data, metadata: metadata);
 		}
 
-		//Ugh, these are all garbage. Remove in favor of the transformation ones.
+
+		//TODO Ugh, these are all garbage. Remove in favor of the transformation ones.
 		public Task<IRSocketChannel> RequestChannel<TData>(IRSocketStream stream, TData data, ReadOnlySpan<byte> metadata = default, int initial = INITIALDEFAULT) => RequestChannel(stream, RequestDataSerializer.Serialize(data), metadata, initial);
 		public Task<IRSocketChannel> RequestChannel<TMetadata>(IRSocketStream stream, ReadOnlySpan<byte> data, TMetadata metadata = default, int initial = INITIALDEFAULT) => RequestChannel(stream, data, RequestMetadataSerializer.Serialize(metadata), initial);
 		public Task<IRSocketChannel> RequestChannel<TData, TMetadata>(IRSocketStream stream, TData data, TMetadata metadata = default, int initial = INITIALDEFAULT) => RequestChannel(stream, RequestDataSerializer.Serialize(data), RequestMetadataSerializer.Serialize(metadata), initial);

RSocket.Core/RSocketProtocol.cs

@@ -765,7 +765,7 @@ public ref struct Setup
 			public int Length => Header.Length + InnerLength + Header.MetadataHeaderLength + MetadataLength + DataLength;
 
 
-			public Setup(TimeSpan keepalive, TimeSpan lifetime, string metadataMimeType = null, string dataMimeType = null) : this((int)keepalive.TotalMilliseconds, (int)lifetime.TotalMilliseconds, string.IsNullOrEmpty(metadataMimeType) ? string.Empty : metadataMimeType, string.IsNullOrEmpty(dataMimeType) ? string.Empty : dataMimeType) { }
+			public Setup(TimeSpan keepalive, TimeSpan lifetime, string metadataMimeType = null, string dataMimeType = null, ReadOnlySequence<byte> data = default, ReadOnlySequence<byte> metadata = default) : this((int)keepalive.TotalMilliseconds, (int)lifetime.TotalMilliseconds, string.IsNullOrEmpty(metadataMimeType) ? string.Empty : metadataMimeType, string.IsNullOrEmpty(dataMimeType) ? string.Empty : dataMimeType, data: data, metadata: metadata) { }
 
 			public Setup(Int32 keepalive, Int32 lifetime, string metadataMimeType, string dataMimeType, byte[] resumeToken = default, ReadOnlySequence<byte> data = default, ReadOnlySequence<byte> metadata = default)
 			{