Removing use of System.Threading.Channels

Author: jimevans

dotnet/src/webdriver/BUILD.bazel

@@ -65,8 +65,6 @@ csharp_library(
  "@identitymodel.tokens//:Microsoft.IdentityModel.Tokens",
  "@json.net//:Newtonsoft.Json",
  "@system.drawing.common//:System.Drawing.Common",
- "@system.threading.channels//:System.Threading.Channels",
- "@system.threading.tasks.extensions//:System.Threading.Tasks.Extensions",
  ],
 )
 

dotnet/src/webdriver/DevTools/DevToolsSession.cs

@@ -21,7 +21,6 @@
 using System.Globalization;
 using System.Net.Http;
 using System.Threading;
-using System.Threading.Channels;
 using System.Threading.Tasks;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
@@ -46,14 +45,12 @@ public class DevToolsSession : IDevToolsSession
 
  private WebSocketConnection connection;
  private ConcurrentDictionary<long, DevToolsCommandData> pendingCommands = new ConcurrentDictionary<long, DevToolsCommandData>();
- private readonly Channel<string> messageQueue;
+ private readonly BlockingCollection<string> messageQueue = new BlockingCollection<string>();
  private readonly Task messageQueueMonitorTask;
  private long currentCommandId = 0;
 
  private DevToolsDomains domains;
 
- private CancellationTokenSource receiveCancellationToken;
-
  /// <summary>
  /// Initializes a new instance of the DevToolsSession class, using the specified WebSocket endpoint.
  /// </summary>
@@ -71,11 +68,6 @@ public DevToolsSession(string endpointAddress)
  {
  this.websocketAddress = endpointAddress;
  }
- this.messageQueue = Channel.CreateUnbounded<string>(new UnboundedChannelOptions()
- {
- SingleReader = true,
- SingleWriter = true,
- });
  this.messageQueueMonitorTask = Task.Run(() => this.MonitorMessageQueue());
  this.messageQueueMonitorTask.ConfigureAwait(false);
  }
@@ -166,7 +158,8 @@ public T GetVersionSpecificDomains<T>() where T : DevToolsSessionDomains
  /// </summary>
  /// <typeparam name="TCommandResponse"></typeparam>
  /// <typeparam name="TCommand">A command object implementing the <see cref="ICommand"/> interface.</typeparam>
- /// <param name="cancellationToken">A CancellationToken object to allow for cancellation of the command.</param>
+ /// <param name="command">A CancellationToken object to allow for cancellation of the command.</param>
+ /// <param name="cancellationToken">The command to be sent.</param>
  /// <param name="millisecondsTimeout">The execution timeout of the command in milliseconds.</param>
  /// <param name="throwExceptionIfResponseNotReceived"><see langword="true"/> to throw an exception if a response is not received; otherwise, <see langword="false"/>.</param>
  /// <returns>The command response object implementing the <see cref="ICommandResponse{T}"/> interface.</returns>
@@ -295,7 +288,7 @@ internal async Task StartSession(int requestedProtocolVersion)
  /// Asynchronously stops the session.
  /// </summary>
  /// <param name="manualDetach"><see langword="true"/> to manually detach the session
