Device Simulator for Event Hubs

發行項
02/26/2015

This post will show how to build a simulator to send messages to Azure Event Hubs.

Background

I have been doing a lot of work with Event Hubs stream processing lately, and I find that I need different types of device simulators. I need one that sends a few messages, one that continuously sends a few messages, one that sends a lot of messages, and so on. I have probably 10 different console apps that do nothing more than send messages to Event Hubs in a loop, so I decided to make one simulator that fits all my needs.

This code adapts the code found in the Service Bus Event Hubs Getting Started sample.

The application makes it possible to simulate a number of devices sending X number of messages, with a sleep duration between each operation. You can run a finite number of times, or run continuously, and the application enables the end user to cancel the operation by pressing ENTER.

This is a console application built with .NET, but I could have just as easily created a Java application that uses JMS to send messages. I showed how to do this in the post Use JMS and Azure Event Hubs with Eclipse.

Reusing Existing Code

I created a library called EventHubDemo.Common that has two classes, MetricEvent and EventHubManager. Those classes are detailed in my post Scaling Azure Event Hubs Processing with Worker Roles as I use the same code between the sender (this post) and the receiver.

MetricEvent.cs

using System.Runtime.Serialization;
namespace EventHubDemo.Common.Contracts
{
[DataContract]
public class MetricEvent
{
[DataMember]
public int DeviceId { get; set; }
[DataMember]
public int Temperature { get; set; }
}
}

EventHubManager.cs

using Microsoft.ServiceBus;
using Microsoft.ServiceBus.Messaging;
using System;
using System.Diagnostics;
namespace EventHubDemo.Common.Utility
{
public class EventHubManager
{
public static string GetServiceBusConnectionString()
{
string connectionString = Microsoft.WindowsAzure.CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");
if (string.IsNullOrEmpty(connectionString))
{
Trace.WriteLine("Did not find Service Bus connections string in appsettings (app.config)");
return string.Empty;
}
ServiceBusConnectionStringBuilder builder = new ServiceBusConnectionStringBuilder(connectionString);
builder.TransportType = TransportType.Amqp;
return builder.ToString();
}
public static NamespaceManager GetNamespaceManager()
{
return NamespaceManager.CreateFromConnectionString(GetServiceBusConnectionString());
}
public static NamespaceManager GetNamespaceManager(string connectionString)
{
return NamespaceManager.CreateFromConnectionString(connectionString);
}
public static void CreateEventHubIfNotExists(string eventHubName, int numberOfPartitions, NamespaceManager manager)
{
try
{
// Create the Event Hub
Trace.WriteLine("Creating Event Hub...");
EventHubDescription ehd = new EventHubDescription(eventHubName);
ehd.PartitionCount = numberOfPartitions;
manager.CreateEventHubIfNotExistsAsync(ehd).Wait();
}
catch (AggregateException agexp)
{
Trace.WriteLine(agexp.Flatten());
}
}
}
}

Creating the Console Application

I created a new console application named Sender. I added a reference to my EventHubDemo.Common library, and added NuGet packages for Service Bus and Json.NET.

Realizing that others might use this program, I wanted to give a little documentation at runtime. I created a class MyArgs.cs that is responsible for parsing arguments, providing help, and providing current values.

MyArgs.cs

