Moar Websocket Tests

pglombardo · pglombardo · commit 37c10ff896b5 · 2024-08-21T14:59:06.000+02:00
diff --git a/Source/HiveMQtt/Client/Transport/WebSocketTransport.cs b/Source/HiveMQtt/Client/Transport/WebSocketTransport.cs
@@ -16,8 +16,8 @@
 namespace HiveMQtt.Client.Transport;
 
 using HiveMQtt.Client.Options;
+using System.Buffers;
 using System.Net.WebSockets;
-using System.Runtime.InteropServices;
 
 public class WebSocketTransport : BaseTransport, IDisposable
 {
@@ -29,13 +29,22 @@ public WebSocketTransport(HiveMQClientOptions options)
     {
         this.Options = options;
         this.Socket = new ClientWebSocket();
+
+        Uri uri = new Uri(this.Options.WebSocketServer);
+        this.Options.Host = uri.Host;
+        this.Options.Port = uri.Port;
+
+        if (uri.Scheme is not "ws" and not "wss")
+        {
+            throw new ArgumentException("Invalid WebSocket URI scheme");
+        }
     }
 
     public override async Task<bool> ConnectAsync(CancellationToken cancellationToken = default)
     {
         try
         {
-            await this.Socket.ConnectAsync(new Uri(this.Options.Host), cancellationToken).ConfigureAwait(false);
+            await this.Socket.ConnectAsync(new Uri(this.Options.WebSocketServer), cancellationToken).ConfigureAwait(false);
         }
         catch (WebSocketException ex)
         {
@@ -77,19 +86,24 @@ public override async Task<bool> WriteAsync(byte[] buffer, CancellationToken can
         return true;
     }
 
-    public override Task<TransportReadResult> ReadAsync(CancellationToken cancellationToken = default)
+    public override async Task<TransportReadResult> ReadAsync(CancellationToken cancellationToken = default)
     {
-        throw new NotImplementedException();
+        var buffer = new ArraySegment<byte>(new byte[1024]);
+        var result = await this.Socket.ReceiveAsync(buffer, cancellationToken).ConfigureAwait(false);
+
+        Logger.Trace($"Received {result.Count} bytes");
+
+        return new TransportReadResult(new ReadOnlySequence<byte>(buffer));
     }
 
     public override void AdvanceTo(SequencePosition consumed)
     {
-        throw new NotImplementedException();
+        Logger.Error("WebSocketTransport.AdvanceTo() not implemented");
     }
 
     public override void AdvanceTo(SequencePosition consumed, SequencePosition examined)
     {
-        throw new NotImplementedException();
+        Logger.Error("WebSocketTransport.AdvanceTo() not implemented");
     }
 
     /// <summary>
diff --git a/Tests/HiveMQtt.Test/HiveMQClient/WebSocket/ConnectTest.cs b/Tests/HiveMQtt.Test/HiveMQClient/WebSocket/ConnectTest.cs
@@ -15,7 +15,7 @@ public class ConnectTest
     public async Task ConnectAndDisconnectAsync()
     {
         var options = new HiveMQClientOptionsBuilder()
-            .WithWebSocketServer("ws://localhost:8080/mqtt")
+            .WithWebSocketServer("ws://localhost:8000/mqtt")
             .WithClientId("test")
             .Build();
 
@@ -26,4 +26,242 @@ public async Task ConnectAndDisconnectAsync()
 
         await client.DisconnectAsync().ConfigureAwait(false);
     }
+
+    [Fact]
+    public async Task BasicConnectAndDisconnectAsync()
+    {
+        var options = new HiveMQClientOptionsBuilder()
+                            .WithWebSocketServer("ws://localhost:8000/mqtt")
+                            .WithClientId("BasicConnectAndDisconnectAsync").Build();
+        var client = new HiveMQClient(options);
+        Assert.NotNull(client);
+
+        var connectResult = await client.ConnectAsync().ConfigureAwait(false);
+
+        Assert.True(connectResult.ReasonCode == ConnAckReasonCode.Success);
+        Assert.True(client.IsConnected());
+
+        var disconnectOptions = new DisconnectOptions();
+        Assert.Equal(DisconnectReasonCode.NormalDisconnection, disconnectOptions.ReasonCode);
+
+        var disconnectResult = await client.DisconnectAsync(disconnectOptions).ConfigureAwait(false);
+        Assert.True(disconnectResult);
+        Assert.False(client.IsConnected());
+
+        client.Dispose();
+    }
+
+    [Fact]
+    public async Task RaiseOnFailureToConnectAsync()
+    {
+        // Bad port number
+        var clientOptions = new HiveMQClientOptionsBuilder()
+                            .WithWebSocketServer("ws://localhost:0/mqtt")
+                            .WithClientId("RaiseOnFailureToConnectAsync").Build();
+        var client = new HiveMQClient(clientOptions);
+
+        await Assert.ThrowsAsync<HiveMQttClientException>(client.ConnectAsync).ConfigureAwait(false);
+
+        client.Dispose();
+    }
+
+    [Fact]
+    public async Task TestConnectEventsAsync()
+    {
+        var options = new HiveMQClientOptionsBuilder()
+                            .WithWebSocketServer("ws://localhost:8000/mqtt")
+                            .WithClientId("TestConnectEventsAsync").Build();
+        var client = new HiveMQClient(options);
+
+        // Client Events
+        client.BeforeConnect += BeforeConnectHandler;
+        client.AfterConnect += AfterConnectHandler;
+        client.BeforeDisconnect += BeforeDisconnectHandler;
+        client.AfterDisconnect += AfterDisconnectHandler;
+
+        // Packet Events
+        client.OnConnectSent += OnConnectSentHandler;
+        client.OnConnAckReceived += OnConnAckReceivedHandler;
+
+        // Set up TaskCompletionSource to wait for event handlers to finish
+        var taskCompletionSource = new TaskCompletionSource<bool>();
+        client.OnDisconnectSent += (sender, args) => taskCompletionSource.SetResult(true);
+
+        // Connect and Disconnect
+        var result = await client.ConnectAsync().ConfigureAwait(false);
+        await client.DisconnectAsync().ConfigureAwait(false);
+
+        // Wait for event handlers to finish
+        await taskCompletionSource.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
+        await Task.Delay(1000).ConfigureAwait(false);
+
+        // Assert that all Events were called
+        Assert.True(client.LocalStore.ContainsKey("BeforeConnectHandlerCalled"));
+        Assert.True(client.LocalStore.ContainsKey("AfterConnectHandlerCalled"));
+
+        Assert.True(client.LocalStore.ContainsKey("BeforeDisconnectHandlerCalled"));
+        Assert.True(client.LocalStore.ContainsKey("AfterDisconnectHandlerCalled"));
+
+        Assert.True(client.LocalStore.ContainsKey("OnConnectSentHandlerCalled"));
+        Assert.True(client.LocalStore.ContainsKey("OnConnAckReceivedHandlerCalled"));
+
+        // Remove event handlers
+        client.BeforeConnect -= BeforeConnectHandler;
+        client.AfterConnect -= AfterConnectHandler;
+        client.BeforeDisconnect -= BeforeDisconnectHandler;
+        client.AfterDisconnect -= AfterDisconnectHandler;
+
+        client.OnConnectSent -= OnConnectSentHandler;
+        client.OnConnAckReceived -= OnConnAckReceivedHandler;
+
+        client.Dispose();
+    }
+
+    [Fact]
+    public async Task Test_AfterDisconnectEvent_Async()
+    {
+        var options = new HiveMQClientOptionsBuilder()
+                            .WithWebSocketServer("ws://localhost:8000/mqtt")
+                            .WithClientId("Test_AfterDisconnectEvent_Async").Build();
+        var client = new HiveMQClient(options);
+
+        // Client Events
+        client.AfterDisconnect += AfterDisconnectHandler;
+
+        // Set up TaskCompletionSource to wait for event handlers to finish
+        var taskCompletionSource = new TaskCompletionSource<bool>();
+        client.OnDisconnectSent += (sender, args) => taskCompletionSource.SetResult(true);
+
+        // Connect and Disconnect
+        var result = await client.ConnectAsync().ConfigureAwait(false);
+        await client.DisconnectAsync().ConfigureAwait(false);
+
+        // Wait for event handlers to finish
+        await taskCompletionSource.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
+        await Task.Delay(1000).ConfigureAwait(false);
+
+        // Assert that all Events were called
+        Assert.True(client.LocalStore.ContainsKey("AfterDisconnectHandlerCalled"));
+        Assert.True(client.LocalStore.ContainsKey("AfterDisconnectHandlerCalledCount"));
+        Assert.Equal("1", client.LocalStore["AfterDisconnectHandlerCalledCount"]);
+
+        // Remove event handlers
+        client.AfterDisconnect -= AfterDisconnectHandler;
+
+        client.Dispose();
+    }
+
+    private static void BeforeConnectHandler(object? sender, BeforeConnectEventArgs eventArgs)
+    {
+        Assert.NotNull(sender);
+        if (sender is not null)
+        {
+            var client = (HiveMQClient)sender;
+            client.LocalStore.Add("BeforeConnectHandlerCalled", "true");
+        }
+
+        Assert.NotNull(eventArgs.Options);
+    }
+
+    private static void OnConnectSentHandler(object? sender, OnConnectSentEventArgs eventArgs)
+    {
+        Assert.NotNull(sender);
+        if (sender is not null)
+        {
+            var client = (HiveMQClient)sender;
+            client.LocalStore.Add("OnConnectSentHandlerCalled", "true");
+        }
+
+        Assert.NotNull(eventArgs.ConnectPacket);
+    }
+
+    private static void OnConnAckReceivedHandler(object? sender, OnConnAckReceivedEventArgs eventArgs)
+    {
+        Assert.NotNull(sender);
+        if (sender is not null)
+        {
+            var client = (HiveMQClient)sender;
+            client.LocalStore.Add("OnConnAckReceivedHandlerCalled", "true");
+        }
+
+        Assert.NotNull(eventArgs.ConnAckPacket);
+    }
+
+    private static void AfterConnectHandler(object? sender, AfterConnectEventArgs eventArgs)
+    {
+        Assert.NotNull(sender);
+        if (sender is not null)
+        {
+            var client = (HiveMQClient)sender;
+            client.LocalStore.Add("AfterConnectHandlerCalled", "true");
+        }
+
+        Assert.NotNull(eventArgs.ConnectResult);
+    }
+
+    private static void BeforeDisconnectHandler(object? sender, BeforeDisconnectEventArgs eventArgs)
+    {
+        Assert.NotNull(sender);
+        if (sender is not null)
+        {
+            var client = (HiveMQClient)sender;
+            client.LocalStore.Add("BeforeDisconnectHandlerCalled", "true");
+        }
+    }
+
+    private static void AfterDisconnectHandler(object? sender, AfterDisconnectEventArgs eventArgs)
+    {
+        Assert.NotNull(sender);
+        if (sender is not null)
+        {
+            var client = (HiveMQClient)sender;
+
+            if (client.LocalStore.TryGetValue("AfterDisconnectHandlerCalled", out var value))
+            {
+                var count = Convert.ToInt16(value, CultureInfo.InvariantCulture);
+                count++;
+                client.LocalStore.Add("AfterDisconnectHandlerCalledCount", count.ToString(CultureInfo.InvariantCulture));
+            }
+            else
+            {
+                client.LocalStore.Add("AfterDisconnectHandlerCalled", "true");
+                client.LocalStore.Add("AfterDisconnectHandlerCalledCount", "1");
+            }
+        }
+    }
+
+    private static void OnDisconnectSentHandler(object? sender, OnDisconnectSentEventArgs eventArgs)
+    {
+        Assert.NotNull(sender);
+        if (sender is not null)
+        {
+            var client = (HiveMQClient)sender;
+
+            if (client.LocalStore.TryGetValue("OnDisconnectSentHandlerCalled", out var value))
+            {
+                var count = Convert.ToInt16(value, CultureInfo.InvariantCulture);
+                count++;
+                client.LocalStore.Add("OnDisconnectSentHandlerCalledCount", count.ToString(CultureInfo.InvariantCulture));
+            }
+            else
+            {
+                client.LocalStore.Add("OnDisconnectSentHandlerCalled", "true");
+                client.LocalStore.Add("OnDisconnectSentHandlerCalledCount", "1");
+            }
+        }
+
+        Assert.NotNull(eventArgs.DisconnectPacket);
+    }
+
+    private static void OnDisconnectReceivedHandler(object? sender, OnDisconnectReceivedEventArgs eventArgs)
+    {
+        Assert.NotNull(sender);
+        if (sender is not null)
+        {
+            var client = (HiveMQClient)sender;
+            client.LocalStore.Add("OnDisconnectReceivedHandlerCalled", "true");
+        }
+
+        Assert.NotNull(eventArgs.DisconnectPacket);
+    }
 }
diff --git a/Tests/HiveMQtt.Test/HiveMQClient/WebSocket/PublishTest.cs b/Tests/HiveMQtt.Test/HiveMQClient/WebSocket/PublishTest.cs

Original file line number	Diff line number	Diff line change
`@@ -16,8 +16,8 @@`
`16`	`16`	`namespace HiveMQtt.Client.Transport;`
`17`	`17`
`18`	`18`	`using HiveMQtt.Client.Options;`
	`19`	`+using System.Buffers;`
`19`	`20`	`using System.Net.WebSockets;`
`20`		`-using System.Runtime.InteropServices;`
`21`	`21`
`22`	`22`	`public class WebSocketTransport : BaseTransport, IDisposable`
`23`	`23`	`{`
`@@ -29,13 +29,22 @@ public WebSocketTransport(HiveMQClientOptions options)`
`29`	`29`	`{`
`30`	`30`	`this.Options = options;`
`31`	`31`	`this.Socket = new ClientWebSocket();`
	`32`	`+`
	`33`	`+ Uri uri = new Uri(this.Options.WebSocketServer);`
	`34`	`+ this.Options.Host = uri.Host;`
	`35`	`+ this.Options.Port = uri.Port;`
	`36`	`+`
	`37`	`+ if (uri.Scheme is not "ws" and not "wss")`
	`38`	`+ {`
	`39`	`+ throw new ArgumentException("Invalid WebSocket URI scheme");`
	`40`	`+ }`
`32`	`41`	`}`
`33`	`42`
`34`	`43`	`public override async Task<bool> ConnectAsync(CancellationToken cancellationToken = default)`
`35`	`44`	`{`
`36`	`45`	`try`
`37`	`46`	`{`
`38`		`- await this.Socket.ConnectAsync(new Uri(this.Options.Host), cancellationToken).ConfigureAwait(false);`
	`47`	`+ await this.Socket.ConnectAsync(new Uri(this.Options.WebSocketServer), cancellationToken).ConfigureAwait(false);`
`39`	`48`	`}`
`40`	`49`	`catch (WebSocketException ex)`
`41`	`50`	`{`
`@@ -77,19 +86,24 @@ public override async Task<bool> WriteAsync(byte[] buffer, CancellationToken can`
`77`	`86`	`return true;`
`78`	`87`	`}`
`79`	`88`
`80`		`- public override Task<TransportReadResult> ReadAsync(CancellationToken cancellationToken = default)`
	`89`	`+ public override async Task<TransportReadResult> ReadAsync(CancellationToken cancellationToken = default)`
`81`	`90`	`{`
`82`		`- throw new NotImplementedException();`
	`91`	`+ var buffer = new ArraySegment<byte>(new byte[1024]);`
	`92`	`+ var result = await this.Socket.ReceiveAsync(buffer, cancellationToken).ConfigureAwait(false);`
	`93`	`+`
	`94`	`+ Logger.Trace($"Received {result.Count} bytes");`
	`95`	`+`
	`96`	`+ return new TransportReadResult(new ReadOnlySequence<byte>(buffer));`
`83`	`97`	`}`
`84`	`98`
`85`	`99`	`public override void AdvanceTo(SequencePosition consumed)`
`86`	`100`	`{`
`87`		`- throw new NotImplementedException();`
	`101`	`+ Logger.Error("WebSocketTransport.AdvanceTo() not implemented");`
`88`	`102`	`}`
`89`	`103`
`90`	`104`	`public override void AdvanceTo(SequencePosition consumed, SequencePosition examined)`
`91`	`105`	`{`
`92`		`- throw new NotImplementedException();`
	`106`	`+ Logger.Error("WebSocketTransport.AdvanceTo() not implemented");`
`93`	`107`	`}`
`94`	`108`
`95`	`109`	`/// <summary>`