Few months ago Microsoft has announced set of new integration technologies under name LogicApps and API apps. This approach is still in preview, but it is truly promising. To give you a feeling of this approach in a bit bigger context (not only integration) I decided to demonstrate how to get some data from sensor which is driven by Windows Universal App running on Raspberry Pi2 with Windows 10. That sensor sends a telemetry data (temperature) to Service Bus queue (in real environment this would be EventHub or IotHub) by using of Service Bus Open SDK build by daenet GmbH in cooperation with University of Applied Sciences in Frankfurt am Main.
In this sample I will show how different technologies including hardware (Pi2 board) can very easily fit together. Following picture shows the high level architecture.
Sensor will send telemetry data to Service Bus Queue. One WebApplication which simulates WebDevice (device running in JS) will also send telemetry data to the same queue. API App connector will receive telemetry data from Service Bus, but connector can change the logic in the future.
The LogicApp will invoke the trigger in the API App (named ‘DeviceConnector’) and get telemetry via REST Get method. The powerful strategy behind logic apps is that it acts as orchestration (similar to BizTalk server) to get the data. If the way of how and where the telemetry data is posted is chnaged, we do not have to change anything in the LogicApp, as long the LogicApp itself does not deal with the content of the data.
At the end the app will simply forward every telemetry message (device connector can return a list of telemetry messages) to the HTTP endpoint by invoking POST request to Requestb.in (Service useful for testing).
The telemetry data is formatted as defined by following class:
| public class Command
{
public string DeviceId { get; set; }
public string DeviceName { get; set; }
public double Temperature { get; set; }
public string Timestamp { get; set; } |
Api App (‘DeviceConnector’)
First I will implement the API app, which will provide an operation called PollTrigger. Here is the source code of this operation.
| [HttpGet]
[Route("api/device/telemetry")]
public HttpResponseMessage TelemetryDataPollTrigger(
string triggerState = null,
string rootNamespace = null)
{
IDataChannel chn;
if (triggerState != null && triggerState.ToLower() == "mock")
chn = new MockChannel();
else
chn = new SbChannel();
var telemetryData = chn.GetTelemetryData();
if (telemetryData.Count > 0)
{
// Extension method provided by the AppService service SDK.
return this.Request.EventTriggered(new { Data = telemetryData });
}
else
{
// Extension method provided by the AppService service SDK.
return this.Request.EventWaitPoll(new TimeSpan(0, 1, 0));
}
}
|
As next, following method shows how to receive telemetry data. As you see this method will start collecting of telemetry data from Service Bus queue for next 30 seconds. That means, when LogicApp invokes trigger, it will block for 30 seconds. Not this is just an example. There are also other ways how to do this.
In this 30 seconds method GetTelemetryData will collect messages as they are arriving in to the ‘ingress’ queue.
| public List<Command> GetTelemetryData()
{
DateTime startTime = DateTime.Now;
List<Command> commands = new List<Command>();
var queueClient = QueueClient.CreateFromConnectionString(
m_SbConnStr, "ingress", ReceiveMode.ReceiveAndDelete);
while (DateTime.Now - startTime < TimeSpan.FromSeconds(30))
{
var msgs = queueClient.ReceiveBatch(100,
TimeSpan.FromSeconds(15));
if (msgs != null)
{
foreach (var msg in msgs)
{
try
{
Command cmd = msg.GetBody<Command>(new
DataContractJsonSerializer(typeof(Command)));
commands.Add(cmd);
}
catch (Exception) { };
}
}
}
return commands;
}
|
Logic App
As next we have to implement (configure) LogicApp, which will orchestrate two connectors :
- DeviceConnector for receiving of the data.
- HTTP Connector for posting of telemetry data to requestb.in.
First we nee to add DeviceConnector. Following picture shows the configuration of the trigger. I have option to invoke Mocked-Channel (see code above) by specifying ‘mock’ as triggerState. By entering ‘sb’ I will get telemetry from real Service Bus queue.
Posting of data is shown at the next picture:
Universal Windows App on Pi2
Following code snippet shows the code which set GPIO ports to toggle LEDs on every message sent to Service Bus.
| private async void doIt(object sender, RoutedEventArgs e)
{
m_Voltage = !m_Voltage;
if (m_Voltage)
{
m_Pin27.Write(GpioPinValue.High);
m_Pin22.Write(GpioPinValue.High);
await Dispatcher.RunAsync
(Windows.UI.Core.CoreDispatcherPriority.Normal, () =>
{
m_GpioStatus.Text = "ON";
});
await sendTelemetryData();
}
else
{
m_Pin27.Write(GpioPinValue.Low);
m_Pin22.Write(GpioPinValue.Low); |
Following code snippet shows how to send the data:
| private async Task sendTelemetryData()
{
var tokenProvider = new SASTokenProvider("send_receive", "***V+4=");
var sendClient =
new QueueClient("factorybh", "ingress", tokenProvider, "http");
try
{
Message msg = new Message(
new Command
{
DeviceId = "TEMPPi007",
Id = "102928",
DeviceName = "Pi Temperature Sensor",
Temperature = DateTime.Now.Second,
Timestamp = DateTime.Now.ToString(),
})
{
Properties =
{
{ "DeviceName", "Win10 Device" },
{ "Temperature", DateTime.Now.Minute }
}
};
await sendClient.Send(msg);
}
catch (Exception ex)
{ |
And finally this is the board 
which is sending telemetry data (See first picture on left).
Web Device
At the end I also implemented a HTML5/Java Script application which is running in the browser. This application can also post telemetry data to the Service Bus. For this purpose I used daenet’s - Service Bus Java Script SDK.
| var queueClient = new QueueClient(
{
'name': queuePath,
'namespace': "factorybh",
'sasKey': "bADkRExvoXjw7t81Ln/MGRd2u6mHYMVjUV4+zEInz+4=",
'sasKeyName': "send_receive",
'timeOut': 10,
});
$("#btnSend").click(function () {
var msgBody =
{
"Temperature": txtMsg.value,
"id": 1234,
"DeviceId" : "WEBTEMP01",
"DeviceName" : "WEBDEVICE TEMP. SENSOR 001",
"Timestemp" : new Date()
};
var msg = new BrokeredMessage(msgBody, props);
clear();
queueClient.sendMessage(msg, function (messagingResult) {
$("#result").html(messagingResult.result);
}, "application/json");
}); |
Recap
At the end, I started the app on the Pi2 and taken a look in the LogicApp operation log.
Telemetry Data as received from DeviceConnector
And the data as send sequentially to the HTTP endpoint:
Posted
Jun 14 2015, 04:57 PM
by
Damir Dobric