added perf test app

This commit is contained in:
JanEggers
2017-09-10 14:02:21 +02:00
parent 6b1a6c054e
commit a7a5eed0f2
3 changed files with 170 additions and 0 deletions
@@ -82,6 +82,7 @@
<Reference Include="System.Xml" />
</ItemGroup>
<ItemGroup>
<Compile Include="PerformanceTest.cs" />
<Compile Include="Program.cs" />
<Compile Include="Properties\AssemblyInfo.cs" />
</ItemGroup>
@@ -0,0 +1,163 @@
using MQTTnet.Core;
using MQTTnet.Core.Client;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol;
using MQTTnet.Core.Server;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace MQTTnet.TestApp.NetFramework
{
public class PerformanceTest
{
public static async Task RunAsync()
{
var server = Task.Run(() => RunServerAsync());
var client = Task.Run(() => RunClientAsync(50, TimeSpan.FromMilliseconds(10)));
await Task.WhenAll(server, client);
}
private static async Task RunClientAsync( int msgChunkSize, TimeSpan interval )
{
try
{
var options = new MqttClientOptions
{
Server = "localhost",
ClientId = "XYZ",
CleanSession = true
};
var client = new MqttClientFactory().CreateMqttClient(options);
client.ApplicationMessageReceived += (s, e) =>
{
};
client.Connected += async (s, e) =>
{
Console.WriteLine("### CONNECTED WITH SERVER ###");
await client.SubscribeAsync(new List<TopicFilter>
{
new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce)
});
Console.WriteLine("### SUBSCRIBED ###");
};
client.Disconnected += async (s, e) =>
{
Console.WriteLine("### DISCONNECTED FROM SERVER ###");
await Task.Delay(TimeSpan.FromSeconds(5));
try
{
await client.ConnectAsync();
}
catch
{
Console.WriteLine("### RECONNECTING FAILED ###");
}
};
try
{
await client.ConnectAsync();
}
catch (Exception exception)
{
Console.WriteLine("### CONNECTING FAILED ###" + Environment.NewLine + exception);
}
Console.WriteLine("### WAITING FOR APPLICATION MESSAGES ###");
var last = DateTime.Now;
var msgs = 0;
while (true)
{
for (int i = 0; i < msgChunkSize; i++)
{
var applicationMessage = new MqttApplicationMessage(
"A/B/C",
Encoding.UTF8.GetBytes("Hello World"),
MqttQualityOfServiceLevel.AtLeastOnce,
false
);
//do not await to send as much messages as possible
await client.PublishAsync(applicationMessage);
msgs++;
}
var now = DateTime.Now;
if (last < now - TimeSpan.FromSeconds(1))
{
Console.WriteLine( $"sending {msgs} inteded {msgChunkSize / interval.TotalSeconds}" );
msgs = 0;
last = now;
}
await Task.Delay(interval);
}
}
catch (Exception exception)
{
Console.WriteLine(exception);
}
}
private static void RunServerAsync()
{
try
{
var options = new MqttServerOptions
{
ConnectionValidator = p =>
{
if (p.ClientId == "SpecialClient")
{
if (p.Username != "USER" || p.Password != "PASS")
{
return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
}
}
return MqttConnectReturnCode.ConnectionAccepted;
}
};
var mqttServer = new MqttServerFactory().CreateMqttServer(options);
var last = DateTime.Now;
var msgs = 0;
mqttServer.ApplicationMessageReceived += (sender, args) =>
{
msgs++;
var now = DateTime.Now;
if (last < now - TimeSpan.FromSeconds(1))
{
Console.WriteLine($"received {msgs}");
msgs = 0;
last = now;
}
};
mqttServer.Start();
Console.WriteLine("Press any key to exit.");
Console.ReadLine();
mqttServer.Stop();
}
catch (Exception e)
{
Console.WriteLine(e);
}
Console.ReadLine();
}
}
}
@@ -19,6 +19,7 @@ namespace MQTTnet.TestApp.NetFramework
Console.WriteLine("MQTTnet - TestApp.NetFramework");
Console.WriteLine("1 = Start client");
Console.WriteLine("2 = Start server");
Console.WriteLine("3 = Start performance test");
var pressedKey = Console.ReadKey(true);
if (pressedKey.Key == ConsoleKey.D1)
{
@@ -30,6 +31,11 @@ namespace MQTTnet.TestApp.NetFramework
Task.Run(() => RunServerAsync(args));
Thread.Sleep(Timeout.Infinite);
}
else if (pressedKey.Key == ConsoleKey.D3)
{
Task.Run(() => PerformanceTest.RunAsync());
Thread.Sleep(Timeout.Infinite);
}
}
private static async Task RunClientAsync(string[] arguments)