- /// from its attached target; otherswise <see langword="false""/>.</param>
+ /// from its attached target; otherswise <see langword="false"/>.</param>
  /// <returns>A task that represents the asynchronous operation.</returns>
  internal async Task StopSession(bool manualDetach)
  {
@@ -413,38 +406,60 @@ private void OnTargetDetached(object sender, TargetDetachedEventArgs e)
  private async Task InitializeSocketConnection()
  {
  LogTrace("Creating WebSocket");
- this.connection = new WebSocketConnection();
+ this.connection = new WebSocketConnection(this.openConnectionWaitTimeSpan, this.closeConnectionWaitTimeSpan);
  connection.DataReceived += OnConnectionDataReceived;
  await connection.Start(this.websocketAddress);
- LogTrace("WebSocket created; starting message listener");
+ LogTrace("WebSocket created");
  }
 
  private async Task TerminateSocketConnection()
  {
+ LogTrace("Closing WebSocket");
  if (this.connection != null && this.connection.IsActive)
  {
  await this.connection.Stop();
  await this.ShutdownMessageQueue();
  }
+ LogTrace("WebSocket closed");
  }
 
  private async Task ShutdownMessageQueue()
  {
- // Attempt to wait for the channel to empty before marking the
- // writer as complete and waiting for the monitor task to end.
  // THe WebSockect connection is always closed before this method
  // is called, so there will eventually be no more data written
- // into the message queue, so this loop should be guaranteed to
- // complete.
- while (this.messageQueue.Reader.TryPeek(out _))
+ // into the message queue, meaning this loop should be guaranteed
+ // to complete.
+ while (this.connection.IsActive)
  {
  await Task.Delay(TimeSpan.FromMilliseconds(10));
  }
 
- this.messageQueue.Writer.Complete();
+ this.messageQueue.CompleteAdding();
  await this.messageQueueMonitorTask;
  }
 
+ private void MonitorMessageQueue()
+ {
+ // Loop until the BlockingCollection is marked as completed for adding
+ // (meaning no more information will be added into the collection, and
+ // it is empty (meaning all items in the collection have been processed).
+ while (!this.messageQueue.IsCompleted)
+ {
+ try
+ {
+ // The Take() method blocks until there is something to
+ // remove from the BlockingCollection.
+ this.ProcessMessage(this.messageQueue.Take());
+ }
+ catch (InvalidOperationException)
+ {
+ // InvalidOperationException is normal when the collection
+ // is marked as completed while being blocked by the Take()
+ // method.
+ }
+ }
+ }
+
  private void ProcessMessage(string message)
  {
  var messageObject = JObject.Parse(message);
@@ -498,25 +513,6 @@ private void ProcessMessage(string message)
  LogTrace("Recieved Other: {0}", message);
  }
 
- /// <summary>
- /// Reads incoming messages in the message queue.
- /// </summary>
- private void ReadIncomingMessages()
- {
- while (this.messageQueue.Reader.TryRead(out string message))
- {
- this.ProcessMessage(message);
- }
- }
-
- private async Task MonitorMessageQueue()
- {
- while (await this.messageQueue.Reader.WaitToReadAsync())
- {
- this.ReadIncomingMessages();
- }
- }
-
  private void OnDevToolsEventReceived(DevToolsEventReceivedEventArgs e)
  {
  if (DevToolsEventReceived != null)
@@ -527,7 +523,7 @@ private void OnDevToolsEventReceived(DevToolsEventReceivedEventArgs e)
 
  private void OnConnectionDataReceived(object sender, WebSocketConnectionDataReceivedEventArgs e)
  {
- _ = this.messageQueue.Writer.TryWrite(e.Data);
+ this.messageQueue.Add(e.Data);
  }
 
  private void LogTrace(string message, params object[] args)

dotnet/src/webdriver/DevTools/WebSocketConnection.cs

@@ -142,7 +142,10 @@ public virtual async Task Stop()
  // Whether we closed the socket or timed out, we cancel the token causing ReceiveAsync to abort the socket.
  // The finally block at the end of the processing loop will dispose of the ClientWebSocket object.
  this.clientTokenSource.Cancel();
- this.dataReceiveTask?.Wait();
+ if (this.dataReceiveTask != null)
+ {
+ await this.dataReceiveTask;
+ }
  }
 
  /// <summary>

dotnet/src/webdriver/WebDriver.csproj

@@ -48,15 +48,6 @@
  <EmbeddedResource Remove="obj\**" />
  <None Remove="cdp\**" />
  <None Remove="obj\**" />
- <Compile Remove="Internal\IFindsByClassName.cs" />
- <Compile Remove="Internal\IFindsByCssSelector.cs" />
- <Compile Remove="Internal\IFindsById.cs" />
- <Compile Remove="Internal\IFindsByLinkText.cs" />
- <Compile Remove="Internal\IFindsByName.cs" />
- <Compile Remove="Internal\IFindsByPartialLinkText.cs" />
- <Compile Remove="Internal\IFindsByTagName.cs" />
- <Compile Remove="Internal\IFindsByXPath.cs" />
- <Compile Remove="Interactions\TouchActions.cs" />
  </ItemGroup>
 
  <ItemGroup>
@@ -76,7 +67,6 @@
  <ItemGroup>
  <!--PackageReference Include="Microsoft.IdentityModel.Tokens" Version="6.19.0" PrivateAssets="All" /-->
  <PackageReference Include="System.Drawing.Common" Version="7.0.0" />
- <PackageReference Include="System.Threading.Channels" Version="7.0.0" />
  </ItemGroup>
 
  <ItemGroup>

dotnet/workspace.bzl

@@ -41,18 +41,6 @@ def selenium_register_dotnet():
  sha256 = "b78141813321b8e039140db77e0d1640e1ae1f49e357b8495d24ff2e7fd99e4b",
  )
 
- import_nuget_package(
- name = "system.threading.channels",
- file = "third_party/dotnet/nuget/packages/system.threading.channels.7.0.0.nupkg",
- sha256 = "0aed208d0b0b211f18be1d01e1c38f252615ab5d1afb717da55cff0b8dc235e3",
- )
-
- import_nuget_package(
- name = "system.threading.tasks.extensions",
- file = "third_party/dotnet/nuget/packages/system.threading.tasks.extensions.4.5.4.nupkg",
- sha256 = "a304a963cc0796c5179f9c6b7d8022bbce3b2fa7c029eb6196f631f7b462d678",
- )
-
  import_nuget_package(
  name = "moq",
  file = "third_party/dotnet/nuget/packages/moq.4.12.0.nupkg",