Merge remote-tracking branch 'upstream/trunk' into dotnet-bidi-events

Author: nvborisenko

dotnet/src/webdriver/BiDi/Communication/Broker.cs

@@ -112,15 +112,17 @@ public async Task ConnectAsync(CancellationToken cancellationToken)
         await _transport.ConnectAsync(cancellationToken).ConfigureAwait(false);
 
         _receiveMessagesCancellationTokenSource = new CancellationTokenSource();
-        _receivingMessageTask = _myTaskFactory.StartNew(async () => await ReceiveMessagesAsync(_receiveMessagesCancellationTokenSource.Token), TaskCreationOptions.LongRunning).Unwrap();
-        _eventEmitterTask = _myTaskFactory.StartNew(async () => await ProcessEventsAwaiterAsync(), TaskCreationOptions.LongRunning).Unwrap();
+        _receivingMessageTask = _myTaskFactory.StartNew(async () => await ReceiveMessagesAsync(_receiveMessagesCancellationTokenSource.Token)).Unwrap();
+        _eventEmitterTask = _myTaskFactory.StartNew(ProcessEventsAwaiterAsync).Unwrap();
     }
 
     private async Task ReceiveMessagesAsync(CancellationToken cancellationToken)
     {
         while (!cancellationToken.IsCancellationRequested)
         {
-            var message = await _transport.ReceiveAsJsonAsync<Message>(_jsonSerializerContext, cancellationToken);
+            var data = await _transport.ReceiveAsync(cancellationToken).ConfigureAwait(false);
+
+            var message = JsonSerializer.Deserialize(new ReadOnlySpan<byte>(data), _jsonSerializerContext.Message);
 
             switch (message)
             {
@@ -208,7 +210,9 @@ private async Task<JsonElement> ExecuteCommandCoreAsync<TCommand>(TCommand comma
 
         _pendingCommands[command.Id] = tcs;
 
-        await _transport.SendAsJsonAsync(command, _jsonSerializerContext, cts.Token).ConfigureAwait(false);
+        var data = JsonSerializer.SerializeToUtf8Bytes(command, typeof(TCommand), _jsonSerializerContext);
+
+        await _transport.SendAsync(data, cts.Token).ConfigureAwait(false);
 
         return await tcs.Task.ConfigureAwait(false);
     }

dotnet/src/webdriver/BiDi/Communication/Transport/ITransport.cs

@@ -20,7 +20,6 @@
 using System.Threading.Tasks;
 using System.Threading;
 using System;
-using System.Text.Json.Serialization;
 
 #nullable enable
 
@@ -30,8 +29,7 @@ interface ITransport : IDisposable
 {
     Task ConnectAsync(CancellationToken cancellationToken);
 
-    Task<T> ReceiveAsJsonAsync<T>(JsonSerializerContext jsonSerializerContext, CancellationToken cancellationToken);
+    Task<byte[]> ReceiveAsync(CancellationToken cancellationToken);
 
-    Task SendAsJsonAsync<TCommand>(TCommand command, JsonSerializerContext jsonSerializerContext, CancellationToken cancellationToken)
-        where TCommand : Command;
+    Task SendAsync(byte[] data, CancellationToken cancellationToken);
 }

dotnet/src/webdriver/BiDi/Communication/Transport/WebSocketTransport.cs

@@ -22,10 +22,8 @@
 using System.Net.WebSockets;
 using System.Threading.Tasks;
 using System.Threading;
-using System.Text.Json;
 using System.Text;
 using OpenQA.Selenium.Internal.Logging;
-using System.Text.Json.Serialization;
 
 #nullable enable
 
@@ -45,7 +43,7 @@ public async Task ConnectAsync(CancellationToken cancellationToken)
         await _webSocket.ConnectAsync(_uri, cancellationToken).ConfigureAwait(false);
     }
 
-    public async Task<T> ReceiveAsJsonAsync<T>(JsonSerializerContext jsonSerializerContext, CancellationToken cancellationToken)
+    public async Task<byte[]> ReceiveAsync(CancellationToken cancellationToken)
     {
         using var ms = new MemoryStream();
 
@@ -61,31 +59,28 @@ public async Task<T> ReceiveAsJsonAsync<T>(JsonSerializerContext jsonSerializerC
 
         ms.Seek(0, SeekOrigin.Begin);
 
+        byte[] data = ms.ToArray();
+
         if (_logger.IsEnabled(LogEventLevel.Trace))
         {
-            _logger.Trace($"BiDi RCV <-- {Encoding.UTF8.GetString(ms.ToArray())}");
+            _logger.Trace($"BiDi RCV <-- {Encoding.UTF8.GetString(data)}");
         }
 
-        var res = await JsonSerializer.DeserializeAsync(ms, typeof(T), jsonSerializerContext, cancellationToken).ConfigureAwait(false);
-
-        return (T)res!;
+        return data;
     }
 
-    public async Task SendAsJsonAsync<TCommand>(TCommand command, JsonSerializerContext jsonSerializerContext, CancellationToken cancellationToken)
-        where TCommand : Command
+    public async Task SendAsync(byte[] data, CancellationToken cancellationToken)
     {
-        var buffer = JsonSerializer.SerializeToUtf8Bytes(command, typeof(TCommand), jsonSerializerContext);
-
         await _socketSendSemaphoreSlim.WaitAsync(cancellationToken);
 
         try
         {
             if (_logger.IsEnabled(LogEventLevel.Trace))
             {
-                _logger.Trace($"BiDi SND --> {Encoding.UTF8.GetString(buffer)}");
+                _logger.Trace($"BiDi SND --> {Encoding.UTF8.GetString(data)}");
             }
 
-            await _webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, cancellationToken).ConfigureAwait(false);
+            await _webSocket.SendAsync(new ArraySegment<byte>(data), WebSocketMessageType.Text, true, cancellationToken).ConfigureAwait(false);
         }
         finally
         {