chore: Convert from DataFlow to Channels

Aligns better with other parts of the project
buttplugio · Jun 25, 2023 · b75f9f1 · b75f9f1
1 parent e3ea88a
commit b75f9f1
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 12 deletions.
diff --git a/...Client.Connectors.WebsocketConnector/Buttplug.Client.Connectors.WebsocketConnector.csproj b/...Client.Connectors.WebsocketConnector/Buttplug.Client.Connectors.WebsocketConnector.csproj
@@ -25,7 +25,7 @@
 
   <ItemGroup>
     <PackageReference Include="deniszykov.WebSocketListener" Version="4.2.15" />
-    <PackageReference Include="System.Threading.Tasks.Dataflow" Version="7.0.0" />
+    <PackageReference Include="System.Threading.Channels" Version="7.0.0" />
   </ItemGroup>
 
   <ItemGroup>

diff --git a/Buttplug.Client.Connectors.WebsocketConnector/ButtplugWebsocketConnector.cs b/Buttplug.Client.Connectors.WebsocketConnector/ButtplugWebsocketConnector.cs
@@ -1,11 +1,9 @@
 using Buttplug.Core;
 
 using System;
-using System.Collections.Generic;
-using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-using System.Threading.Tasks.Dataflow;
+using System.Threading.Channels;
 using Buttplug.Core.Messages;
 using vtortola.WebSockets;
 using vtortola.WebSockets.Rfc6455;
@@ -20,7 +18,6 @@ public class ButtplugWebsocketConnector : ButtplugRemoteJSONConnector, IButtplug
         private WebSocketClient _wsClient;
 
         private WebSocket _ws;
-
         public bool Connected => _ws?.IsConnected == true;
 
         public event EventHandler Disconnected;
@@ -32,7 +29,7 @@ public class ButtplugWebsocketConnector : ButtplugRemoteJSONConnector, IButtplug
 
         private readonly Uri _uri;
 
-        private readonly BufferBlock<string> _outgoingMessages = new BufferBlock<string>();
+        private Channel<string> _channel = Channel.CreateBounded<string>(256);
 
         private Task _readTask;
 
@@ -116,7 +113,7 @@ public async Task DisconnectAsync(CancellationToken token = default)
         public async Task<ButtplugMessage> SendAsync(ButtplugMessage msg, CancellationToken token)
         {
             var (msgString, msgPromise) = PrepareMessage(msg);
-            await _outgoingMessages.SendAsync(msgString, token).ConfigureAwait(false);
+            await _channel.Writer.WriteAsync(msgString);
             return await msgPromise.ConfigureAwait(false);
         }
 
@@ -125,7 +122,7 @@ private async Task RunClientLoop(CancellationToken token)
             try
             {
                 var readTask = _ws.ReadStringAsync(token);
-                var writeTask = _outgoingMessages.OutputAvailableAsync(token);
+                var writeTask = _channel.Reader.ReadAsync(token).AsTask();
                 while (_ws.IsConnected && !token.IsCancellationRequested)
                 {
                     var msgTasks = new Task[]
@@ -150,15 +147,13 @@ private async Task RunClientLoop(CancellationToken token)
                     {
                         try
                         {
-                            IList<string> msgs = new List<string>();
-                            _outgoingMessages.TryReceiveAll(out msgs);
-                            var outMsgs = msgs.Aggregate(string.Empty, (current, msg) => current + msg);
+                            var outMsgs = await ((Task<string>)msgTasks[1]).ConfigureAwait(false);
                             if (_ws?.IsConnected == true)
                             {
                                 await _ws.WriteStringAsync(outMsgs, token).ConfigureAwait(false);
                             }
 
-                            writeTask = _outgoingMessages.OutputAvailableAsync(token);
+                            writeTask = _channel.Reader.ReadAsync(token).AsTask();
                         }
                         catch (WebSocketException e)
                         {