Skip to content

Commit

Permalink
Fixing QoS 2 and adding test application with test protocol (#301)
Browse files Browse the repository at this point in the history
  • Loading branch information
@Ellerbach
Ellerbach authored Oct 25, 2022
1 parent fa7af0a commit e37ac02
Show file tree
Hide file tree
Showing 8 changed files with 584 additions and 29 deletions.
77 changes: 48 additions & 29 deletions M2Mqtt/MqttClient.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ namespace nanoFramework.M2Mqtt
/// <summary>
/// MQTT Client
/// </summary>
public class MqttClient
public class MqttClient : IDisposable
{
// broker hostname (or ip address) and port
private string _brokerHostName;
Expand Down Expand Up @@ -530,41 +530,57 @@ public void Disconnect()
/// <summary>
/// Close client
/// </summary>
private void Close()
public void Close()
{
// stop receiving thread
_isRunning = false;

// wait end receive event thread
if (_receiveEventWaitHandle != null)
try
{
_receiveEventWaitHandle.Set();
}
// stop receiving thread
_isRunning = false;

// wait end process inflight thread
if (_inflightWaitHandle != null)
{
_inflightWaitHandle.Set();
}
// wait end receive event thread
if (_receiveEventWaitHandle != null)
{
_receiveEventWaitHandle.Set();
}

// unlock keep alive thread and wait
_keepAliveEvent.Set();
// wait end process inflight thread
if (_inflightWaitHandle != null)
{
_inflightWaitHandle.Set();
}

if (_keepAliveEventEnd != null)
{
_keepAliveEventEnd.WaitOne();
}
// unlock keep alive thread and wait
_keepAliveEvent.Set();

// clear all queues
_inflightQueue.Clear();
_internalQueue.Clear();
_waitingForAnswer.Clear();
_eventQueue.Clear();
if (_keepAliveEventEnd != null)
{
// We're waiting a bit but still we will exit not to block
_keepAliveEventEnd.WaitOne(1000, true);
}

// clear all queues
_inflightQueue.Clear();
_internalQueue.Clear();
_waitingForAnswer.Clear();
_eventQueue.Clear();

// close network channel
_channel.Close();

// close network channel
_channel.Close();
IsConnected = false;

IsConnected = false;
}
catch
{
// We are doing our best to close everything, even if there are issues
}
}

/// <inheritdoc/>
public void Dispose()
{
OnConnectionClosing();
Close();
}

/// <summary>
Expand Down Expand Up @@ -990,7 +1006,7 @@ private bool EnqueueInflight(MqttMsgBase msg, MqttMsgFlow flow)
// check number of messages inside inflight queue
enqueue = (_inflightQueue.Count < _settings.InflightQueueSize);
#if DEBUG
Debug.WriteLine($"_inflightQueue.Count {_inflightQueue.Count }");
Debug.WriteLine($"_inflightQueue.Count {_inflightQueue.Count}");
#endif
}

Expand Down Expand Up @@ -1131,6 +1147,7 @@ private void EnqueueInternal(MqttMsgBase msg)
MqttMsgContext msgCtx = null;
lock (_inflightQueue)
{
enqueue = true;
msgCtx = (MqttMsgContext)_inflightQueue.Get(msgCtxFinder.Find);
}
// the PUBLISH message isn't in the inflight queue, it was already sent so we need to ignore this PUBREC
Expand Down Expand Up @@ -1441,7 +1458,9 @@ private void DispatchEventThread()
lock (_eventQueue)
{
if (_eventQueue.Count > 0)
{
internalEvent = (InternalEvent)_eventQueue.Dequeue();
}
}

