Add cancel and complete

src/Bedrock.Framework/Protocols/ProtocolReader.cs

@@ -241,6 +241,27 @@ public void Advance()
             _hasMessage = false;
         }
 
+        public void CancelPendingRead()
+        {
+            if (_disposed)
+            {
+                throw new ObjectDisposedException(GetType().Name);
+            }
+
+            _reader.CancelPendingRead();
+            _isCanceled = true;
+        }
+
+        public async ValueTask CompleteAsync()
+        {
+            if (_disposed)
+            {
+                throw new ObjectDisposedException(GetType().Name);
+            }
+
+            await _reader.CompleteAsync();
+        }
+
         public ValueTask DisposeAsync()
         {
             _disposed = true;

src/Bedrock.Framework/Protocols/ProtocolWriter.cs

@@ -95,6 +95,26 @@ public async ValueTask WriteManyAsync<TWriteMessage>(IMessageWriter<TWriteMessag
             }
         }
 
+        public void CancelPendingFlush()
+        {
+            if (_disposed)
+            {
+                return;
+            }
+
+            _writer.CancelPendingFlush();
+        }
+
+        public async ValueTask CompleteAsync()
+        {
+            if (_disposed)
+            {
+                return;
+            }
+
+            await _writer.CompleteAsync();
+        }
+
         public async ValueTask DisposeAsync()
         {
             await _semaphore.WaitAsync().ConfigureAwait(false);