Establish new channel connection model using server listeners

This change refactors our existing channel model to move all connection
logic outside of the ChannelBase implementations so that the
language and debug client/service pairs can be simplified.  This change
also allows us to remove a long-standing hack in our Host unit tests
which added an artifical delay to give the channel and MessageDispatcher
time to get established.

Author: daviwil

src/PowerShellEditorServices.Host/EditorServicesHost.cs

@@ -3,7 +3,8 @@
 // Licensed under the MIT license. See LICENSE file in the project root for full license information.
 //
 
-using Microsoft.PowerShell.EditorServices.Console;
+using Microsoft.PowerShell.EditorServices.Extensions;
+using Microsoft.PowerShell.EditorServices.Protocol.MessageProtocol;
 using Microsoft.PowerShell.EditorServices.Protocol.MessageProtocol.Channel;
 using Microsoft.PowerShell.EditorServices.Protocol.Server;
 using Microsoft.PowerShell.EditorServices.Session;
@@ -12,10 +13,8 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Management.Automation.Runspaces;
-using System.Management.Automation.Host;
 using System.Reflection;
-using System.Threading;
-using Microsoft.PowerShell.EditorServices.Extensions;
+using System.Threading.Tasks;
 
 namespace Microsoft.PowerShell.EditorServices.Host
 {
@@ -36,12 +35,18 @@ public class EditorServicesHost
 
         private bool enableConsoleRepl;
         private HostDetails hostDetails;
+        private ProfilePaths profilePaths;
         private string bundledModulesPath;
         private DebugAdapter debugAdapter;
         private EditorSession editorSession;
         private HashSet<string> featureFlags;
         private LanguageServer languageServer;
 
+        private TcpSocketServerListener languageServiceListener;
+        private TcpSocketServerListener debugServiceListener;
+
+        private TaskCompletionSource<bool> serverCompletedTask;
+
         #endregion
 
         #region Properties
@@ -152,25 +157,40 @@ public void StartLogging(string logFilePath, LogLevel logLevel)
         /// <param name="languageServicePort">The port number for the language service.</param>
         /// <param name="profilePaths">The object containing the profile paths to load for this session.</param>
         public void StartLanguageService(int languageServicePort, ProfilePaths profilePaths)
+        {
+            this.profilePaths = profilePaths;
+
+            this.languageServiceListener =
+                new TcpSocketServerListener(
+                    MessageProtocolType.LanguageServer,
+                    languageServicePort);
+
+            this.languageServiceListener.ClientConnect += this.OnLanguageServiceClientConnect;
+            this.languageServiceListener.Start();
+
+            Logger.Write(
+                LogLevel.Normal,
+                string.Format(
+                    "Language service started, listening on port {0}",
+                    languageServicePort));
+        }
+
+        private async void OnLanguageServiceClientConnect(
+            object sender,
+            TcpSocketServerChannel serverChannel)
         {
             this.editorSession =
                 CreateSession(
                     this.hostDetails,
-                    profilePaths,
+                    this.profilePaths,
                     this.enableConsoleRepl);
 
             this.languageServer =
                 new LanguageServer(
                     this.editorSession,
-                    new TcpSocketServerChannel(languageServicePort));
-
-            this.languageServer.Start().Wait();
+                    serverChannel);
 
-            Logger.Write(
-                LogLevel.Normal,
-                string.Format(
-                    "Language service started, listening on port {0}",
-                    languageServicePort));
+            await this.languageServer.Start();
         }
 
         /// <summary>
@@ -182,12 +202,29 @@ public void StartDebugService(
             ProfilePaths profilePaths,
             bool useExistingSession)
         {
-            if (this.enableConsoleRepl && useExistingSession)
+            this.debugServiceListener =
+                new TcpSocketServerListener(
+                    MessageProtocolType.LanguageServer,
+                    debugServicePort);
+
+            this.debugServiceListener.ClientConnect += OnDebugServiceClientConnect;
+            this.debugServiceListener.Start();
+
+            Logger.Write(
+                LogLevel.Normal,
+                string.Format(
+                    "Debug service started, listening on port {0}",
+                    debugServicePort));
+        }
+
+        private async void OnDebugServiceClientConnect(object sender, TcpSocketServerChannel serverChannel)
+        {
+            if (this.enableConsoleRepl)
             {
                 this.debugAdapter =
                     new DebugAdapter(
                         this.editorSession,
-                        new TcpSocketServerChannel(debugServicePort),
+                        serverChannel,
                         false);
             }
             else
@@ -196,42 +233,26 @@ public void StartDebugService(
                     this.CreateDebugSession(
                         this.hostDetails,
                         profilePaths,
-                        this.languageServer.EditorOperations);
+                        this.languageServer?.EditorOperations);
 
                 this.debugAdapter =
                     new DebugAdapter(
                         debugSession,
-                        new TcpSocketServerChannel(debugServicePort),
+                        serverChannel,
                         true);
             }
 
             this.debugAdapter.SessionEnded +=
                 (obj, args) =>
                 {
-                    // Only restart if we're reusing the existing session
-                    // or if we're not using the console REPL, otherwise
-                    // the process should terminate
-                    if (useExistingSession)
-                    {
-                        Logger.Write(
-                            LogLevel.Normal,
-                            "Previous debug session ended, restarting debug service...");
-
-                        this.StartDebugService(debugServicePort, profilePaths, true);
-                    }
-                    else if (!this.enableConsoleRepl)
-                    {
-                        this.StartDebugService(debugServicePort, profilePaths, false);
-                    }
-                };
+                    Logger.Write(
+                        LogLevel.Normal,
+                        "Previous debug session ended, restarting debug service listener...");
 