// it's an event with a message inside
Expand Down
64 changes: 64 additions & 0 deletions Tests/MemoryLeakTestApp/MemoryLeakTestApp.nfproj
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
<?xml version="1.0" encoding="utf-8"?>
<Project ToolsVersion="Current" DefaultTargets="Build" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
<PropertyGroup Label="Globals">
<NanoFrameworkProjectSystemPath>$(MSBuildExtensionsPath)\nanoFramework\v1.0\</NanoFrameworkProjectSystemPath>
</PropertyGroup>
<Import Project="$(NanoFrameworkProjectSystemPath)NFProjectSystem.Default.props" Condition="Exists('$(NanoFrameworkProjectSystemPath)NFProjectSystem.Default.props')" />
<PropertyGroup>
<Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
<Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
<ProjectTypeGuids>{11A8DD76-328B-46DF-9F39-F559912D0360};{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}</ProjectTypeGuids>
<ProjectGuid>fbdd4d19-7943-4666-bd31-a965a614588a</ProjectGuid>
<OutputType>Exe</OutputType>
<AppDesignerFolder>Properties</AppDesignerFolder>
<FileAlignment>512</FileAlignment>
<RootNamespace>MemoryLeakTestApp</RootNamespace>
<AssemblyName>MemoryLeakTestApp</AssemblyName>
<TargetFrameworkVersion>v1.0</TargetFrameworkVersion>
<RestorePackagesWithLockFile>true</RestorePackagesWithLockFile>
<RestoreLockedMode Condition="'$(TF_BUILD)' == 'True' or '$(ContinuousIntegrationBuild)' == 'True'">true</RestoreLockedMode>
</PropertyGroup>
<Import Project="$(NanoFrameworkProjectSystemPath)NFProjectSystem.props" Condition="Exists('$(NanoFrameworkProjectSystemPath)NFProjectSystem.props')" />
<ItemGroup>
<Compile Include="Program.cs" />
<Compile Include="Properties\AssemblyInfo.cs" />
</ItemGroup>
<ItemGroup>
<Reference Include="mscorlib">
<HintPath>..\..\packages\nanoFramework.CoreLibrary.1.12.0\lib\mscorlib.dll</HintPath>
</Reference>
<Reference Include="nanoFramework.Runtime.Events">
<HintPath>..\..\packages\nanoFramework.Runtime.Events.1.11.1\lib\nanoFramework.Runtime.Events.dll</HintPath>
</Reference>
<Reference Include="nanoFramework.Runtime.Native">
<HintPath>..\..\packages\nanoFramework.Runtime.Native.1.5.4\lib\nanoFramework.Runtime.Native.dll</HintPath>
</Reference>
<Reference Include="nanoFramework.System.Collections">
<HintPath>..\..\packages\nanoFramework.System.Collections.1.4.0\lib\nanoFramework.System.Collections.dll</HintPath>
</Reference>
<Reference Include="nanoFramework.System.Text">
<HintPath>..\..\packages\nanoFramework.System.Text.1.2.20\lib\nanoFramework.System.Text.dll</HintPath>
</Reference>
<Reference Include="System.IO.Streams">
<HintPath>..\..\packages\nanoFramework.System.IO.Streams.1.1.24\lib\System.IO.Streams.dll</HintPath>
</Reference>
<Reference Include="System.Net">
<HintPath>..\..\packages\nanoFramework.System.Net.1.10.34\lib\System.Net.dll</HintPath>
</Reference>
<Reference Include="System.Threading">
<HintPath>..\..\packages\nanoFramework.System.Threading.1.1.8\lib\System.Threading.dll</HintPath>
</Reference>
</ItemGroup>
<ItemGroup>
<None Include="packages.config" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\M2Mqtt\nanoFramework.M2Mqtt.nfproj" />
</ItemGroup>
<Import Project="$(NanoFrameworkProjectSystemPath)NFProjectSystem.CSharp.targets" Condition="Exists('$(NanoFrameworkProjectSystemPath)NFProjectSystem.CSharp.targets')" />
<ProjectExtensions>
<ProjectCapabilities>
<ProjectConfigurationsDeclaredAsItems />
</ProjectCapabilities>
</ProjectExtensions>
</Project>
166 changes: 166 additions & 0 deletions Tests/MemoryLeakTestApp/Program.cs
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,166 @@
using nanoFramework.M2Mqtt;
using nanoFramework.M2Mqtt.Messages;
using System;
using System.Diagnostics;
using System.Net.NetworkInformation;
using System.Reflection;
using System.Text;
using System.Threading;

