Merge branch 'pipelines' into server-additions

# Conflicts:
#	StackExchange.Redis.Server/RedisServer.cs
#	StackExchange.Redis.Server/RespServer.cs

KestrelRedisServer/RedisConnectionHandler.cs

@@ -9,6 +9,6 @@ public class RedisConnectionHandler : ConnectionHandler
         private readonly RespServer _server;
         public RedisConnectionHandler(RespServer server) => _server = server;
         public override Task OnConnectedAsync(ConnectionContext connection)
-            => _server.RunClient(connection.Transport);
+            => _server.RunClientAsync(connection.Transport);
     }
 }

KestrelRedisServer/Startup.cs

@@ -19,8 +19,20 @@ public void ConfigureServices(IServiceCollection services)
         public void Dispose() => _server.Dispose();
 
         // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
-        public void Configure(IApplicationBuilder app, IHostingEnvironment env)
+        public void Configure(IApplicationBuilder app, IHostingEnvironment env, IApplicationLifetime lifetime)
         {
+            _server.Shutdown.ContinueWith((t, s) =>
+            {
+                try
+                {   // if the resp server is shutdown by a client: stop the kestrel server too
+                    if (t.Result == RespServer.ShutdownReason.ClientInitiated)
+                    {
+                        ((IApplicationLifetime)s).StopApplication();
+                    }
+                }
+                catch { }
+            }, lifetime);
+
             if (env.IsDevelopment()) app.UseDeveloperExceptionPage();
             app.Run(context => context.Response.WriteAsync(_server.GetStats()));
         }

NRediSearch.Test/ClientTests/AggregationTest.cs

@@ -10,7 +10,7 @@ public class AggregationTest : RediSearchTestBase
         public AggregationTest(ITestOutputHelper output) : base(output) { }
 
         [Fact]
