diff --git a/Directory.Build.props b/Directory.Build.props index 25da027..13b9139 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -1,7 +1,7 @@ net9.0 - 0.0.4 + 0.0.5 rc1 Paul Bleess MIT diff --git a/src/PipeExtensions.cs b/src/PipeExtensions.cs new file mode 100644 index 0000000..6bb4de7 --- /dev/null +++ b/src/PipeExtensions.cs @@ -0,0 +1,13 @@ +using System.IO.Pipelines; + +namespace Yamux +{ + public static class PipeExtensions + { + /// + /// Creates a yamux session from the duplex pipe + /// + public static Session AsYamuxSession(this IDuplexPipe pipe, bool isClient, bool keepOpen = false, SessionOptions? options = null) + => new Session(new PipePeer(pipe), isClient, keepOpen, options); + } +} diff --git a/src/PipePeer.cs b/src/PipePeer.cs new file mode 100644 index 0000000..cbf4e18 --- /dev/null +++ b/src/PipePeer.cs @@ -0,0 +1,66 @@ +using System.IO.Pipelines; + +namespace Yamux +{ + public class PipePeer : ITransport + { + private readonly PipeReader _reader; + private readonly PipeWriter _writer; + + public PipePeer(IDuplexPipe pipe) + { + ArgumentNullException.ThrowIfNull(pipe); + _reader = pipe.Input; + _writer = pipe.Output; + } + + public async ValueTask ReadAsync(Memory data, CancellationToken cancel) + { + if (data.IsEmpty) + return 0; + + var result = await _reader.ReadAsync(cancel); + var buffer = result.Buffer; + + if (buffer.IsEmpty && result.IsCompleted) + { + _reader.AdvanceTo(buffer.End); + return 0; + } + + var len = (int)Math.Min(buffer.Length, data.Length); + var remaining = len; + foreach (var segment in buffer) + { + if (remaining <= 0) + break; + var toCopy = Math.Min(segment.Length, remaining); + segment.Span[..toCopy].CopyTo(data.Span[(len - remaining)..]); + remaining -= toCopy; + } + _reader.AdvanceTo(buffer.GetPosition(len)); + + return len; + } + + public async ValueTask WriteAsync(ReadOnlyMemory data, CancellationToken cancel) + { + if (data.IsEmpty) + return; + + await _writer.WriteAsync(data, cancel); + } + + public void Close() + { + _reader.Complete(); + _writer.Complete(); + } + + public void Dispose() + { + _reader.Complete(); + _writer.Complete(); + } + } +} diff --git a/test/Yamux.Tests/PipeTransportTests.cs b/test/Yamux.Tests/PipeTransportTests.cs new file mode 100644 index 0000000..3b7ab88 --- /dev/null +++ b/test/Yamux.Tests/PipeTransportTests.cs @@ -0,0 +1,98 @@ +using AwesomeAssertions; +using Bogus; +using System.Buffers; +using System.IO.Pipelines; +using System.Text; + +namespace Yamux.Tests; + +public class PipeTransportTests +{ + private static (IDuplexPipe Client, IDuplexPipe Server) CreatePipePair() + { + var clientPipe = new Pipe(); + var serverPipe = new Pipe(); + + var client = new DuplexPipe(serverPipe.Reader, clientPipe.Writer); + var server = new DuplexPipe(clientPipe.Reader, serverPipe.Writer); + + return (client, server); + } + + [Fact] + public async Task SingleOneWayTest() + { + var faker = new Faker(); + var data = faker.Random.Chars(count: 1024 * 750); + var buffer = Encoding.UTF8.GetBytes(data).AsMemory(); + var result = new byte[buffer.Length].AsMemory(); + + (var client, var server) = CreatePipePair(); + + var serverTask = Task.Run(async () => + { + await using var serverSession = server.AsYamuxSession(false); + serverSession.Start(); + + using var channel = await serverSession.AcceptAsync(); + + long index = 0; + ReadResult res; + do + { + res = await channel.Input.ReadAsync(); + if (res.Buffer.Length > 0) + { + res.Buffer.CopyTo(result.Slice((int)index, (int)res.Buffer.Length).Span); + index += res.Buffer.Length; + channel.Input.AdvanceTo(res.Buffer.End, res.Buffer.End); + } + } while (!res.IsCanceled && !res.IsCompleted); + + channel.Close(); + await channel.WhenRemoteCloseAsync(TimeSpan.FromSeconds(1)); + channel.Dispose(); + }); + + var clientTask = Task.Run(async () => + { + await using var clientSession = client.AsYamuxSession(true); + clientSession.Start(); + + using var channel = await clientSession.OpenChannelAsync(); + + var size = buffer.Length; + int current = 0; + int chunkSize = 1024 * 4; + + while (current < size) + { + if (buffer.Length > chunkSize) + { + var end = current + chunkSize; + var slice = buffer.Slice(current, end >= buffer.Length ? buffer.Length - current : chunkSize); + await channel.WriteAsync(slice, CancellationToken.None); + } + current += chunkSize; + } + + channel.Close(); + await channel.WhenRemoteCloseAsync(TimeSpan.FromSeconds(1)); + }); + + await Task.WhenAll(serverTask, clientTask); + result.ToArray().Should().BeEquivalentTo(buffer.ToArray()); + } + + private sealed class DuplexPipe : IDuplexPipe + { + public DuplexPipe(PipeReader input, PipeWriter output) + { + Input = input; + Output = output; + } + + public PipeReader Input { get; } + public PipeWriter Output { get; } + } +} \ No newline at end of file