📝 Update usage sample and docs

Author: kzu

readme.md

@@ -18,18 +18,37 @@ await client.ConnectAsync(serverUri, CancellationToken.None);
 
 using IWebSocketPipe pipe = WebSocketPipe.Create(client, closeWhenCompleted: true);
 
-var read = Task.Run(async () =>
-{
-    var read = await pipe.Input.ReadAsync();
-    Output.WriteLine("Client: " + Encoding.UTF8.GetString(read.Buffer));
-    await pipe.CompleteAsync(WebSocketCloseStatus.NormalClosure, "Client Done");
-});
-
+// Start the pipe before hooking up the processing
 var run = pipe.RunAsync();
 
-await pipe.Output.WriteAsync(Encoding.UTF8.GetBytes("hello").AsMemory());
-
+// Wait for completion of processing code
+await Task.WhenAny(
+    ReadIncoming(pipe.Input),
+    SendOutgoing(pipe.Output));
 
+// When the processing completes, the overall pipe run will also complete
+await run;
 
+// Reads incoming data and writes to the console
+async Task ReadIncoming(PipeReader reader)
+{
+    while (await reader.ReadAsync() is var result && !result.IsCompleted)
+    {
+        Console.WriteLine($"Received: {Encoding.UTF8.GetString(result.Buffer)}");
+        reader.AdvanceTo(result.Buffer.Start, result.Buffer.End);
+    }
+    Console.WriteLine($"Done reading.");
+}
+
+// Reads console input and writes to pipe until an empty line is entered
+async Task SendOutgoing(PipeWriter writer)
+{
+    while (Console.ReadLine() is var line && line?.Length > 0)
+    {
+        Encoding.UTF8.GetBytes(line, writer);
+    }
+    await writer.CompleteAsync();
+    Console.WriteLine($"Done writing.");
+}
 ```
 

src/Tests/EndToEnd.cs

@@ -28,7 +28,10 @@ public async Task RunAsync()
 
             var run = pipe.RunAsync();
 
-            await pipe.Output.WriteAsync(Encoding.UTF8.GetBytes("hello").AsMemory());
+            pipe.Output.Advance(Encoding.UTF8.GetBytes("hello", pipe.Output.GetSpan()));
+            await pipe.Output.FlushAsync();
+
+            //await pipe.Output.WriteAsync(Encoding.UTF8.GetBytes("hello").AsMemory());
 
             await read;
             await run;
@@ -44,7 +47,7 @@ async Task Echo(IDuplexPipe pipe)
                 Output.WriteLine($"Echoing: {Encoding.UTF8.GetString(result.Buffer)}");
                 // Just assume we get a single-segment entry, for simplicity
                 await pipe.Output.WriteAsync(result.Buffer.First);
-                pipe.Input.AdvanceTo(result.Buffer.Start, result.Buffer.End);
+                pipe.Input.AdvanceTo(result.Buffer.End);
             }
             Output.WriteLine($"Server: Done.");
         }

src/Tests/SimpleWebSocketPipeTests.cs

@@ -176,7 +176,7 @@ public async Task WhenReceivingChunks_PipeReaderExposesFullMessage()
                     for (var i = 0; i < result.Buffer.Length; i++)
                         await s.SendAsync(result.Buffer.Slice(i, 1).First, WebSocketMessageType.Binary, i == result.Buffer.Length - 1, default);
 
-                    serverPipe.Input.AdvanceTo(result.Buffer.Start, result.Buffer.End);
+                    serverPipe.Input.AdvanceTo(result.Buffer.End);
                 }
                 await serverRun;
             }, null, Output);
@@ -198,7 +198,7 @@ async Task Echo(IWebSocketPipe pipe)
             while (await pipe.Input.ReadAsync() is var result && !result.IsCompleted)
             {
                 await pipe.Output.WriteAsync(result.Buffer.First);
-                pipe.Input.AdvanceTo(result.Buffer.Start, result.Buffer.End);
+                pipe.Input.AdvanceTo(result.Buffer.End);
             }
         }
     }

src/WebSocketPipe/SimpleWebSocketPipe.cs

@@ -44,7 +44,7 @@ public SimpleWebSocketPipe(WebSocket webSocket, WebSocketPipeOptions options)
         public async Task RunAsync(CancellationToken cancellation = default)
         {
             if (webSocket.State != WebSocketState.Open)
-                throw new InvalidOperationException();
+                throw new InvalidOperationException($"WebSocket must be opened. State was {webSocket.State}");
 
             var writing = FillInputAsync(cancellation);
             var reading = SendOutputAsync(cancellation);
@@ -162,7 +162,7 @@ async Task SendOutputAsync(CancellationToken cancellation)
                         }
                     }
 
-                    outputPipe.Reader.AdvanceTo(result.Buffer.Start, result.Buffer.End);
+                    outputPipe.Reader.AdvanceTo(result.Buffer.End);
 
                 }
                 catch (Exception ex) when (ex is OperationCanceledException ||