-        public void testAggregations()
+        public void TestAggregations()
         {
             /**
              127.0.0.1:6379> FT.CREATE test_index SCHEMA name TEXT SORTABLE count NUMERIC SORTABLE
@@ -35,7 +35,6 @@ public void testAggregations()
                     .GroupBy("@name", Reducers.Sum("@count").As("sum"))
                 .SortBy(SortedField.Descending("@sum"), 10);
 
-
             // actual search
             AggregationResult res = cl.Aggregate(r);
             var r1 = res.GetRow(0);

StackExchange.Redis.Server/RedisServer.cs

@@ -277,7 +277,7 @@ protected virtual TypedRedisValue Set(RedisClient client, RedisRequest request)
         [RedisCommand(1)]
         protected new virtual TypedRedisValue Shutdown(RedisClient client, RedisRequest request)
         {
-            DoShutdown();
+            DoShutdown(ShutdownReason.ClientInitiated);
             return TypedRedisValue.OK;
         }
         [RedisCommand(2)]
@@ -378,8 +378,8 @@ StringBuilder AddHeader()
                     {
                         sb.Append("process:").Append(process.Id).AppendLine();
                     }
-                    var port = TcpPort();
-                    if (port >= 0) sb.Append("tcp_port:").Append(port).AppendLine();
+                    //var port = TcpPort();
+                    //if (port >= 0) sb.Append("tcp_port:").Append(port).AppendLine();
                     break;
                 case "Clients":
                     AddHeader().Append("connected_clients:").Append(ClientCount).AppendLine();

StackExchange.Redis.Server/RespServer.cs

@@ -5,8 +5,6 @@
 using System.IO;
 using System.IO.Pipelines;
 using System.Linq;
-using System.Net;
-using System.Net.Sockets;
 using System.Reflection;
 using System.Text;
 using System.Threading;
@@ -17,9 +15,14 @@ namespace StackExchange.Redis.Server
 {
     public abstract partial class RespServer : IDisposable
     {
+        public enum ShutdownReason
+        {
+            ServerDisposed,
+            ClientInitiated,
+        }
         private readonly List<RedisClient> _clients = new List<RedisClient>();
         private readonly TextWriter _output;
-        private Socket _listener;
+        
         public RespServer(TextWriter output = null)
         {
             _output = output;
@@ -175,44 +178,9 @@ internal int NetArity()
                 return varadic ? -minMagnitude : minMagnitude;
             }
         }
-        delegate TypedRedisValue RespOperation(RedisClient client, RedisRequest request);
-
-        protected int TcpPort()
-        {
-            var ep = _listener?.LocalEndPoint;
-            if (ep is IPEndPoint ip) return ip.Port;
-            if (ep is DnsEndPoint dns) return dns.Port;
-            return -1;
-        }
 
-        private Action<object> _runClientCallback;
-        // KeepAlive here just to make the compiler happy that we've done *something* with the task
-        private Action<object> RunClientCallback => _runClientCallback ??
-            (_runClientCallback = state => GC.KeepAlive(RunClient((IDuplexPipe)state)));
-
-        public void Listen(
-            EndPoint endpoint,
-            AddressFamily addressFamily = AddressFamily.InterNetwork,
-            SocketType socketType = SocketType.Stream,
-            ProtocolType protocolType = ProtocolType.Tcp,
-            PipeOptions sendOptions = null, PipeOptions receiveOptions = null)
-        {
-            Socket listener = new Socket(addressFamily, socketType, protocolType);
-            listener.Bind(endpoint);
-            listener.Listen(20);
-
-            _listener = listener;
-            StartOnScheduler(receiveOptions?.ReaderScheduler, _ => ListenForConnections(
-                sendOptions ?? PipeOptions.Default, receiveOptions ?? PipeOptions.Default), null);
-
-            Log("Server is listening on " + Format.ToString(endpoint));
-        }
+        delegate TypedRedisValue RespOperation(RedisClient client, RedisRequest request);
         
-        private static void StartOnScheduler(PipeScheduler scheduler, Action<object> callback, object state)
-        {
-            if (scheduler == PipeScheduler.Inline) scheduler = null;
-            (scheduler ?? PipeScheduler.ThreadPool).Schedule(callback, state);
-        }
         // for extensibility, so that a subclass can get their own client type
         // to be used via ListenForConnections
         public virtual RedisClient CreateClient() => new RedisClient();
@@ -244,33 +212,14 @@ public bool RemoveClient(RedisClient client)
                 return _clients.Remove(client);
             }
         }
-        private async void ListenForConnections(PipeOptions sendOptions, PipeOptions receiveOptions)
-        {
-            try
-            {
-                while (true)
-                {
-                    var client = await _listener.AcceptAsync();
-                    SocketConnection.SetRecommendedServerOptions(client);
-                    var pipe = SocketConnection.Create(client, sendOptions, receiveOptions);
-                    StartOnScheduler(receiveOptions.ReaderScheduler, RunClientCallback, pipe);
-                }
-            }
-            catch (NullReferenceException) { }
-            catch (ObjectDisposedException) { }
-            catch (Exception ex)
-            {
-                if (!_isShutdown) Log("Listener faulted: " + ex.Message);
-            }
-        }
 
-        private readonly TaskCompletionSource<int> _shutdown = new TaskCompletionSource<int>();
+        private readonly TaskCompletionSource<ShutdownReason> _shutdown = new TaskCompletionSource<ShutdownReason>(TaskCreationOptions.RunContinuationsAsynchronously);
         private bool _isShutdown;
         protected void ThrowIfShutdown()
         {
             if (_isShutdown) throw new InvalidOperationException("The server is shutting down");
         }
-        protected void DoShutdown(PipeScheduler scheduler = null)
+        protected void DoShutdown(ShutdownReason reason, PipeScheduler scheduler = null)
         {
             if (_isShutdown) return;
             Log("Server shutting down...");
@@ -280,22 +229,16 @@ protected void DoShutdown(PipeScheduler scheduler = null)
                 foreach (var client in _clients) client.Dispose();
                 _clients.Clear();
             }
-            StartOnScheduler(scheduler,
-                state => ((TaskCompletionSource<int>)state).TrySetResult(0), _shutdown);
+            _shutdown.TrySetResult(reason);
         }
-        public Task Shutdown => _shutdown.Task;
+        public Task<ShutdownReason> Shutdown => _shutdown.Task;
         public void Dispose() => Dispose(true);
         protected virtual void Dispose(bool disposing)
         {
-            DoShutdown();
-            var socket = _listener;
-            if (socket != null)
-            {
-                try { socket.Dispose(); } catch { }
-            }
+            DoShutdown(ShutdownReason.ServerDisposed);
         }
 
-        public async Task RunClient(IDuplexPipe pipe)
+        public async Task RunClientAsync(IDuplexPipe pipe)
         {
             Exception fault = null;
             RedisClient client = null;
@@ -335,7 +278,7 @@ public async Task RunClient(IDuplexPipe pipe)
                 }
             }
         }
-        private void Log(string message)
+        public void Log(string message)
         {
             var output = _output;
             if (output != null)

StackExchange.Redis.Server/RespSocketServer.cs

+using System;
+using System.Net;
+using System.Threading.Tasks;
+using Pipelines.Sockets.Unofficial;
+
+namespace StackExchange.Redis.Server
+{
+    public sealed class RespSocketServer : SocketServer
+    {
+        private readonly RespServer _server;
+        public RespSocketServer(RespServer server)
+        {
+            _server = server ?? throw new ArgumentNullException(nameof(server));
+            server.Shutdown.ContinueWith((t, o) => ((SocketServer)o).Dispose(), this);
+        }
+        protected override void OnStarted(EndPoint endPoint)
+            => _server.Log("Server is listening on " + endPoint);
+
+        protected override Task OnClientConnectedAsync(in ClientConnection client)
+            => _server.RunClientAsync(client.Transport);
+
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing) _server.Dispose();
+        }
+    }
+}

StackExchange.Redis.Tests/Constraints.cs

@@ -26,8 +26,8 @@ public async Task TestManualIncr()
                 for (int i = 0; i < 10; i++)
                 {
                     conn.KeyDelete(key, CommandFlags.FireAndForget);
-                    Assert.Equal(1, await ManualIncrAsync(conn, key));
-                    Assert.Equal(2, await ManualIncrAsync(conn, key));
+                    Assert.Equal(1, await ManualIncrAsync(conn, key).ForAwait());
+                    Assert.Equal(2, await ManualIncrAsync(conn, key).ForAwait());
                     Assert.Equal(2, (long)conn.StringGet(key));
                 }
             }

StackExchange.Redis.Tests/Execute.cs

@@ -21,7 +21,7 @@ public async Task DBExecute()
                 var actual = (string)db.Execute("GET", key);
                 Assert.Equal("some value", actual);
 
-                actual = (string)await db.ExecuteAsync("GET", key);
+                actual = (string)await db.ExecuteAsync("GET", key).ForAwait();
                 Assert.Equal("some value", actual);
             }
         }
@@ -35,7 +35,7 @@ public async Task ServerExecute()
                 var actual = (string)server.Execute("echo", "some value");
                 Assert.Equal("some value", actual);
 
-                actual = (string)await server.ExecuteAsync("echo", "some value");
+                actual = (string)await server.ExecuteAsync("echo", "some value").ForAwait();
                 Assert.Equal("some value", actual);
             }
         }

StackExchange.Redis.Tests/MassiveOps.cs

@@ -12,6 +12,24 @@ public class MassiveOps : TestBase
     {
         public MassiveOps(ITestOutputHelper output) : base(output) { }
 
+        [Fact]
+        public async Task LongRunning()
+        {
+            var key = Me();
+            using (var conn = Create())
+            {
+                var db = conn.GetDatabase();
+                db.KeyDelete(key, CommandFlags.FireAndForget);
+                db.StringSet(key, "test value", flags: CommandFlags.FireAndForget);
+                for (var i = 0; i < 200; i++)
+                {
+                    var val = await db.StringGetAsync(key).ForAwait();
+                    Assert.Equal("test value", (string)val);
+                    await Task.Delay(50).ForAwait();
+                }
+            }
+        }
+
         [Theory]
         [InlineData(true)]
         [InlineData(false)]

StackExchange.Redis.Tests/Profiling.cs

@@ -69,8 +69,6 @@ public void Simple()
                 AssertProfiledCommandValues(eval, conn, dbId);
 
                 AssertProfiledCommandValues(echo, conn, dbId);
-
-                
             }
         }
 
@@ -156,7 +154,6 @@ public void ManyContexts()
                 var prefix = Me();
                 conn.RegisterProfiler(profiler.GetSession);
 
-                
                 var threads = new List<Thread>();
 
                 var results = new IEnumerable<IProfiledCommand>[16];
@@ -204,13 +201,13 @@ public void ManyContexts()
 
         internal class PerThreadProfiler
         {
-            ThreadLocal<ProfilingSession> perThreadSession = new ThreadLocal<ProfilingSession>(() => new ProfilingSession());
+            private readonly ThreadLocal<ProfilingSession> perThreadSession = new ThreadLocal<ProfilingSession>(() => new ProfilingSession());
 
             public ProfilingSession GetSession() => perThreadSession.Value;
         }
         internal class AsyncLocalProfiler
         {
-            AsyncLocal<ProfilingSession> perThreadSession = new AsyncLocal<ProfilingSession>();
+            private readonly AsyncLocal<ProfilingSession> perThreadSession = new AsyncLocal<ProfilingSession>();
 
             public ProfilingSession GetSession()
             {
@@ -275,7 +272,6 @@ public void LowAllocationEnumerable()
             }
         }
 
-
         [FactLongRunning]
         public void ProfilingMD_Ex1()
         {
@@ -385,7 +381,7 @@ public async Task ProfilingMD_Ex2_Async()
                     {
                         for (var j = 0; j < 100; j++)
                         {
-                            await db.StringSetAsync(prefix + j, "" + j);
+                            await db.StringSetAsync(prefix + j, "" + j).ForAwait();
                         }
 
                         perThreadTimings.Add(profiler.GetSession().FinishProfiling().ToList());
@@ -396,7 +392,7 @@ public async Task ProfilingMD_Ex2_Async()
 
                 var timeout = Task.Delay(10000);
                 var complete = Task.WhenAll(tasks);
-                if (timeout == await Task.WhenAny(timeout, complete))
+                if (timeout == await Task.WhenAny(timeout, complete).ForAwait())
                 {
                     throw new TimeoutException();
                 }

StackExchange.Redis.Tests/SSL.cs

@@ -147,25 +147,25 @@ public async Task ConnectToSSLServer(bool useSsl, bool specifyHost)
             }
         }
 
-        private void TestConcurrent(IDatabase db, RedisKey key, int SyncLoop, int Threads)
-        {
-            long value;
-            db.KeyDelete(key, CommandFlags.FireAndForget);
-            var time = RunConcurrent(delegate
-            {
-                for (int i = 0; i < SyncLoop; i++)
-                {
-                    db.StringIncrement(key);
-                }
-            }, Threads, timeout: 45000);
-            value = (long)db.StringGet(key);
-            Assert.Equal(SyncLoop * Threads, value);
-            Log("Sync: {0} INCR using {1} threads, {2:###,##0}ms, {3} ops/s; final value: {4}",
-                SyncLoop * Threads, Threads,
-                (long)time.TotalMilliseconds,
-                (long)((SyncLoop * Threads) / time.TotalSeconds),
-                value);
-        }
+        //private void TestConcurrent(IDatabase db, RedisKey key, int SyncLoop, int Threads)
+        //{
+        //    long value;
+        //    db.KeyDelete(key, CommandFlags.FireAndForget);
+        //    var time = RunConcurrent(delegate
+        //    {
+        //        for (int i = 0; i < SyncLoop; i++)
+        //        {
+        //            db.StringIncrement(key);
+        //        }
+        //    }, Threads, timeout: 45000);
+        //    value = (long)db.StringGet(key);
+        //    Assert.Equal(SyncLoop * Threads, value);
+        //    Log("Sync: {0} INCR using {1} threads, {2:###,##0}ms, {3} ops/s; final value: {4}",
+        //        SyncLoop * Threads, Threads,
+        //        (long)time.TotalMilliseconds,
+        //        (long)((SyncLoop * Threads) / time.TotalSeconds),
+        //        value);
+        //}
 
         [Fact]
         public void RedisLabsSSL()

StackExchange.Redis.Tests/TestBase.cs

@@ -280,7 +280,7 @@ protected static TimeSpan RunConcurrent(Action work, int threads, int timeout =
             ManualResetEvent allDone = new ManualResetEvent(false);
             object token = new object();
             int active = 0;
-            ThreadStart callback = delegate
+            void callback()
             {
                 lock (token)
                 {
@@ -301,7 +301,7 @@ protected static TimeSpan RunConcurrent(Action work, int threads, int timeout =
                     watch.Stop();
                     allDone.Set();
                 }
-            };
+            }
 
             var threadArr = new Thread[threads];
             for (int i = 0; i < threads; i++)

StackExchange.Redis/StackExchange.Redis.csproj

@@ -23,7 +23,7 @@
   </ItemGroup>
   
   <ItemGroup>
-    <PackageReference Include="Pipelines.Sockets.Unofficial" Version="0.2.1-alpha.68" />
+    <PackageReference Include="Pipelines.Sockets.Unofficial" Version="0.2.1-alpha.72" />
     <PackageReference Include="System.Threading.Channels" Version="4.5.0" />
   </ItemGroup>
 </Project>
\ No newline at end of file

StackExchange.Redis/StackExchange/Redis/ChannelMessageQueue.cs

@@ -21,9 +21,11 @@ namespace StackExchange.Redis
         /// See Object.GetHashCode
         /// </summary>
         public override int GetHashCode() => Channel.GetHashCode() ^ Message.GetHashCode();
+
         /// <summary>
         /// See Object.Equals
         /// </summary>
+        /// <param name="obj">The <see cref="object"/> to compare.</param>
         public override bool Equals(object obj) => obj is ChannelMessage cm
             && cm.Channel == Channel && cm.Message == Message;
         internal ChannelMessage(ChannelMessageQueue queue, RedisChannel channel, RedisValue value)
@@ -48,7 +50,6 @@ internal ChannelMessage(ChannelMessageQueue queue, RedisChannel channel, RedisVa
         public RedisValue Message { get; }
     }
 
-
     /// <summary>
     /// Represents a message queue of ordered pub/sub notifications
     /// </summary>
@@ -80,7 +81,7 @@ internal ChannelMessageQueue(RedisChannel redisChannel, RedisSubscriber parent)
             _queue.Reader.Completion.ContinueWith(
                 (t, state) => ((ChannelMessageQueue)state).IsCompleted = true, this, TaskContinuationOptions.ExecuteSynchronously);
         }
-        static readonly UnboundedChannelOptions s_ChannelOptions = new UnboundedChannelOptions
+        private static readonly UnboundedChannelOptions s_ChannelOptions = new UnboundedChannelOptions
         {
             SingleWriter = true,
             SingleReader = false,
@@ -102,21 +103,23 @@ private void HandleMessage(RedisChannel channel, RedisValue value)
             }
         }
 
-
         /// <summary>
-        /// Consume a message from the channel
+        /// Consume a message from the channel.
         /// </summary>
+        /// <param name="cancellationToken">The <see cref="CancellationToken"/> to use.</param>
         public ValueTask<ChannelMessage> ReadAsync(CancellationToken cancellationToken = default)
             => _queue.Reader.ReadAsync(cancellationToken);
 
         /// <summary>
-        /// Attempt to synchronously consume a message from the channel
+        /// Attempt to synchronously consume a message from the channel.
         /// </summary>
+        /// <param name="item">The <see cref="ChannelMessage"/> read from the Channel.</param>
         public bool TryRead(out ChannelMessage item) => _queue.Reader.TryRead(out item);
 
         /// <summary>
-        /// Attempt to query the backlog length of the queue
+        /// Attempt to query the backlog length of the queue.
         /// </summary>
+        /// <param name="count">The (approximate) count of items in the Channel.</param>
         public bool TryGetCount(out int count)
         {
             // get this using the reflection
@@ -141,9 +144,11 @@ private void AssertOnMessage(Delegate handler)
             if (Interlocked.CompareExchange(ref _onMessageHandler, handler, null) != null)
                 throw new InvalidOperationException("Only a single " + nameof(OnMessage) + " is allowed");
         }
+
         /// <summary>
-        /// Create a message loop that processes messages sequentially
+        /// Create a message loop that processes messages sequentially.
         /// </summary>
+        /// <param name="handler">The handler to run when receiving a message.</param>
         public void OnMessage(Action<ChannelMessage> handler)
         {
             AssertOnMessage(handler);
@@ -170,8 +175,9 @@ private async void OnMessageSyncImpl()
         }
 
         /// <summary>
-        /// Create a message loop that processes messages sequentially
+        /// Create a message loop that processes messages sequentially.
         /// </summary>
+        /// <param name="handler">The handler to execute when receiving a message.</param>
         public void OnMessage(Func<ChannelMessage, Task> handler)
         {
             AssertOnMessage(handler);
@@ -226,7 +232,7 @@ internal static bool IsOneOf(Action<RedisChannel, RedisValue> handler)
         {
             try
             {
-                return handler != null && handler.Target is ChannelMessageQueue
+                return handler?.Target is ChannelMessageQueue
                     && handler.Method.Name == nameof(HandleMessage);
             }
             catch
@@ -236,12 +242,15 @@ internal static bool IsOneOf(Action<RedisChannel, RedisValue> handler)
         }
 
         /// <summary>
-        /// Stop receiving messages on this channel
+        /// Stop receiving messages on this channel.
         /// </summary>
+        /// <param name="flags">The flags to use when unsubscribing.</param>
         public void Unsubscribe(CommandFlags flags = CommandFlags.None) => UnsubscribeImpl(null, flags);
+
         /// <summary>
-        /// Stop receiving messages on this channel
+        /// Stop receiving messages on this channel.
         /// </summary>
+        /// <param name="flags">The flags to use when unsubscribing.</param>
         public Task UnsubscribeAsync(CommandFlags flags = CommandFlags.None) => UnsubscribeAsyncImpl(null, flags);
     }
 }

StackExchange.Redis/StackExchange/Redis/ConnectionMultiplexer.Profiling.cs

@@ -5,13 +5,14 @@ namespace StackExchange.Redis
 {
     public partial class ConnectionMultiplexer
     {
-        Func<ProfilingSession> _profilingSessionProvider;
+        private Func<ProfilingSession> _profilingSessionProvider;
 
         /// <summary>
         /// Register a callback to provide an on-demand ambient session provider based on the
         /// calling context; the implementing code is responsible for reliably resolving the same provider
         /// based on ambient context, or returning null to not profile
         /// </summary>
+        /// <param name="profilingSessionProvider">The session provider to register.</param>
         public void RegisterProfiler(Func<ProfilingSession> profilingSessionProvider) => _profilingSessionProvider = profilingSessionProvider;
     }
 }

StackExchange.Redis/StackExchange/Redis/ConnectionMultiplexer.cs

@@ -945,7 +945,7 @@ internal ServerEndPoint GetServerEndPoint(EndPoint endpoint, TextWriter log = nu
                     {
                         if (isDisposed) throw new ObjectDisposedException(ToString());
 
-                        server = new ServerEndPoint(this, endpoint, log);
+                        server = new ServerEndPoint(this, endpoint);
                         servers.Add(endpoint, server);
                         isNew = true;
                         _serverSnapshot = _serverSnapshot.Add(server);
@@ -991,7 +991,6 @@ private ConnectionMultiplexer(ConfigurationOptions configuration)
 
         partial void OnCreateReaderWriter(ConfigurationOptions configuration);
 
-
         internal const int MillisecondsPerHeartbeat = 1000;
         private sealed class TimerToken
         {
@@ -1020,7 +1019,6 @@ public TimerToken(ConnectionMultiplexer muxer)
                 }
             };
 
-            
             internal static IDisposable Create(ConnectionMultiplexer connection)
             {
                 var token = new TimerToken(connection);
@@ -1030,8 +1028,6 @@ internal static IDisposable Create(ConnectionMultiplexer connection)
             }
         }
 
-
-
         private int _activeHeartbeatErrors;
         private void OnHeartbeat()
         {

StackExchange.Redis/StackExchange/Redis/Interfaces/IConnectionMultiplexer.cs

@@ -74,6 +74,7 @@ public interface IConnectionMultiplexer
         /// calling context; the implementing code is responsible for reliably resolving the same provider
         /// based on ambient context, or returning null to not profile
         /// </summary>
+        /// <param name="profilingSessionProvider">The profiling session provider.</param>
         void RegisterProfiler(Func<ProfilingSession> profilingSessionProvider);
 
         /// <summary>
@@ -259,10 +260,10 @@ public interface IConnectionMultiplexer
         /// <returns>The number of instances known to have received the message (however, the actual number can be higher)</returns>
         Task<long> PublishReconfigureAsync(CommandFlags flags = CommandFlags.None);
 
-
         /// <summary>
         /// Get the hash-slot associated with a given key, if applicable; this can be useful for grouping operations
         /// </summary>
+        /// <param name="key">The key to get a the slot for.</param>
         int GetHashSlot(RedisKey key);
     }
 }

StackExchange.Redis/StackExchange/Redis/Interfaces/IServer.cs

@@ -591,12 +591,18 @@ public partial interface IServer : IRedis
         /// <summary>
         /// Swaps two Redis databases, so that immediately all the clients connected to a given database will see the data of the other database, and the other way around
         /// </summary>
+        /// <param name="first">The ID of the first database.</param>
+        /// <param name="second">The ID of the second database.</param>
+        /// <param name="flags">The command flags to use.</param>
         /// <remarks>https://redis.io/commands/swapdb</remarks>
         void SwapDatabases(int first, int second, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
         /// Swaps two Redis databases, so that immediately all the clients connected to a given database will see the data of the other database, and the other way around
         /// </summary>
+        /// <param name="first">The ID of the first database.</param>
+        /// <param name="second">The ID of the second database.</param>
+        /// <param name="flags">The command flags to use.</param>
         /// <remarks>https://redis.io/commands/swapdb</remarks>
         Task SwapDatabasesAsync(int first, int second, CommandFlags flags = CommandFlags.None);
 

StackExchange.Redis/StackExchange/Redis/Message.cs

@@ -614,8 +614,6 @@ internal void SetException(Exception exception)
 
         internal void SetEnqueued()=> performance?.SetEnqueued();
 
-
-
         internal void SetRequestSent()
         {
             Status = CommandStatus.Sent;

StackExchange.Redis/StackExchange/Redis/MessageCompletable.cs

@@ -38,7 +38,6 @@ public bool TryComplete(bool isAsync)
             }
             else
             {
-
                 if (syncHandler != null)
                 {
                     ConnectionMultiplexer.TraceWithoutContext("Invoking (sync)...: " + (string)channel, "Subscription");

StackExchange.Redis/StackExchange/Redis/PhysicalConnection.cs

@@ -96,7 +96,6 @@ internal async void BeginConnectAsync(TextWriter log)
             if(endpoint == null)
             {
                 log?.WriteLine("No endpoint");
-
             }
 
             Trace("Connecting...");
@@ -223,14 +222,7 @@ private enum ReadMode : byte
 
         public long LastWriteSecondsAgo => unchecked(Environment.TickCount - Thread.VolatileRead(ref lastWriteTickCount)) / 1000;
 
-        private bool IncludeDetailInExceptions
-        {
-            get
-            {
-                var bridge = BridgeCouldBeNull;
-                return bridge == null ? false : bridge.Multiplexer.IncludeDetailInExceptions;
-            }
-        }
+        private bool IncludeDetailInExceptions => BridgeCouldBeNull?.Multiplexer.IncludeDetailInExceptions ?? false;
 
         [Conditional("VERBOSE")]
         internal void Trace(string message) => BridgeCouldBeNull?.Multiplexer?.Trace(message, physicalName);
@@ -249,7 +241,6 @@ internal void Shutdown()
             _ioPipe = null;
             _socket = null;
 
-
             if (ioPipe != null)
             {
                 Trace("Disconnecting...");
@@ -336,7 +327,7 @@ public void RecordConnectionFailed(ConnectionFailureType failureType, Exception
                     var bridge = BridgeCouldBeNull;
                     if (bridge != null)
                     {
-                        exMessage.Append(" on " + Format.ToString(bridge.ServerEndPoint?.EndPoint) + "/" + connectionType);
+                        exMessage.Append(" on ").Append(Format.ToString(bridge.ServerEndPoint?.EndPoint)).Append("/").Append(connectionType);
 
                         data.Add(Tuple.Create("FailureType", failureType.ToString()));
                         data.Add(Tuple.Create("EndPoint", Format.ToString(bridge.ServerEndPoint?.EndPoint)));
@@ -538,7 +529,6 @@ internal void OnBridgeHeartbeat()
             {
                 if (_writtenAwaitingResponse.Count != 0)
                 {
-                    
                     var bridge = BridgeCouldBeNull;
                     if (bridge == null) return;
 
@@ -579,9 +569,9 @@ internal void SetUnknownDatabase()
         internal void Write(RedisKey key)
         {
             var val = key.KeyValue;
-            if (val is string)
+            if (val is string s)
             {
-                WriteUnified(_ioPipe.Output, key.KeyPrefix, (string)val, outEncoder);
+                WriteUnified(_ioPipe.Output, key.KeyPrefix, s, outEncoder);
             }
             else
             {
@@ -923,7 +913,7 @@ internal static void WriteUnified(PipeWriter writer, byte[] prefix, string value
                 // ${total-len}\r\n         3 + MaxInt32TextLen
                 // {prefix}{value}\r\n
                 int encodedLength = Encoding.UTF8.GetByteCount(value),
-                    prefixLength = prefix == null ? 0 : prefix.Length,
+                    prefixLength = prefix?.Length ?? 0,
                     totalLength = prefixLength + encodedLength;
 
                 if (totalLength == 0)
@@ -1271,7 +1261,7 @@ internal void OnHeartbeat()
                     // - so: only use that if we're making progress
                     if (!(allowSyncRead && input.TryRead(out var readResult)))
                     {
-                        readResult = await input.ReadAsync();
+                        readResult = await input.ReadAsync().ForAwait();
                     }
 
                     var buffer = readResult.Buffer;
@@ -1458,7 +1448,7 @@ private static RawResult ReadLineTerminatedString(ResultType type, in ReadOnlySe
                     throw new InvalidOperationException("Unexpected response prefix: " + (char)prefix);
             }
         }
-        static RawResult ParseInlineProtocol(RawResult line)
+        private static RawResult ParseInlineProtocol(RawResult line)
         {
             if (!line.HasValue) return RawResult.Nil; // incomplete line
 
@@ -1472,7 +1462,5 @@ static RawResult ParseInlineProtocol(RawResult line)
             }
             return new RawResult(oversized, count);
         }
-
-
     }
 }

StackExchange.Redis/StackExchange/Redis/Profiling/ProfilingSession.cs

@@ -3,26 +3,27 @@
 namespace StackExchange.Redis.Profiling
 {
     /// <summary>
-    /// Lightweight profiling session that can be optionally registered (via ConnectionMultiplexer.RegisterProfiler) to track messages
+    /// Lightweight profiling session that can be optionally registered (via ConnectionMultiplexer.RegisterProfiler) to track messages.
     /// </summary>
     public sealed class ProfilingSession
     {
         /// <summary>
-        /// Caller-defined state object
+        /// Caller-defined state object.
         /// </summary>
         public object UserToken { get; }
         /// <summary>
-        /// Create a new profiling session, optionally including a caller-defined state object
+        /// Create a new profiling session, optionally including a caller-defined state object.
         /// </summary>
+        /// <param name="userToken">The state object to use for this session.</param>
         public ProfilingSession(object userToken = null) => UserToken = userToken;
 
-        object _untypedHead;
+        private object _untypedHead;
 
         internal void Add(ProfiledCommand command)
         {
             if (command == null) return;
 
-            object cur = Thread.VolatileRead(ref _untypedHead); ;
+            object cur = Thread.VolatileRead(ref _untypedHead);
             while (true)
             {
                 command.NextElement = (ProfiledCommand)cur;
@@ -34,7 +35,7 @@ internal void Add(ProfiledCommand command)
 
         /// <summary>
         /// Reset the session and yield the commands that were captured for enumeration; if additional commands
-        /// are added, they can be retrieved via additional calls to FinishProfiling
+        /// are added, they can be retrieved via additional calls to FinishProfiling.
         /// </summary>
         public ProfiledCommandEnumerable FinishProfiling()
         {

StackExchange.Redis/StackExchange/Redis/RawResult.cs

@@ -89,7 +89,7 @@ public override string ToString()
             // specifically; the line: abc    "def ghi" jkl
             // is 3 tokens: "abc", "def ghi" and "jkl"
             public Tokenizer GetEnumerator() => this;
-            BufferReader _value;
+            private BufferReader _value;
 
             public Tokenizer(ReadOnlySequence<byte> value)
             {
@@ -128,10 +128,8 @@ public bool MoveNext()
                     _value.Consume(1); // drop the terminator itself;
                 }
                 return true;
-
             }
             public ReadOnlySequence<byte> Current { get; private set; }
-
         }
         internal RedisChannel AsRedisChannel(byte[] channelPrefix, RedisChannel.PatternMode mode)
         {

StackExchange.Redis/StackExchange/Redis/RedisResult.cs

@@ -15,7 +15,6 @@ public abstract class RedisResult
         /// <returns> new <see cref="RedisResult"/>.</returns>
         public static RedisResult Create(RedisValue value, ResultType? resultType = null) => new SingleRedisResult(value, resultType);
 
-
         /// <summary>
         /// Create a new RedisResult representing an array of values.
         /// </summary>
@@ -36,12 +35,12 @@ public static RedisResult Create(RedisResult[] values)
         /// <summary>
         /// An empty array result
         /// </summary>
-        public static RedisResult EmptyArray { get; } = new ArrayRedisResult(Array.Empty<RedisResult>());
+        internal static RedisResult EmptyArray { get; } = new ArrayRedisResult(Array.Empty<RedisResult>());
 
         /// <summary>
         /// A null array result
         /// </summary>
-        public static RedisResult NullArray { get; } = new ArrayRedisResult(null);
+        internal static RedisResult NullArray { get; } = new ArrayRedisResult(null);
 
         // internally, this is very similar to RawResult, except it is designed to be usable
         // outside of the IO-processing pipeline: the buffers are standalone, etc
@@ -90,23 +89,6 @@ internal static RedisResult TryCreate(PhysicalConnection connection, RawResult r
         /// Indicates whether this result was a null result
         /// </summary>
         public abstract bool IsNull { get; }
-        /// <summary>
-        /// A successful result
-        /// </summary>
-        public static RedisResult OK { get; } = Create("OK", ResultType.SimpleString);
-        /// <summary>
-        /// An integer-zero result
-        /// </summary>
-        public static RedisResult Zero { get; } = Create(0, ResultType.Integer);
-        /// <summary>
-        /// An integer-one result
-        /// </summary>
-        public static RedisResult One { get; } = Create(1, ResultType.Integer);
-
-        /// <summary>
-        /// A null bulk-string result
-        /// </summary>
-        public static RedisResult Null { get; } = Create(RedisValue.Null, ResultType.BulkString);
         
         /// <summary>
         /// Interprets the result as a <see cref="string"/>.
@@ -267,8 +249,8 @@ internal override byte[][] AsByteArrayArray()
                 : _value.Length == 0 ? Array.Empty<byte[]>()
                 : Array.ConvertAll(_value, x => x.AsByteArray());
 
-            private bool IsSingleton => _value != null && _value.Length == 1;
-            private bool IsEmpty => _value != null && _value.Length == 0;
+            private bool IsSingleton => _value?.Length == 1;
+            private bool IsEmpty => _value?.Length == 0;
             internal override double AsDouble()
             {
                 if (IsSingleton) return _value[0].AsDouble();
@@ -363,12 +345,15 @@ internal override string[] AsStringArray()
         }
 
         /// <summary>
-        /// Create a RedisResult from a key
+        /// Create a <see cref="RedisResult"/> from a key.
         /// </summary>
+        /// <param name="key">The <see cref="RedisKey"/> to create a <see cref="RedisResult"/> from.</param>
         public static RedisResult Create(RedisKey key) => Create(key.AsRedisValue(), ResultType.BulkString);
+
         /// <summary>
-        /// Create a RedisResult from a channel
+        /// Create a <see cref="RedisResult"/> from a channel.
         /// </summary>
+        /// <param name="channel">The <see cref="RedisChannel"/> to create a <see cref="RedisResult"/> from.</param>
         public static RedisResult Create(RedisChannel channel) => Create((byte[])channel, ResultType.BulkString);
 
         private sealed class ErrorRedisResult : RedisResult

StackExchange.Redis/StackExchange/Redis/RedisSubscriber.cs

@@ -325,7 +325,7 @@ public Task SubscribeAsync(RedisChannel channel, Action<RedisChannel, RedisValue
         public async Task<ChannelMessageQueue> SubscribeAsync(RedisChannel channel, CommandFlags flags = CommandFlags.None)
         {
             var c = new ChannelMessageQueue(channel, this);
-            await c.SubscribeAsync(flags);
+            await c.SubscribeAsync(flags).ForAwait();
             return c;
         }
 

StackExchange.Redis/StackExchange/Redis/RedisValue.cs

@@ -26,9 +26,6 @@ private RedisValue(long overlappedValue64, ReadOnlyMemory<byte> memory, object o
         }
 
         private readonly static object Sentinel_Integer = new object();
-
-        
-
         private readonly static object Sentinel_Raw = new object();
         private readonly static object Sentinel_Double = new object();
 
@@ -41,9 +38,11 @@ public object Box()
             if (obj is null || obj is string || obj is byte[]) return obj;
             return this;
         }
+
         /// <summary>
-        /// Parse this object as a value - to be used alongside Box
+        /// Parse this object as a value - to be used alongside Box.
         /// </summary>
+        /// <param name="value">The value to unbox.</param>
         public static RedisValue Unbox(object value)
         {
             if (value == null) return RedisValue.Null;

StackExchange.Redis/StackExchange/Redis/ScriptParameterMapper.cs

@@ -106,17 +106,16 @@ private static string MakeOrdinalScriptWithoutKeys(string rawScript, string[] ar
             return ret.ToString();
         }
 
-        private static Dictionary<Type, MethodInfo> _conversionOperators;
+        private static readonly Dictionary<Type, MethodInfo> _conversionOperators;
         static ScriptParameterMapper()
         {
             var tmp = new Dictionary<Type, MethodInfo>();
-            foreach(var method in typeof(RedisValue).GetMethods(BindingFlags.Public | BindingFlags.Static))
+            foreach (var method in typeof(RedisValue).GetMethods(BindingFlags.Public | BindingFlags.Static))
             {
-                if(method.ReturnType == typeof(RedisValue) &&
-                    (method.Name == "op_Implicit" || method.Name == "op_Explicit"))
+                if (method.ReturnType == typeof(RedisValue) && (method.Name == "op_Implicit" || method.Name == "op_Explicit"))
                 {
                     var p = method.GetParameters();
-                    if (p != null && p.Length == 1)
+                    if (p?.Length == 1)
                     {
                         tmp[p[0].ParameterType] = method;
                     }
@@ -176,7 +175,8 @@ public static bool IsValidParameterHash(Type t, LuaScript script, out string mis
                 }
 
                 var memberType = member is FieldInfo ? ((FieldInfo)member).FieldType : ((PropertyInfo)member).PropertyType;
-                if(!ConvertableTypes.Contains(memberType)){
+                if (!ConvertableTypes.Contains(memberType))
+                {
                     missingMember = null;
                     badTypeMember = argName;
                     return false;
@@ -210,7 +210,7 @@ public static bool IsValidParameterHash(Type t, LuaScript script, out string mis
 
             Expression GetMember(Expression root, MemberInfo member)
             {
-                switch(member.MemberType)
+                switch (member.MemberType)
                 {
                     case MemberTypes.Property:
                         return Expression.Property(root, (PropertyInfo)member);
@@ -264,7 +264,7 @@ Expression GetMember(Expression root, MemberInfo member)
                     BindingFlags.NonPublic | BindingFlags.Instance);
 
                 keysResultArr = new Expression[keys.Count];
-                for(int i = 0; i < keysResultArr.Length; i++)
+                for (int i = 0; i < keysResultArr.Length; i++)
                 {
                     var member = GetMember(objTyped, keys[i]);
                     keysResultArr[i] = Expression.Condition(needsKeyPrefix,
@@ -274,7 +274,6 @@ Expression GetMember(Expression root, MemberInfo member)
                 keysResult = Expression.NewArrayInit(typeof(RedisKey), keysResultArr);
             }
 
-            
             if (args.Count == 0)
             {
                 // if there are no args, don't allocate

StackExchange.Redis/StackExchange/Redis/ServerEndPoint.cs

@@ -43,7 +43,7 @@ internal void ResetNonConnected()
             subscription?.ResetNonConnected();
         }
 
-        public ServerEndPoint(ConnectionMultiplexer multiplexer, EndPoint endpoint, TextWriter log)
+        public ServerEndPoint(ConnectionMultiplexer multiplexer, EndPoint endpoint)
         {
             Multiplexer = multiplexer;
             EndPoint = endpoint;
@@ -642,7 +642,6 @@ internal void WriteDirectOrQueueFireAndForget<T>(PhysicalConnection connection,
                     {
                         bridge.WriteMessageTakingWriteLock(connection, message);
                     }
-                    
                 }
             }
         }

StackExchange.Redis/StackExchange/Redis/StreamPendingInfo.cs

@@ -36,6 +36,5 @@ namespace StackExchange.Redis
         /// An array of consumers within the consumer group that have pending messages.
         /// </summary>
         public StreamConsumer[] Consumers { get; }
-
     }
 }

TestConsole/Program.cs

@@ -7,10 +7,11 @@ static class Program
 {
     static async Task Main()
     {
-        using (var server = new MemoryCacheRedisServer(Console.Out))
+        using (var resp = new MemoryCacheRedisServer(Console.Out))
+        using (var socket = new RespSocketServer(resp))
         {
-            server.Listen(new IPEndPoint(IPAddress.Loopback, 6378));
-            await server.Shutdown;
+            socket.Listen(new IPEndPoint(IPAddress.Loopback, 6378));
+            await resp.Shutdown;
         }
     }
 }