-            this.debugAdapter.Start().Wait();
+                    this.debugServiceListener.Start();
+                };
 
-            Logger.Write(
-                LogLevel.Normal,
-                string.Format(
-                    "Debug service started, listening on port {0}",
-                    debugServicePort));
+            await this.debugAdapter.Start();
         }
 
         /// <summary>
@@ -251,17 +272,9 @@ public void StopServices()
         /// </summary>
         public void WaitForCompletion()
         {
-            // Wait based on which server is started.  If the language server
-            // hasn't been started then we may only need to wait on the debug
-            // adapter to complete.
-            if (this.languageServer != null)
-            {
-                this.languageServer.WaitForExit();
-            }
-            else if (this.debugAdapter != null)
-            {
-                this.debugAdapter.WaitForExit();
-            }
+            // TODO: We need a way to know when to complete this task!
+            this.serverCompletedTask = new TaskCompletionSource<bool>();
+            this.serverCompletedTask.Task.Wait();
         }
 
         #endregion

src/PowerShellEditorServices.Protocol/Client/DebugAdapterClientBase.cs

@@ -32,11 +32,6 @@ await this.SendRequest(
         }
 
         protected override Task OnStart()
-        {
-            return Task.FromResult(true);
-        }
-
-        protected override Task OnConnect()
         {
             // Initialize the debug adapter
             return this.SendRequest(

src/PowerShellEditorServices.Protocol/Client/LanguageServiceClient.cs

@@ -28,11 +28,6 @@ protected override Task Initialize()
             // Add handlers for common events
             this.SetEventHandler(PublishDiagnosticsNotification.Type, HandlePublishDiagnosticsEvent);
 
-            return Task.FromResult(true);
-        }
-
-        protected override Task OnConnect()
-        {
             // Send the 'initialize' request and wait for the response
             var initializeParams = new InitializeParams
             {

src/PowerShellEditorServices.Protocol/MessageProtocol/Channel/ChannelBase.cs

@@ -14,11 +14,6 @@ namespace Microsoft.PowerShell.EditorServices.Protocol.MessageProtocol.Channel
     /// </summary>
     public abstract class ChannelBase
     {
-        /// <summary>
-        /// Gets a boolean that is true if the channel is connected or false if not.
-        /// </summary>
-        public bool IsConnected { get; protected set; }
-
         /// <summary>
         /// Gets the MessageReader for reading messages from the channel.
         /// </summary>
@@ -48,14 +43,6 @@ public void Start(MessageProtocolType messageProtocolType)
             this.Initialize(messageSerializer);
         }
 
-        /// <summary>
-        /// Returns a Task that allows the consumer of the ChannelBase
-        /// implementation to wait until a connection has been made to
-        /// the opposite endpoint whether it's a client or server.
-        /// </summary>
-        /// <returns>A Task to be awaited until a connection is made.</returns>
-        public abstract Task WaitForConnection();
-
         /// <summary>
         /// Stops the channel.
         /// </summary>

src/PowerShellEditorServices.Protocol/MessageProtocol/Channel/NamedPipeClientChannel.cs

@@ -11,51 +11,15 @@ namespace Microsoft.PowerShell.EditorServices.Protocol.MessageProtocol.Channel
 {
     public class NamedPipeClientChannel : ChannelBase
     {
-        private string pipeName;
         private NamedPipeClientStream pipeClient;
 
-        public NamedPipeClientChannel(string pipeName)
+        public NamedPipeClientChannel(NamedPipeClientStream pipeClient)
         {
-            this.pipeName = pipeName;
-        }
-
-        public override async Task WaitForConnection()
-        {
-#if CoreCLR
-            await this.pipeClient.ConnectAsync();
-#else
-            this.IsConnected = false;
-
-            while (!this.IsConnected)
-            {
-                try
-                {
-                    // Wait for 500 milliseconds so that we don't tie up the thread
-                    this.pipeClient.Connect(500);
-                    this.IsConnected = this.pipeClient.IsConnected;
-                }
-                catch (TimeoutException)
-                {
-                    // Connect timed out, wait and try again
-                    await Task.Delay(1000);
-                    continue;
-                }
-            }
-#endif
-
-            // If we've reached this point, we're connected
-            this.IsConnected = true;
+            this.pipeClient = pipeClient;
         }
 
         protected override void Initialize(IMessageSerializer messageSerializer)
         {
-            this.pipeClient =
-                new NamedPipeClientStream(
-                    ".",
-                    this.pipeName,
-                    PipeDirection.InOut,
-                    PipeOptions.Asynchronous);
-
             this.MessageReader =
                 new MessageReader(
                     this.pipeClient,
@@ -74,6 +38,41 @@ protected override void Shutdown()
                 this.pipeClient.Dispose();
             }
         }
+
+        public static async Task<NamedPipeClientChannel> Connect(
+            string pipeName,
+            MessageProtocolType messageProtocolType)
+        {
+            var pipeClient =
+                new NamedPipeClientStream(
+                    ".",
+                    pipeName,
+                    PipeDirection.InOut,
+                    PipeOptions.Asynchronous);
+
+#if CoreCLR
+            await pipeClient.ConnectAsync();
+#else
+            while (!pipeClient.IsConnected)
+            {
+                try
+                {
+                    // Wait for 500 milliseconds so that we don't tie up the thread
+                    pipeClient.Connect(500);
+                }
+                catch (TimeoutException)
+                {
+                    // Connect timed out, wait and try again
+                    await Task.Delay(1000);
+                    continue;
+                }
+            }
+#endif
+            var clientChannel = new NamedPipeClientChannel(pipeClient);
+            clientChannel.Start(messageProtocolType);
+
+            return clientChannel;
+        }
     }
 }