using System;
using System.Text;
namespace EventHubDemo.Sender
{
public class MyArgs
{
public MyArgs()
{
}
public string eventHubName { get; set; }
public int numberOfDevices { get; set; }
public int numberOfMessages { get; set; }
public int numberOfPartitions { get; set; }
public int sleepSeconds { get; set; }
public int iterations { get; set; }
public void ParseArgs(string[] args)
{
if (args.Length != 6)
{
throw new ArgumentException("Incorrect number of arguments. Expected 6 args <eventhubname> <NumberOfDevices> <NumberOfMessagesToSend> <NumberOfPartitions> <sleepSeconds> <iterations>", args.ToString());
}
else
{
eventHubName = args[0];
numberOfDevices = Int32.Parse(args[1]);
numberOfMessages = Int32.Parse(args[2]);
numberOfPartitions = Int32.Parse(args[3]);
sleepSeconds = int.Parse(args[4]);
iterations = int.Parse(args[5]);
}
}
internal string GetHelp()
{
StringBuilder sb = new StringBuilder();
sb.AppendLine("Usage: Sender.exe EventHubName NumberOfDevices NumberOfMessagesToSend NumberOfPartitions SleepSeconds Iterations");
sb.AppendLine();
sb.AppendLine("Parameters:");
sb.AppendLine("\tEventHubName:\t\tName of the Event Hub to send messages to");
sb.AppendLine("\tNumberOfDevices:\t\tNumber of devices to simulate");
sb.AppendLine("\tNumberOfMessagesToSend:\t\tNumber of messages to send");
sb.AppendLine("\tNumberOfPartitions:\t\tNumber of Event Hub partitions");
sb.AppendLine("\tSleepSeconds:\t\tNumber of seconds to sleep between iterations (0 to send as fast as possible)");
sb.AppendLine("\tIterations:\t\tNumber of iterations (-1 to continuously send)");
sb.AppendLine();
return sb.ToString();
}
public override string ToString()
{
StringBuilder sb = new StringBuilder();
sb.AppendLine("Event Hub name: " + eventHubName);
sb.AppendLine("Number of devices: " + numberOfDevices);
sb.AppendLine("Number of messages: " + numberOfMessages);
sb.AppendLine("Number of partitions: " + numberOfPartitions);
sb.AppendLine("Seconds to sleep between iterations: " + sleepSeconds);
sb.AppendLine("Number of iterations: " + iterations);
return sb.ToString();
}
}
}

I then adapted the code from Service Bus Event Hubs Getting Started to create a class responsible for sending messages to Event Hub.

Sender.cs

using EventHubDemo.Common.Contracts;
using Microsoft.ServiceBus.Messaging;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Text;
using System.Threading.Tasks;
namespace EventHubDemo.Sender
{
public class Sender
{
private string eventHubName;
private int numberOfDevices;
private int numberOfMessages;
public Sender(string eventHubName, int numberOfDevices, int numberOfMessages)
{
this.eventHubName = eventHubName;
this.numberOfDevices = numberOfDevices;
this.numberOfMessages = numberOfMessages;
}
public void SendEvents()
{
// Create EventHubClient
EventHubClient client = EventHubClient.Create(this.eventHubName);
try
{
List<Task> tasks = new List<Task>();
// Send messages to Event Hub
Trace.TraceInformation("Sending messages to Event Hub " + client.Path);
Random random = new Random();
for (int i = 0; i < this.numberOfMessages; ++i)
{
// Create the device/temperature metric
MetricEvent info = new MetricEvent() { DeviceId = random.Next(numberOfDevices), Temperature = random.Next(100) };
var serializedString = JsonConvert.SerializeObject(info);
EventData data = new EventData(Encoding.UTF8.GetBytes(serializedString));
// Set user properties if needed
data.Properties.Add("Type", "Telemetry_" + DateTime.Now.ToLongTimeString());
OutputMessageInfo("SENDING: ", data, info);
// Send the metric to Event Hub
tasks.Add(client.SendAsync(data));
}
;
Task.WaitAll(tasks.ToArray());
}
catch (Exception exp)
{
Trace.TraceError("Error on send: " + exp.Message);
}
client.CloseAsync().Wait();
}
static void OutputMessageInfo(string action, EventData data, MetricEvent info)
{
if (data == null)
{
return;
}
if (info != null)
{
Trace.TraceInformation("{0}: Device {1}, Temperature {2}.", action, info.DeviceId, info.Temperature);
}
}
}
}

I’ve used a variation of this Sender class in a number of ways, using various loops to simulate. What I finally settled on is to call the Sender.cs class and using a little bit of threading code.

Program.cs

