[IO]System.IO.Pipelines

 1 async Task ProcessLinesAsync(NetworkStream stream)
 2 {
 3     byte[] buffer = ArrayPool<byte>.Shared.Rent(1024);
 4     var bytesBuffered = 0;
 5     var bytesConsumed = 0;
 6 
 7     while (true)
 8     {
 9         // Calculate the amount of bytes remaining in the buffer.
10         var bytesRemaining = buffer.Length - bytesBuffered;
11 
12         if (bytesRemaining == 0)
13         {
14             // Double the buffer size and copy the previously buffered data into the new buffer.
15             var newBuffer = ArrayPool<byte>.Shared.Rent(buffer.Length * 2);
16             Buffer.BlockCopy(buffer, 0, newBuffer, 0, buffer.Length);
17             // Return the old buffer to the pool.
18             ArrayPool<byte>.Shared.Return(buffer);
19             buffer = newBuffer;
20             bytesRemaining = buffer.Length - bytesBuffered;
21         }
22 
23         var bytesRead = await stream.ReadAsync(buffer, bytesBuffered, bytesRemaining);
24         if (bytesRead == 0)
25         {
26             // EOF
27             break;
28         }
29 
30         // Keep track of the amount of buffered bytes.
31         bytesBuffered += bytesRead;
32         var linePosition = -1;
33 
34         do
35         {
36             // Look for a EOL in the buffered data.
37             linePosition = Array.IndexOf(buffer, (byte)‘\n‘, bytesConsumed,
38                                          bytesBuffered - bytesConsumed);
39 
40             if (linePosition >= 0)
41             {
42                 // Calculate the length of the line based on the offset.
43                 var lineLength = linePosition - bytesConsumed;
44 
45                 // Process the line.
46                 ProcessLine(buffer, bytesConsumed, lineLength);
47 
48                 // Move the bytesConsumed to skip past the line consumed (including \n).
49                 bytesConsumed += lineLength + 1;
50             }
51         }
52         while (linePosition >= 0);
53     }
54 }

 1 async Task ProcessLinesAsync(Socket socket)
 2 {
 3     var pipe = new Pipe();
 4     Task writing = FillPipeAsync(socket, pipe.Writer);
 5     Task reading = ReadPipeAsync(pipe.Reader);
 6 
 7     await Task.WhenAll(reading, writing);
 8 }
 9 
10 async Task FillPipeAsync(Socket socket, PipeWriter writer)
11 {
12     const int minimumBufferSize = 512;
13 
14     while (true)
15     {
16         // Allocate at least 512 bytes from the PipeWriter.
17         Memory<byte> memory = writer.GetMemory(minimumBufferSize);
18         try
19         {
20             int bytesRead = await socket.ReceiveAsync(memory, SocketFlags.None);
21             if (bytesRead == 0)
22             {
23                 break;
24             }
25             // Tell the PipeWriter how much was read from the Socket.
26             writer.Advance(bytesRead);
27         }
28         catch (Exception ex)
29         {
30             LogError(ex);
31             break;
32         }
33 
34         // Make the data available to the PipeReader.
35         FlushResult result = await writer.FlushAsync();
36 
37         if (result.IsCompleted)
38         {
39             break;
40         }
41     }
42 
43      // By completing PipeWriter, tell the PipeReader that there‘s no more data coming.
44     await writer.CompleteAsync();
45 }
46 
47 async Task ReadPipeAsync(PipeReader reader)
48 {
49     while (true)
50     {
51         ReadResult result = await reader.ReadAsync();
52         ReadOnlySequence<byte> buffer = result.Buffer;
53 
54         while (TryReadLine(ref buffer, out ReadOnlySequence<byte> line))
55         {
56             // Process the line.
57             ProcessLine(line);
58         }
59 
60         // Tell the PipeReader how much of the buffer has been consumed.
61         reader.AdvanceTo(buffer.Start, buffer.End);
62 
63         // Stop reading if there‘s no more data coming.
64         if (result.IsCompleted)
65         {
66             break;
67         }
68     }
69 
70     // Mark the PipeReader as complete.
71     await reader.CompleteAsync();
72 }
73 
74 bool TryReadLine(ref ReadOnlySequence<byte> buffer, out ReadOnlySequence<byte> line)
75 {
76     // Look for a EOL in the buffer.
77     SequencePosition? position = buffer.PositionOf((byte)‘\n‘);
78 
79     if (position == null)
80     {
81         line = default;
82         return false;
83     }
84 
85     // Skip the line + the \n.
86     line = buffer.Slice(0, position.Value);
87     buffer = buffer.Slice(buffer.GetPosition(1, position.Value));
88     return true;
89 }