Using an Azure Service bus queue for two-way (IoT) communication

 

Part 2 of Reading temperatures & controlling a fan with a RP2, Azure Service Bus and a Microsoft Band

Intro
In this post I will explain how the Windows 10 device to which the Band is paired and the Raspberry PI2 'talk' to each other. As the previous post (and the title of this post) already shows, this is done using an Azure Service Bus queue. The code for this is in two of the five projects in the demo solution

• TemperatureReader.Shared contains the data objects shuttled over the Azure Service Bus queue, as well as all the global definitions
• TemperatureReader.ServiceBus contains the actual code for the Service Bus code.

Getting started
Before you can get the code working, you will need to define an Azure Service Bus, because you cannot use mine - I like to keep it for demos myself and keep my Azure bill reasonable, thank you :). If you look into the TemperatureReader.Shared project in the demo solution you will find a class "Settings" that, amongst other things, has the following pretty long and rather cryptic definition:

public static readonly string TemperatureBusConnectionString =
"Endpoint=sb://yournamespacehere.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXXXXXXYOURKEYHEREXXXXXXXXXXXXXXXX";

To be able to use queues in a Service Bus, you must first claim a namespace. Doing so currently is only supported by the 'old' Azure Portal, so you need to go there to create it:


You need to select the tab "Service Bus" below, then hit the "Create" button. Then you get the pop-up as displayed on the right. Make up a nice namespace name (Azure will tell you if it's available or not), then select a region. As I am living in the Netherlands, I typically choose "West Europe" but feel free to choose otherwise. The rest of the settings can be left as they are - although you might consider setting the message tier to 'basic'. Hit the check mark, and you get the image below.

 

 

 

 

Your Azure Service bus namespace has been been created. This is stage 1 - now the actual queues need to be created. From the code in Settings.cs you can see there will need to be two queues on this Service Bus:

public static readonly string TemperatureQueue = "temperaturedatabus";
public static readonly int TemperatureQueueTtl = 10;

public static readonly string FanSwitchQueue = "fanswitchcommandbus";
public static readonly int FanSwitchQueueTtl = 10;

The first one is for sending temperature data from the PI2 to the Windows 10 device (and ultimately the Band) and the second one to send commands to switch the fan on and off. I have set the time to live (TTL) for messages on each queue at 10 seconds, meaning that any data older than 10 seconds can be discarded if it's not picked up. You can go ahead and click the arrow that I emphasized using a red box next to "thisisademo" and configure the queue manually, or just let the code (see below) to take care of it. There is only one more thing you will need to do: select the Azure Service Bus namespace that you just have created, then hit "Connection Information" at the bottom. By hovering over the text that starts with "Endpoint=sb://" you get a blue "copy" icon at the right and if you click that, that whole connection string gets put into the clipboard for your convenience.Then you can paste that into the TemperatureBusConnectionString constant in Settings.cs

A base queue client
In any project that you want to use Azure Service Bus in, you will need to use the NuGet package "WindowsAzure.Messaging.Managed" (assuming your are using a managed language like C# or VB.NET).

Then it's important to know that any client (and with that I mean an object of type Microsoft.WindowsAzure.Messaging.Queue) can both put objects in the queue as well as read them - and doing so is pretty stupid as I found out the hard way. A Queue object that does both basically eats his own messages - most of the time, but not all of the time. Happy debugging. Anyway, I wanted a queue client that

• I could either ask to read from the queue or put data on the queue, but not both
• could only send or receive data of a specified type
• would automatically create a queue on the service bus if it does not exists

I came up with this:

using System;
using System.Diagnostics;
using System.Threading.Tasks;
using Microsoft.WindowsAzure.Messaging;

namespace TemperatureReader.ServiceBus
{
  public class QueueClient<T> : IQueueClient<T>
  {
    private readonly string _queueName;
    private readonly string _connectionString;
    private readonly int _messageTimeToLive;
    private readonly QueueMode _queueMode;
    private Queue _queue;
    
    public QueueClient(string queueName, string connectionString, 
      QueueMode mode = QueueMode.Send, int messageTimeToLive = 10)
    {
      _queueName = queueName;
      _connectionString = connectionString;
      _queueMode = mode;
      _messageTimeToLive = messageTimeToLive;
    }

    public async virtual Task Start()
    {
      try
      {
        var settings = new QueueSettings 
          { DefaultMessageTimeToLive = TimeSpan.FromSeconds(_messageTimeToLive) };
        await Queue.CreateAsync(_queueName, _connectionString, settings);
        Debug.WriteLine($"Queue {_queueName} created");
      }
      catch (Exception)
      {
        Debug.WriteLine($"Queue {_queueName} already exists");
      }

      _queue = new Queue(_queueName, _connectionString);
      if (_queueMode == QueueMode.Listen)
      {
        _queue.OnMessage(message =>
        {
          var data = message.GetBody<T>();
          OnDataReceived?.Invoke(this, data);
        });
      }
    }

    public virtual async Task PostData(T tData)
    {
      if (this._queueMode == QueueMode.Send)
      {
        await _queue.SendAsync(tData);
      }
      else
      {
        throw new ArgumentException(
          "Cannot send data using a QueueMode.Listen client");
      }
    }

    public void Stop()
    {
      if (_queue != null)
      {
        _queue.Dispose();
        _queue = null;
      }
    }

    public event EventHandler<T> OnDataReceived;
  }
}

If a QueueClient object is created, it accepts the name of the queue, the queue connection string (that impossible long string in Settings.TemperatureBpusConnectionString), a message time-to-live and a QueueMode (can be either Send or Listen). Notice it has a type parameter - that is the type of object you are allowed to put on the queue - or can expect to come off it.

To get the thing going, you will need to call Start. This first tries to create the queue. Unfortunately I have not been able to find out how to check if a queue exists first, so I have resorted to the rather crude method of swallowing an exception if creating the queue fails - and assume that the exception was caused by an already existing queue. Then it will create an actual Queue object (Microsoft.WindowsAzure.Messaging) - and if it's a Listen QueueClient, it will attach an anonymous method to the OnMessage method of that queue. That method will hoist the data object from the queue and pass it on to whoever is subscribed to the OnDataReceived event of the QueueClient itself.

The QueueClient's post method is basically a typed wrapper on the Queue PostData method, except that it will prohibit from posting if this is a Listen client - to prevent it swallowing it's own message.

Finally there is the Stop method, that just disposes the queue.

Using specific implementations
Both TemperatureQueueClient and FanSwitchQueueClient are child classes, providing specific implementations of a queue client. For instance, TemperatureClient just handles temperature data and all you have to do is provide an override constructor to prevent you from having to set all that properties every time you use them:

using TemperatureReader.Shared;

namespace TemperatureReader.ServiceBus
{
  public class TemperatureQueueClient : QueueClient<TemperatureData>
  {
    public TemperatureQueueClient(QueueMode mode = QueueMode.Listen) :
      base(Settings.TemperatureQueue, 
        Settings.TemperatureBusConnectionString, mode, 
        Settings.TemperatureQueueTtl)
    {
    }
  }
}

The only thing there is left to set is the QueueMode, since you typically want to create one listening and one sending queue client.

If you now look in the MainPage_Loaded in MainPage.Xaml.cs in TemperatureReader (the IoT project) you will see a TempeatureQueueClient, processing temperature data, being set up as Send queue client while a FanSwitchQueueClient is a Listen queue client - which makes sense, as the Raspberry PI2 sends temperature data, but listens for commands to switch on or off the fan (ultimately coming from a Microsoft Band, somewhere in the world).

private async void MainPage_Loaded(object sender, Windows.UI.Xaml.RoutedEventArgs e)
{
  // stuff omitted
  var poster = new TemperatureQueueClient(QueueMode.Send);
  var fanCommandListener = new FanSwitchQueueClient(QueueMode.Listen);

In the client app that is intended for a Windows 10 device to which the Band is paired (typically a phone) it's the other way around. In MainViewModel.CreateNew you will see

public static MainViewModel CreateNew()
{
  var fanStatusPoster = new FanSwitchQueueClient(QueueMode.Send);

Indicating the queue client FanSwitchQueueClient is sending here, while the TempeatureQueueClient - embedded in TemperatureListener - is created in the constructor as a Listen client:

public TemperatureListener() 
{ 
  _client = new TemperatureQueueClient(QueueMode.Listen); 
  _client.OnDataReceived += ProcessTemperatureData; 
}

Conclusion
An Azure Service Bus client is a great way for simple one-to-one real time connections that can allow for some lag to happen between connected devices. It's very simple to use, can partly be created from code, is secured by default and does no require you to deploy a separate website to Azure to host a connection hub - the queue is hosted on Azure itself. And the awesome thing is - the code runs on both a Phone and a Raspberry PI2. It's in the same assembly. That is the power of the Universal Windows Platform.

Next time I will show how to actually measure temperature on a Raspberry PI2.