namespace MemoryLeakTestApp
{
public class Program
{
private const string Brocker = "192.168.1.2";
private const string Ssid = "yourssid";
private const string Password = "you_wifi_password";
private const int NumberOfLoops = 5;
private static MqttClient client;
private static uint freeRam = 0;

public static void Main()
{
client = new MqttClient(Brocker);

string clientId = Guid.NewGuid().ToString();
client.Connect(clientId);

// Basic QoS tests
client.Publish("temp/test", Encoding.UTF8.GetBytes($"Message QoS 0"), MqttQoSLevel.AtMostOnce, false);
Thread.Sleep(200);
client.Publish("temp/test", Encoding.UTF8.GetBytes($"Message QoS 1"), MqttQoSLevel.AtLeastOnce, false);
Thread.Sleep(200);
client.Publish("temp/test", Encoding.UTF8.GetBytes($"Message QoS 2"), MqttQoSLevel.ExactlyOnce, false);
Thread.Sleep(200);

// Advance tests for different QoS
Publish(MqttQoSLevel.AtMostOnce);
Thread.Sleep(2000);
Publish(MqttQoSLevel.AtLeastOnce);
Thread.Sleep(2000);
Publish(MqttQoSLevel.ExactlyOnce);
Thread.Sleep(2000);

client.Publish("temp/test", Encoding.UTF8.GetBytes($"Memory left after all the test"), MqttQoSLevel.AtMostOnce, false);
freeRam = nanoFramework.Runtime.Native.GC.Run(true);
client.Publish("temp/free-ram", Encoding.UTF8.GetBytes(freeRam.ToString("F0")), MqttQoSLevel.AtMostOnce, false);
// Wait a bit
Thread.Sleep(5_000);
client.Publish("temp/test", Encoding.UTF8.GetBytes($"Memory left after all the test: 5s"), MqttQoSLevel.AtMostOnce, false);
freeRam = nanoFramework.Runtime.Native.GC.Run(true);
client.Publish("temp/free-ram", Encoding.UTF8.GetBytes(freeRam.ToString("F0")), MqttQoSLevel.AtMostOnce, false);
// Wait more
Thread.Sleep(35_000);
client.Publish("temp/test", Encoding.UTF8.GetBytes($"Memory left after all the test: 35s"), MqttQoSLevel.AtMostOnce, false);
freeRam = nanoFramework.Runtime.Native.GC.Run(true);
client.Publish("temp/free-ram", Encoding.UTF8.GetBytes(freeRam.ToString("F0")), MqttQoSLevel.AtMostOnce, false);
Thread.Sleep(120_000);
client.Publish("temp/test", Encoding.UTF8.GetBytes($"Memory left after all the test: 120s"), MqttQoSLevel.AtMostOnce, false);
freeRam = nanoFramework.Runtime.Native.GC.Run(true);
client.Publish("temp/free-ram", Encoding.UTF8.GetBytes(freeRam.ToString("F0")), MqttQoSLevel.AtMostOnce, false);
Thread.Sleep(Timeout.Infinite);

// Testing dispose
client.Dispose();
}

private static void Publish(MqttQoSLevel level)
{
client.Publish("temp/test", Encoding.UTF8.GetBytes($"single message QoS {level}"), level, false);
for (int i = 0; i < NumberOfLoops; i++)
{
freeRam = nanoFramework.Runtime.Native.GC.Run(true);
client.Publish("temp/free-ram", Encoding.UTF8.GetBytes($"{i}-{freeRam.ToString("F0")}"), level, false);
Thread.Sleep(1000);
}

client.Publish("temp/test", Encoding.UTF8.GetBytes($"two messages without delays QoS {level}"), level, false);
for (int i = 0; i < NumberOfLoops; i++)
{
freeRam = nanoFramework.Runtime.Native.GC.Run(true);
client.Publish("temp/free-ram", Encoding.UTF8.GetBytes($"0/{i}-{freeRam.ToString("F0")}"), level, false);
client.Publish("temp/free-ram", Encoding.UTF8.GetBytes($"1/{i}-{freeRam.ToString("F0")}"), level, false);
Thread.Sleep(1000);
}

client.Publish("temp/test", Encoding.UTF8.GetBytes($"two messages with delay QoS {level}"), level, false);
for (int i = 0; i < NumberOfLoops; i++)
{
freeRam = nanoFramework.Runtime.Native.GC.Run(true);
client.Publish("temp/free-ram", Encoding.UTF8.GetBytes($"0/{i}-{freeRam.ToString("F0")}"), level, false);
Thread.Sleep(50);
client.Publish("temp/free-ram", Encoding.UTF8.GetBytes($"1/{i}-{freeRam.ToString("F0")}"), level, false);
Thread.Sleep(1000);
}

client.Publish("temp/test", Encoding.UTF8.GetBytes($"Stress test with no delay QoS {level}"), level, false);
for (int i = 0; i < NumberOfLoops; i++)
{
freeRam = nanoFramework.Runtime.Native.GC.Run(true);
client.Publish("temp/free-ram", Encoding.UTF8.GetBytes($"{i}-{freeRam.ToString("F0")}"), level, false);
}
}

public static void SetupAndConnectNetwork()
{
NetworkInterface[] nis = NetworkInterface.GetAllNetworkInterfaces();
if (nis.Length > 0)
{
// get the first interface
NetworkInterface ni = nis[0];

if (ni.NetworkInterfaceType == NetworkInterfaceType.Wireless80211)
{
// network interface is Wi-Fi
Debug.WriteLine("Network connection is: Wi-Fi");

Wireless80211Configuration wc = Wireless80211Configuration.GetAllWireless80211Configurations()[ni.SpecificConfigId];
if (wc.Ssid != Ssid && wc.Password != Password)
{
// have to update Wi-Fi configuration
wc.Ssid = Ssid;
wc.Password = Password;
wc.SaveConfiguration();
}
else
{ // Wi-Fi configuration matches
}
}
else
{
// network interface is Ethernet
Debug.WriteLine("Network connection is: Ethernet");

ni.EnableDhcp();
}

// wait for DHCP to complete
WaitIP();
}
else
{
throw new NotSupportedException("ERROR: there is no network interface configured.\r\nOpen the 'Edit Network Configuration' in Device Explorer and configure one.");
}
}

static void WaitIP()
{
Debug.WriteLine("Waiting for IP...");

while (true)
{
NetworkInterface ni = NetworkInterface.GetAllNetworkInterfaces()[0];
if (ni.IPv4Address != null && ni.IPv4Address.Length > 0)
{
if (ni.IPv4Address[0] != '0')
{
Debug.WriteLine($"We have an IP: {ni.IPv4Address}");
break;
}
}

Thread.Sleep(500);
}
}
}
}
33 changes: 33 additions & 0 deletions Tests/MemoryLeakTestApp/Properties/AssemblyInfo.cs
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
using System.Reflection;
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices;

// General Information about an assembly is controlled through the following
// set of attributes. Change these attribute values to modify the information
// associated with an assembly.
[assembly: AssemblyTitle("CSharp.BlankApplication")]
[assembly: AssemblyDescription("")]
[assembly: AssemblyConfiguration("")]
[assembly: AssemblyCompany("")]
[assembly: AssemblyProduct("CSharp.BlankApplication")]
[assembly: AssemblyCopyright("Copyright © 2022")]
[assembly: AssemblyTrademark("")]
[assembly: AssemblyCulture("")]

// Setting ComVisible to false makes the types in this assembly not visible
// to COM components. If you need to access a type in this assembly from
// COM, set the ComVisible attribute to true on that type.
[assembly: ComVisible(false)]

// Version information for an assembly consists of the following four values:
//
// Major Version
// Minor Version
// Build Number
// Revision
//
// You can specify all the values or you can default the Build and Revision Numbers
// by using the '*' as shown below:
// [assembly: AssemblyVersion("1.0.*")]
[assembly: AssemblyVersion("1.0.0.0")]
[assembly: AssemblyFileVersion("1.0.0.0")]
Loading

0 comments on commit e37ac02

Please sign in to comment.