using EventHubDemo.Common.Utility;
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
namespace EventHubDemo.Sender
{
class Program
{
protected ManualResetEvent runCompleteEvent = new ManualResetEvent(false);
protected CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
private MyArgs a;
public Program(MyArgs args)
{
a = args;
}
static void Main(string[] args)
{
MyArgs a = new MyArgs();
try
{
a.ParseArgs(args);
}
catch
{
Console.WriteLine(a.GetHelp());
return;
}
Trace.TraceInformation(a.ToString());
Program p = new Program(a);
var token = p.cancellationTokenSource.Token;
Task.Factory.StartNew(() => p.Run(token, a));
Task.Factory.StartNew(() => p.WaitForEnter());
p.runCompleteEvent.WaitOne();
}
protected void WaitForEnter()
{
Console.WriteLine("Press enter key to stop worker.");
Console.ReadLine();
cancellationTokenSource.Cancel();
}
private void Run(CancellationToken token, MyArgs a)
{
if (a.iterations == -1)
{
//Continuously iterate
while (!token.IsCancellationRequested)
{
SendMessages(a);
//Convert to milliseconds
Thread.Sleep(a.sleepSeconds * 1000);
}
runCompleteEvent.Set();
}
else
{
//Iterate a finite number of times, enabling the user
// to cancel the operation
for (int i = 0; i < a.iterations; i++)
{
if (!token.IsCancellationRequested)
{
SendMessages(a);
//Convert to milliseconds
Thread.Sleep(a.sleepSeconds * 1000);
}
else
{
break;
}
}
runCompleteEvent.Set();
}
}
private static void SendMessages(MyArgs a)
{
var namespaceManager = EventHubManager.GetNamespaceManager();
EventHubManager.CreateEventHubIfNotExists(a.eventHubName, a.numberOfPartitions, namespaceManager);
Sender s = new Sender(a.eventHubName, a.numberOfDevices, a.numberOfMessages);
Trace.TraceInformation("Sending events");
s.SendEvents();
}
}
}

Our program’s entry point is, of course, the static Main method. We parse the arguments and then reference the cancellation token. This allows us to check for token.IsCancellationRequested to see if the user cancelled the current operation. We then use a ManualResetEvent to wait for a signal that the application is complete. This allows us to run for a single iteration and then terminate, or to continuously iterate.

When the Run method is called, we check to see if the user wants to run for a finite number of times or to continuously execute. In both cases, we enable the user to cancel the operation.

Side note: See Lync and Outlook teams? The white screen of death is avoidable.

We call the Service Bus library to send messages to the Event Hub, and sleep for X number of seconds between iterations.

App.Config

This part is also worth mentioning… we store the connection string for Service Bus and the Azure Storage Account connection string in app.config. I also added the System.Diagnostics.ConsoleTraceListener so that all Trace output messages appear in the Console window.

Code Snippet

<appSettings>
<add key="Microsoft.ServiceBus.ConnectionString"
value="Endpoint=sb://kirkevans.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=REDACTED=" />
<add key="AzureStorageConnectionString"
value="DefaultEndpointsProtocol=https;AccountName=kirkestorage;AccountKey=REDACTED=="/>
</appSettings>
<system.diagnostics>
<trace autoflush="true">
<listeners>
<add name="consoleListener"
type="System.Diagnostics.ConsoleTraceListener"/>
</listeners>
</trace>
</system.diagnostics>

Debugging With Visual Studio

To enable debugging in Visual Studio, you need to provide the command-line arguments. Realizing that you might not have had to write a console application in a few years, just go to the project’s properties and set the command line arguments on the Debug tab.

The command line arguments I used were:

devicereadings 1000 5 16 1 –1

The Event Hub name is “devicereadings”, I am simulating 1000 devices, sending 5 messages per batch. There are 16 partitions used for the Event Hub, and we sleep for 1 second between batches. The program runs continuously until the user terminates.

The Result

We start the Sender using our desired parameters.

To see if it is actually working, I run a worker role that processes the Event Hub streams in the local compute emulator (details on how to create this are in the post Scaling Azure Event Hubs Processing with Worker Roles).

Summary

This post shows how to create a simple device simulator to send messages to Azure Event Hubs. I realize that using Console Applications is not the most interesting demo in the world, and it would be nice to do something with this data besides just dumping it to Console output. We’ll look at that in some upcoming posts.

For More Information

Service Bus Event Hubs Getting Started

Scaling Azure Event Hubs Processing with Worker Roles

Use JMS and Azure Event Hubs with Eclipse

Comments

Anonymous
November 20, 2015
Any chance to bundle this into an exe?
- Anonymous
  September 12, 2016
  Where Can I find the AccountName and the Account Key ?I keep on getting the error "No valid combination of account information found"
Anonymous
September 12, 2016
Where Can I find the AccountName and the Account Key ?I keep on getting the error "No valid combination of account information found"

共用方式為