Add an overload for BrowseAsync that accepts a delegate for handling each directory

src/ISoulseekClient.cs

@@ -416,6 +416,8 @@ public interface ISoulseekClient : IDisposable, IDiagnosticGenerator
         /// <exception cref="SoulseekClientException">Thrown when an exception is encountered during the operation.</exception>
         Task<BrowseResponse> BrowseAsync(string username, BrowseOptions options = null, CancellationToken? cancellationToken = null);
 
+        Task<BrowseResponse> BrowseAsync(string username, Action<Directory, bool> directoryHandler, BrowseOptions options = null, CancellationToken? cancellationToken = null);
+
         /// <summary>
         ///     Asynchronously changes the password for the currently logged in user.
         /// </summary>

src/Network/IMessageConnection.cs

@@ -24,6 +24,7 @@
 namespace Soulseek.Network
 {
     using System;
+    using System.IO;
     using System.Threading;
     using System.Threading.Tasks;
     using Soulseek.Messaging.Messages;
@@ -84,6 +85,21 @@ internal interface IMessageConnection : IConnection
         /// </summary>
         string Username { get; }
 
+        /// <summary>
+        ///     Registers an override for handling of the specified <paramref name="messageCode"/>, which will divert the
+        ///     received data packets to the specified <paramref name="stream"/> instead of the attached message handler,
+        ///     and will invoke the specified <paramref name="callback"/> when the message has been fully recieved.
+        /// </summary>
+        /// <remarks>
+        ///     Registrations are added to a FIFO queue internally, and messages will be streamed to handlers in the order
+        ///     they are registered and received. There is no way to guarantee that the remote client will respond in
+        ///     chronological order, so avoid using this for messages that are variable in this way (e.g. search responses).
+        /// </remarks>
+        /// <param name="messageCode">The message code of the message for which to override handling.</param>
+        /// <param name="stream">The stream to write the message data to.</param>
+        /// <param name="callback">The callback to invoke when the message has been fully received.</param>
+        void RegisterMessageHandlingOverride(int messageCode, Stream stream, Action callback);
+
         /// <summary>
         ///     Begins the internal continuous read loop, if it has not yet started.
         /// </summary>

src/Network/MessageConnection.cs

@@ -24,10 +24,13 @@
 namespace Soulseek.Network
 {
     using System;
+    using System.Collections.Concurrent;
     using System.Collections.Generic;
+    using System.IO;
     using System.Net;
     using System.Threading;
     using System.Threading.Tasks;
+    using Soulseek.Messaging;
     using Soulseek.Messaging.Messages;
     using Soulseek.Network.Tcp;
 
@@ -140,6 +143,33 @@ internal MessageConnection(IPEndPoint ipEndPoint, ConnectionOptions options = nu
         /// </summary>
         public string Username { get; } = string.Empty;
 
+        private ConcurrentDictionary<int, ConcurrentQueue<(Stream Stream, Action Callback)>> MessageHandlingOverrideRegistrations { get; } = new ConcurrentDictionary<int, ConcurrentQueue<(Stream Stream, Action Callback)>>();
+
+        /// <summary>
+        ///     Registers an override for handling of the specified <paramref name="messageCode"/>, which will divert the
+        ///     received data packets to the specified <paramref name="stream"/> instead of the attached message handler,
+        ///     and will invoke the specified <paramref name="callback"/> when the message has been fully recieved.
+        /// </summary>
+        /// <remarks>
+        ///     Registrations are added to a FIFO queue internally, and messages will be streamed to handlers in the order
+        ///     they are registered and received. There is no way to guarantee that the remote client will respond in
+        ///     chronological order, so avoid using this for messages that are variable in this way (e.g. search responses).
+        /// </remarks>
+        /// <param name="messageCode">The message code of the message for which to override handling.</param>
+        /// <param name="stream">The stream to write the message data to.</param>
+        /// <param name="callback">The callback to invoke when the message has been fully received.</param>
+        public void RegisterMessageHandlingOverride(int messageCode, Stream stream, Action callback)
+        {
+            MessageHandlingOverrideRegistrations.AddOrUpdate(
+                key: messageCode,
+                addValue: new ConcurrentQueue<(Stream Stream, Action Callback)>(new[] { (stream, callback) }),
+                updateValueFactory: (k, v) =>
+                {
+                    v.Enqueue((stream, callback));
+                    return v;
+                });
+        }
+
         /// <summary>
         ///     Begins the internal continuous read loop, if it has not yet started.
         /// </summary>
@@ -236,23 +266,36 @@ void RaiseMessageDataRead(object sender, ConnectionDataEventArgs e)
 
                         DataRead += RaiseMessageDataRead;
 
-                        var payloadBytes = await ReadAsync(length - CodeLength, CancellationToken.None).ConfigureAwait(false);
-                        message.AddRange(payloadBytes);
-
-                        var messageBytes = message.ToArray();
-
-                        if (SoulseekClient.RaiseEventsAsynchronously)
+                        // if a message stream 'hook' has been installed via InstallMessageStreamHook, stream the remainder
+                        // of the message to the provided stream. the caller will be notified that the read is complete
+                        // via MessageRead -> PeerMessageHandler.HandleMessageRead -> regular message handling
+                        // the caller must avoid trying to use the browse response, since it would have been streamed instead of passed
+                        if (BitConverter.ToInt32(codeBytes) == (int)MessageCode.Peer.BrowseResponse
+                            && MessageHandlingOverrideRegistrations.TryGetValue((int)MessageCode.Peer.BrowseResponse, out var queue)
+                            && queue.TryDequeue(out var entry))
                         {
-                            Task.Run(() =>
-                            {
-                                Interlocked.CompareExchange(ref MessageRead, null, null)?
-                                    .Invoke(this, new MessageEventArgs(messageBytes));
-                            }, CancellationToken.None).Forget();
+                            await ReadAsync(length - CodeLength, entry.Stream, cancellationToken: CancellationToken.None).ConfigureAwait(false);
+                            entry.Callback();
                         }
                         else
                         {
-                            Interlocked.CompareExchange(ref MessageRead, null, null)?
-                                .Invoke(this, new MessageEventArgs(messageBytes));
+                            var payloadBytes = await ReadAsync(length - CodeLength, CancellationToken.None).ConfigureAwait(false);
+                            message.AddRange(payloadBytes);
+                            var messageBytes = message.ToArray();
+
+                            if (SoulseekClient.RaiseEventsAsynchronously)
+                            {
+                                Task.Run(() =>
+                                {
+                                    Interlocked.CompareExchange(ref MessageRead, null, null)?
+                                        .Invoke(this, new MessageEventArgs(messageBytes));
+                                }, CancellationToken.None).Forget();
+                            }
+                            else
+                            {
+                                Interlocked.CompareExchange(ref MessageRead, null, null)?
+                                    .Invoke(this, new MessageEventArgs(messageBytes));
+                            }
                         }
                     }
                     finally

src/Network/Tcp/Connection.cs

@@ -418,7 +418,7 @@ public Task<byte[]> ReadAsync(long length, CancellationToken? cancellationToken
         ///     is not connected.
         /// </exception>
         /// <exception cref="ConnectionReadException">Thrown when an unexpected error occurs.</exception>
-        public Task ReadAsync(long length, Stream outputStream, Func<int, CancellationToken, Task<int>> governor, Action<int, int, int> reporter = null, CancellationToken? cancellationToken = null)
+        public Task ReadAsync(long length, Stream outputStream, Func<int, CancellationToken, Task<int>> governor = null, Action<int, int, int> reporter = null, CancellationToken? cancellationToken = null)
         {
             if (length < 0)
             {

src/Network/Tcp/IConnection.cs

@@ -165,7 +165,7 @@ internal interface IConnection : IDisposable
         ///     is not connected.
         /// </exception>
         /// <exception cref="ConnectionReadException">Thrown when an unexpected error occurs.</exception>
-        Task ReadAsync(long length, Stream outputStream, Func<int, CancellationToken, Task<int>> governor, Action<int, int, int> reporter = null, CancellationToken? cancellationToken = null);
+        Task ReadAsync(long length, Stream outputStream, Func<int, CancellationToken, Task<int>> governor = null, Action<int, int, int> reporter = null, CancellationToken? cancellationToken = null);
 
         /// <summary>
         ///     Waits for the connection to disconnect, returning the message or throwing the Exception which caused the disconnect.

src/SoulseekClient.cs

@@ -777,6 +777,11 @@ public Task<BrowseResponse> BrowseAsync(string username, BrowseOptions options =
             return BrowseInternalAsync(username, options, cancellationToken ?? CancellationToken.None);
         }
 
+        public async Task BrowseAsync(string username, Action<Directory, bool> directoryHandler, BrowseOptions options = null, CancellationToken cancellationToken = default)
+        {
+            // todo: implement me!
+        }
+
         /// <summary>
         ///     Asynchronously changes the password for the currently logged in user.
         /// </summary>