Implement QUIC abort exceptions, fix channel usage, and implement some tests.

src/libraries/System.Net.Quic/ref/System.Net.Quic.cs

@@ -95,4 +95,8 @@ public class QuicStreamAbortedException : QuicException
         public QuicStreamAbortedException(string message, long errorCode) : base(message) { }
         public long ErrorCode { get; }
     }
+    public class QuicOperationAbortedException : QuicException
+    {
+        public QuicOperationAbortedException(string message) : base(message) { }
+    }
 }

src/libraries/System.Net.Quic/src/Interop/MsQuicEnums.cs

@@ -55,7 +55,7 @@ internal enum QUIC_STREAM_SHUTDOWN_FLAG : uint
     }
 
     [Flags]
-    internal enum QUIC_RECEIVE_FLAG : byte
+    internal enum QUIC_RECEIVE_FLAG : uint
     {
         NONE = 0,
         ZERO_RTT = 0x1,
@@ -133,12 +133,12 @@ internal enum QUIC_PARAM_STREAM : uint
         IDEAL_SEND_BUFFER = 2
     }
 
-    internal enum QUIC_LISTENER_EVENT : byte
+    internal enum QUIC_LISTENER_EVENT : uint
     {
         NEW_CONNECTION = 0
     }
 
-    internal enum QUIC_CONNECTION_EVENT : byte
+    internal enum QUIC_CONNECTION_EVENT : uint
     {
         CONNECTED = 0,
         SHUTDOWN_INITIATED_BY_TRANSPORT = 1,
@@ -152,7 +152,7 @@ internal enum QUIC_CONNECTION_EVENT : byte
         IDEAL_PROCESSOR_CHANGED = 9,
     }
 
-    internal enum QUIC_STREAM_EVENT : byte
+    internal enum QUIC_STREAM_EVENT : uint
     {
         START_COMPLETE = 0,
         RECEIVE = 1,

src/libraries/System.Net.Quic/src/Interop/MsQuicNativeMethods.cs

@@ -183,7 +183,7 @@ internal struct ConnectionEventDataShutdownBegin
         [StructLayout(LayoutKind.Sequential)]
         internal struct ConnectionEventDataShutdownBeginPeer
         {
-            internal ushort ErrorCode;
+            internal long ErrorCode;
         }
 
         [StructLayout(LayoutKind.Sequential)]
@@ -264,7 +264,7 @@ internal struct ConnectionEvent
             internal bool EarlyDataAccepted => Data.Connected.EarlyDataAccepted;
             internal ulong NumBytes => Data.IdealSendBuffer.NumBytes;
             internal uint ShutdownBeginStatus => Data.ShutdownBegin.Status;
-            internal ushort ShutdownBeginPeerStatus => Data.ShutdownBeginPeer.ErrorCode;
+            internal long ShutdownBeginPeerStatus => Data.ShutdownBeginPeer.ErrorCode;
             internal bool ShutdownTimedOut => Data.ShutdownComplete.TimedOut;
             internal ushort BiDirectionalCount => Data.StreamsAvailable.BiDirectionalCount;
             internal ushort UniDirectionalCount => Data.StreamsAvailable.UniDirectionalCount;
@@ -324,13 +324,13 @@ internal bool IsCanceled()
         [StructLayout(LayoutKind.Sequential)]
         internal struct StreamEventDataPeerSendAbort
         {
-            internal ushort ErrorCode;
+            internal long ErrorCode;
         }
 
         [StructLayout(LayoutKind.Sequential)]
         internal struct StreamEventDataPeerRecvAbort
         {
-            internal ushort ErrorCode;
+            internal long ErrorCode;
         }
 
         [StructLayout(LayoutKind.Sequential)]

src/libraries/System.Net.Quic/src/Resources/Strings.resx

@@ -117,10 +117,16 @@
   <resheader name="writer">
     <value>System.Resources.ResXResourceWriter, System.Windows.Forms, Version=4.0.0.0, Culture=neutral, PublicKeyToken=b77a5c561934e089</value>
   </resheader>
+  <data name="net_quic_connectionaborted" xml:space="preserve">
+    <value>Connection aborted by peer ({0}).</value>
+  </data>
   <data name="net_quic_notsupported" xml:space="preserve">
     <value>QUIC is not supported on this platform. See http://aka.ms/dotnetquic</value>
   </data>
-  <data name="net_quic_placeholdertext" xml:space="preserve">
-    <value>Placeholder text</value>
+  <data name="net_quic_operationaborted" xml:space="preserve">
+    <value>Operation aborted.</value>
+  </data>
+  <data name="net_quic_streamaborted" xml:space="preserve">
+    <value>Stream aborted by peer ({0}).</value>
   </data>
 </root>

src/libraries/System.Net.Quic/src/System.Net.Quic.csproj

@@ -32,6 +32,7 @@
     <Compile Include="System\Net\Quic\QuicConnection.cs" />
     <Compile Include="System\Net\Quic\QuicListener.cs" />
     <Compile Include="System\Net\Quic\QuicListenerOptions.cs" />
+    <Compile Include="System\Net\Quic\QuicOperationAbortedException.cs" />
     <Compile Include="System\Net\Quic\QuicStream.cs" />
     <Compile Include="System\Net\Quic\QuicException.cs" />
     <Compile Include="System\Net\Quic\QuicConnectionAbortedException.cs" />

src/libraries/System.Net.Quic/src/System/Net/Quic/Implementations/MsQuic/MsQuicConnection.cs

@@ -39,6 +39,7 @@ internal sealed class MsQuicConnection : QuicConnectionProvider
         private bool _disposed;
         private bool _connected;
         private MsQuicSecurityConfig _securityConfig;
+        private long _abortErrorCode = -1;
 
         // Queue for accepted streams
         private readonly Channel<MsQuicStream> _acceptQueue = Channel.CreateUnbounded<MsQuicStream>(new UnboundedChannelOptions()
@@ -200,6 +201,7 @@ private uint HandleEventShutdownInitiatedByTransport(ConnectionEvent connectionE
 
         private uint HandleEventShutdownInitiatedByPeer(ConnectionEvent connectionEvent)
         {
+            _abortErrorCode = connectionEvent.Data.ShutdownBeginPeer.ErrorCode;
             _acceptQueue.Writer.Complete();
             return MsQuicStatusCodes.Success;
         }
@@ -237,18 +239,23 @@ internal override async ValueTask<QuicStreamProvider> AcceptStreamAsync(Cancella
 
             ThrowIfDisposed();
 
-            if (await _acceptQueue.Reader.WaitToReadAsync(cancellationToken))
+            MsQuicStream stream;
+
+            try
+            {
+                stream = await _acceptQueue.Reader.ReadAsync(cancellationToken).ConfigureAwait(false);
+            }
+            catch (ChannelClosedException)
             {
-                if (_acceptQueue.Reader.TryRead(out MsQuicStream stream))
+                throw _abortErrorCode switch
                 {
-                    if (NetEventSource.IsEnabled) NetEventSource.Exit(this);
-                    return stream;
-                }
+                    -1 => new QuicOperationAbortedException(), // Shutdown initiated by us.
+                    long err => new QuicConnectionAbortedException(err) // Shutdown initiated by peer.
+                };
             }
 
             if (NetEventSource.IsEnabled) NetEventSource.Exit(this);
-
-            return null;
+            return stream;
         }
 
         internal override QuicStreamProvider OpenUnidirectionalStream()

src/libraries/System.Net.Quic/src/System/Net/Quic/Implementations/MsQuic/MsQuicListener.cs

@@ -12,7 +12,7 @@
 
 namespace System.Net.Quic.Implementations.MsQuic
 {
-    internal class MsQuicListener : QuicListenerProvider, IDisposable
+    internal sealed class MsQuicListener : QuicListenerProvider, IDisposable
     {
         // Security configuration for MsQuic
         private MsQuicSession _session;
@@ -65,21 +65,21 @@ internal override async ValueTask<QuicConnectionProvider> AcceptConnectionAsync(
 
             ThrowIfDisposed();
 
-            if (await _acceptConnectionQueue.Reader.WaitToReadAsync())
-            {
-                if (_acceptConnectionQueue.Reader.TryRead(out MsQuicConnection connection))
-                {
-                    // resolve security config here.
-                    await connection.SetSecurityConfigForConnection(_sslOptions.ServerCertificate);
-                    if (NetEventSource.IsEnabled) NetEventSource.Exit(this);
+            MsQuicConnection connection;
 
-                    return connection;
-                }
+            try
+            {
+                connection = await _acceptConnectionQueue.Reader.ReadAsync(cancellationToken).ConfigureAwait(false);
+            }
+            catch (ChannelClosedException)
+            {
+                throw new QuicOperationAbortedException();
             }
 
-            if (NetEventSource.IsEnabled) NetEventSource.Exit(this);
+            await connection.SetSecurityConfigForConnection(_sslOptions.ServerCertificate);
 
-            return null;
+            if (NetEventSource.IsEnabled) NetEventSource.Exit(this);
+            return connection;
         }
 
         public override void Dispose()
@@ -174,7 +174,7 @@ internal unsafe uint ListenerCallbackHandler(
             }
         }
 
-        protected void StopAcceptingConnections()
+        private void StopAcceptingConnections()
         {
             _acceptConnectionQueue.Writer.TryComplete();
         }

src/libraries/System.Net.Quic/src/System/Net/Quic/Implementations/MsQuic/MsQuicStream.cs

@@ -48,10 +48,12 @@ internal sealed class MsQuicStream : QuicStreamProvider
         private StartState _started;
 
         private ReadState _readState;
+        private long _readErrorCode = -1;
 
         private ShutdownWriteState _shutdownState;
 
         private SendState _sendState;
+        private long _sendErrorCode = -1;
 
         // Used by the class to indicate that the stream is m_Readable.
         private readonly bool _canRead;
@@ -245,7 +247,11 @@ internal override async ValueTask<int> ReadAsync(Memory<byte> destination, Cance
                 }
                 else if (_readState == ReadState.Aborted)
                 {
-                    throw new IOException("Reading has been aborted by the peer.");
+                    throw _readErrorCode switch
+                    {
+                        -1 => new QuicOperationAbortedException(),
+                        long err => new QuicStreamAbortedException(err)
+                    };
                 }
             }
 
@@ -402,6 +408,7 @@ internal override void Write(ReadOnlySpan<byte> buffer)
         {
             ThrowIfDisposed();
 
+            // TODO: optimize this.
             WriteAsync(buffer.ToArray()).GetAwaiter().GetResult();
         }
 
@@ -532,14 +539,13 @@ private uint HandleEvent(ref StreamEvent evt)
                     // Peer has told us to abort the reading side of the stream.
                     case QUIC_STREAM_EVENT.PEER_SEND_ABORTED:
                         {
-                            status = HandleEventPeerSendAborted();
+                            status = HandleEventPeerSendAborted(ref evt);
                         }
                         break;
                     // Peer has stopped receiving data, don't send anymore.
-                    // Potentially throw when WriteAsync/FlushAsync.
                     case QUIC_STREAM_EVENT.PEER_RECEIVE_ABORTED:
                         {
-                            status = HandleEventPeerRecvAbort();
+                            status = HandleEventPeerRecvAborted(ref evt);
                         }
                         break;
                     // Occurs when shutdown is completed for the send side.
@@ -598,9 +604,26 @@ private unsafe uint HandleEventRecv(ref MsQuicNativeMethods.StreamEvent evt)
             return MsQuicStatusCodes.Pending;
         }
 
-        private uint HandleEventPeerRecvAbort()
+        private uint HandleEventPeerRecvAborted(ref StreamEvent evt)
         {
             if (NetEventSource.IsEnabled) NetEventSource.Enter(this);
+
+            bool shouldComplete = false;
+            lock (_sync)
+            {
+                if (_sendState == SendState.None)
+                {
+                    shouldComplete = true;
+                }
+                _sendState = SendState.Aborted;
+                _sendErrorCode = evt.Data.PeerSendAbort.ErrorCode;
+            }
+
+            if (shouldComplete)
+            {
+                _sendResettableCompletionSource.CompleteException(new QuicStreamAbortedException(_sendErrorCode));
+            }
+
             if (NetEventSource.IsEnabled) NetEventSource.Exit(this);
 
             return MsQuicStatusCodes.Success;
@@ -696,7 +719,7 @@ private uint HandleEventShutdownComplete()
             return MsQuicStatusCodes.Success;
         }
 
-        private uint HandleEventPeerSendAborted()
+        private uint HandleEventPeerSendAborted(ref StreamEvent evt)
         {
             if (NetEventSource.IsEnabled) NetEventSource.Enter(this);
 
@@ -708,11 +731,12 @@ private uint HandleEventPeerSendAborted()
                     shouldComplete = true;
                 }
                 _readState = ReadState.Aborted;
+                _readErrorCode = evt.Data.PeerSendAbort.ErrorCode;
             }
 
             if (shouldComplete)
             {
-                _receiveResettableCompletionSource.CompleteException(new IOException("Reading has been aborted by the peer."));
+                _receiveResettableCompletionSource.CompleteException(new QuicStreamAbortedException(_readErrorCode));
             }
 
             if (NetEventSource.IsEnabled) NetEventSource.Exit(this);

src/libraries/System.Net.Quic/src/System/Net/Quic/QuicConnectionAbortedException.cs

@@ -6,6 +6,11 @@ namespace System.Net.Quic
 {
     public class QuicConnectionAbortedException : QuicException
     {
+        internal QuicConnectionAbortedException(long errorCode)
+            : this(SR.Format(SR.net_quic_connectionaborted, errorCode), errorCode)
+        {
+        }
+
         public QuicConnectionAbortedException(string message, long errorCode)
             : base (message)
         {

src/libraries/System.Net.Quic/src/System/Net/Quic/QuicOperationAbortedException.cs

@@ -0,0 +1,18 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+namespace System.Net.Quic
+{
+    public class QuicOperationAbortedException : QuicException
+    {
+        internal QuicOperationAbortedException()
+            : base(SR.net_quic_operationaborted)
+        {
+        }
+
+        public QuicOperationAbortedException(string message) : base(message)
+        {
+        }
+    }
+}

src/libraries/System.Net.Quic/src/System/Net/Quic/QuicStreamAbortedException.cs

@@ -6,8 +6,13 @@ namespace System.Net.Quic
 {
     public class QuicStreamAbortedException : QuicException
     {
+        internal QuicStreamAbortedException(long errorCode)
+            : this(SR.Format(SR.net_quic_streamaborted, errorCode), errorCode)
+        {
+        }
+
         public QuicStreamAbortedException(string message, long errorCode)
-            : base (message)
+            : base(message)
         {
             ErrorCode = errorCode;
         }