Random wanderings through Microsoft Azure esp. PaaS plumbing, the IoT bits, AI on Micro controllers, AI on Edge Devices, .NET nanoFramework, .NET Core on *nix and ML.NET+ONNX
class Program
{
private static Model.ApplicationSettings _applicationSettings;
private static IMqttClient _client;
private static bool _publisherBusy = false;
static async Task Main()
{
Console.WriteLine($"{DateTime.UtcNow:yy-MM-dd HH:mm:ss} MQTTNet client starting");
try
{
// load the app settings into configuration
var configuration = new ConfigurationBuilder()
.AddJsonFile("appsettings.json", false, true)
.AddUserSecrets<Program>()
.Build();
_applicationSettings = configuration.GetSection("ApplicationSettings").Get<Model.ApplicationSettings>();
var mqttFactory = new MqttFactory();
using (_client = mqttFactory.CreateMqttClient())
{
// Certificate based authentication
List<X509Certificate2> certificates = new List<X509Certificate2>
{
new X509Certificate2(_applicationSettings.ClientCertificateFileName, _applicationSettings.ClientCertificatePassword)
};
var tlsOptions = new MqttClientTlsOptionsBuilder()
.WithClientCertificates(certificates)
.WithSslProtocols(System.Security.Authentication.SslProtocols.Tls12)
.UseTls(true)
.Build();
MqttClientOptions mqttClientOptions = new MqttClientOptionsBuilder()
.WithClientId(_applicationSettings.ClientId)
.WithTcpServer(_applicationSettings.Host, _applicationSettings.Port)
.WithCredentials(_applicationSettings.UserName, _applicationSettings.Password)
.WithCleanStart(_applicationSettings.CleanStart)
.WithTlsOptions(tlsOptions)
.Build();
var connectResult = await _client.ConnectAsync(mqttClientOptions);
if (connectResult.ResultCode != MqttClientConnectResultCode.Success)
{
throw new Exception($"Failed to connect: {connectResult.ReasonString}");
}
_client.ApplicationMessageReceivedAsync += OnApplicationMessageReceivedAsync;
Console.WriteLine($"Subscribed to Topic");
foreach (string topic in _applicationSettings.SubscribeTopics.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries))
{
var subscribeResult = await _client.SubscribeAsync(topic, _applicationSettings.SubscribeQualityOfService);
Console.WriteLine($" {topic} Result:{subscribeResult.Items.First().ResultCode}");
}
}
//...
}
MQTTnet client console application output
The design of the MQTT protocol means that the hivemq-mqtt-client-dotnet and MQTTnet implementations are similar. Having used both I personally prefer the HiveMQ client library.
For one test deployment it took me an hour to generate the Root, Intermediate and a number of Devices certificates which was a waste of time. At this point I decided investigate writing some applications to simplify the process.
static void Main(string[] args)
{
var serviceProvider = new ServiceCollection()
.AddCertificateManager()
.BuildServiceProvider();
// load the app settings into configuration
var configuration = new ConfigurationBuilder()
.AddJsonFile("appsettings.json", false, true)
.AddUserSecrets<Program>()
.Build();
_applicationSettings = configuration.GetSection("ApplicationSettings").Get<Model.ApplicationSettings>();
//------
Console.WriteLine($"validFrom:{validFrom} ValidTo:{validTo}");
var serverRootCertificate = serviceProvider.GetService<CreateCertificatesClientServerAuth>();
var root = serverRootCertificate.NewRootCertificate(
new DistinguishedName {
CommonName = _applicationSettings.CommonName,
Organisation = _applicationSettings.Organisation,
OrganisationUnit = _applicationSettings.OrganisationUnit,
Locality = _applicationSettings.Locality,
StateProvince = _applicationSettings.StateProvince,
Country = _applicationSettings.Country
},
new ValidityPeriod {
ValidFrom = validFrom,
ValidTo = validTo,
},
_applicationSettings.PathLengthConstraint,
_applicationSettings.DnsName);
root.FriendlyName = _applicationSettings.FriendlyName;
Console.Write("PFX Password:");
string password = Console.ReadLine();
if ( String.IsNullOrEmpty(password))
{
Console.WriteLine("PFX Password invalid");
return;
}
var exportCertificate = serviceProvider.GetService<ImportExportCertificate>();
var rootCertificatePfxBytes = exportCertificate.ExportRootPfx(password, root);
File.WriteAllBytes(_applicationSettings.RootCertificateFilePath, rootCertificatePfxBytes);
Console.WriteLine($"Root certificate file:{_applicationSettings.RootCertificateFilePath}");
Console.WriteLine("press enter to exit");
Console.ReadLine();
}
The application’s configuration was split between application settings file(certificate file paths, validity periods, Organisation etc.) or entered at runtime ( certificate filenames, passwords etc.) The first application generates a Root Certificate using the distinguished name information from the application settings, plus file names and passwords entered by the user.
Root Certificate generation application output
The second application generates an Intermediate Certificate using the Root Certificate, the distinguished name information from the application settings, plus file names and passwords entered by the user.
static void Main(string[] args)
{
var serviceProvider = new ServiceCollection()
.AddCertificateManager()
.BuildServiceProvider();
// load the app settings into configuration
var configuration = new ConfigurationBuilder()
.AddJsonFile("appsettings.json", false, true)
.AddUserSecrets<Program>()
.Build();
_applicationSettings = configuration.GetSection("ApplicationSettings").Get<Model.ApplicationSettings>();
//------
Console.WriteLine($"validFrom:{validFrom} be after ValidTo:{validTo}");
Console.WriteLine($"Root Certificate file:{_applicationSettings.RootCertificateFilePath}");
Console.Write("Root Certificate Password:");
string rootPassword = Console.ReadLine();
if (String.IsNullOrEmpty(rootPassword))
{
Console.WriteLine("Fail");
return;
}
var rootCertificate = new X509Certificate2(_applicationSettings.RootCertificateFilePath, rootPassword);
var intermediateCertificateCreate = serviceProvider.GetService<CreateCertificatesClientServerAuth>();
var intermediateCertificate = intermediateCertificateCreate.NewIntermediateChainedCertificate(
new DistinguishedName
{
CommonName = _applicationSettings.CommonName,
Organisation = _applicationSettings.Organisation,
OrganisationUnit = _applicationSettings.OrganisationUnit,
Locality = _applicationSettings.Locality,
StateProvince = _applicationSettings.StateProvince,
Country = _applicationSettings.Country
},
new ValidityPeriod
{
ValidFrom = validFrom,
ValidTo = validTo,
},
_applicationSettings.PathLengthConstraint,
_applicationSettings.DnsName, rootCertificate);
intermediateCertificate.FriendlyName = _applicationSettings.FriendlyName;
Console.Write("Intermediate certificate Password:");
string intermediatePassword = Console.ReadLine();
if (String.IsNullOrEmpty(intermediatePassword))
{
Console.WriteLine("Fail");
return;
}
var importExportCertificate = serviceProvider.GetService<ImportExportCertificate>();
Console.WriteLine($"Intermediate PFX file:{_applicationSettings.IntermediateCertificatePfxFilePath}");
var intermediateCertificatePfxBtyes = importExportCertificate.ExportChainedCertificatePfx(intermediatePassword, intermediateCertificate, rootCertificate);
File.WriteAllBytes(_applicationSettings.IntermediateCertificatePfxFilePath, intermediateCertificatePfxBtyes);
Console.WriteLine($"Intermediate CER file:{_applicationSettings.IntermediateCertificateCerFilePath}");
var intermediateCertificatePemText = importExportCertificate.PemExportPublicKeyCertificate(intermediateCertificate);
File.WriteAllText(_applicationSettings.IntermediateCertificateCerFilePath, intermediateCertificatePemText);
Console.WriteLine("press enter to exit");
Console.ReadLine();
}
Uploading the Intermediate certificate to Azure Event Grid
The third application generates Device Certificates using the Intermediate Certificate, distinguished name information from the application settings, plus device id, file names and passwords entered by the user.
static void Main(string[] args)
{
var serviceProvider = new ServiceCollection()
.AddCertificateManager()
.BuildServiceProvider();
// load the app settings into configuration
var configuration = new ConfigurationBuilder()
.AddJsonFile("appsettings.json", false, true)
.AddUserSecrets<Program>()
.Build();
_applicationSettings = configuration.GetSection("ApplicationSettings").Get<Model.ApplicationSettings>();
//------
Console.WriteLine($"validFrom:{validFrom} ValidTo:{validTo}");
Console.WriteLine($"Intermediate PFX file:{_applicationSettings.IntermediateCertificateFilePath}");
Console.Write("Intermediate PFX Password:");
string intermediatePassword = Console.ReadLine();
if (String.IsNullOrEmpty(intermediatePassword))
{
Console.WriteLine("Intermediate PFX Password invalid");
return;
}
var intermediate = new X509Certificate2(_applicationSettings.IntermediateCertificateFilePath, intermediatePassword);
Console.Write("Device ID:");
string deviceId = Console.ReadLine();
if (String.IsNullOrEmpty(deviceId))
{
Console.WriteLine("Device ID invalid");
return;
}
var createClientServerAuthCerts = serviceProvider.GetService<CreateCertificatesClientServerAuth>();
var device = createClientServerAuthCerts.NewDeviceChainedCertificate(
new DistinguishedName
{
CommonName = deviceId,
Organisation = _applicationSettings.Organisation,
OrganisationUnit = _applicationSettings.OrganisationUnit,
Locality = _applicationSettings.Locality,
StateProvince = _applicationSettings.StateProvince,
Country = _applicationSettings.Country
},
new ValidityPeriod
{
ValidFrom = validFrom,
ValidTo = validTo,
},
deviceId, intermediate);
device.FriendlyName = deviceId;
Console.Write("Device PFX Password:");
string devicePassword = Console.ReadLine();
if (String.IsNullOrEmpty(devicePassword))
{
Console.WriteLine("Fail");
return;
}
var importExportCertificate = serviceProvider.GetService<ImportExportCertificate>();
string devicePfxPath = string.Format(_applicationSettings.DeviceCertificatePfxFilePath, deviceId);
Console.WriteLine($"Device PFX file:{devicePfxPath}");
var deviceCertificatePath = importExportCertificate.ExportChainedCertificatePfx(devicePassword, device, intermediate);
File.WriteAllBytes(devicePfxPath, deviceCertificatePath);
Console.WriteLine("press enter to exit");
Console.ReadLine();
}
Device Certificate generation application output
Uploading the Intermediate certificate to Azure Event Grid
CREATE PROCEDURE [dbo].[ListingsSpatialNearbyNTSLocation]
@Origin AS GEOGRAPHY,
@distance AS INTEGER
AS
BEGIN
DECLARE @Circle AS GEOGRAPHY = @Origin.STBuffer(@distance);
SELECT TOP(50) UID AS ListingUID
,[Name]
,listing_url as ListingUrl
,Listing.Location.STDistance(@Origin) as Distance
,Listing.Location
FROM Listing
WHERE (Listing.Location.STWithin(@Circle) = 1)
ORDER BY Distance
END
NetTopology Suite Microsoft.SqlServer.Types library load exception
CREATE PROCEDURE [dbo].[ListingsSpatialNearbyNTSSerialize]
@Origin AS GEOGRAPHY,
@distance AS INTEGER
AS
BEGIN
DECLARE @Circle AS GEOGRAPHY = @Origin.STBuffer(@distance);
SELECT TOP(50) UID AS ListingUID
,[Name]
,listing_url as ListingUrl
,Listing.Location.STDistance(@Origin) as Distance
,Location.Serialize() as Location
FROM [listing]
WHERE (Location.STWithin(@Circle) = 1)
ORDER BY Distance
END
class PointHandlerSerialise : SqlMapper.TypeHandler<Point>
{
public override Point Parse(object value)
{
var reader = new SqlServerBytesReader { IsGeography = true };
return (Point)reader.Read((byte[])value);
}
public override void SetValue(IDbDataParameter parameter, Point? value)
{
((SqlParameter)parameter).SqlDbType = SqlDbType.Udt; // @Origin parameter?
((SqlParameter)parameter).UdtTypeName = "GEOGRAPHY";
var writer = new SqlServerBytesWriter { IsGeography = true };
parameter.Value = writer.Write(value);
}
}
Once the location column serialisation was working (I could see a valid response in the debugger) the generation of the response was failing with a “System.Text.Json.JsonException: A possible object cycle was detected. This can either be due to a cycle or if the object depth is larger than the maximum allowed depth of 64″.
NetTopology Suite serialisation “possible object cycle detection” exception
After fixing that issue the response generation failed with “System.ArgumentException: .NET number values such as positive and negative infinity cannot be written as valid JSON.”
CREATE PROCEDURE [dbo].[ListingsSpatialNearbyNTSWkb]
@Origin AS GEOGRAPHY,
@distance AS INTEGER
AS
BEGIN
DECLARE @Circle AS GEOGRAPHY = @Origin.STBuffer(@distance);
SELECT TOP(50) UID AS ListingUID
,[Name]
,listing_url as ListingUrl
,Listing.Location.STDistance(@Origin) as Distance
,Location.STAsBinary() as Location
FROM [listing]
WHERE (Location.STWithin(@Circle) = 1)
ORDER BY Distance
END
CREATE PROCEDURE [dbo].[ListingsSpatialNearbyNTSWkt]
@Origin AS GEOGRAPHY,
@distance AS INTEGER
AS
BEGIN
DECLARE @Circle AS GEOGRAPHY = @Origin.STBuffer(@distance);
SELECT TOP(50) UID AS ListingUID
,[Name]
,listing_url as ListingUrl
,Listing.Location.STDistance(@Origin) as Distance
,Location.STAsText() as Location
FROM [listing]
WHERE (Location.STWithin(@Circle) = 1)
ORDER BY Distance
END
class PointHandlerWkt : SqlMapper.TypeHandler<Point>
{
public override Point Parse(object value)
{
WKTReader wktReader = new WKTReader();
return (Point)wktReader.Read(value.ToString());
}
public override void SetValue(IDbDataParameter parameter, Point? value)
{
((SqlParameter)parameter).SqlDbType = SqlDbType.Udt; // @Origin parameter?
((SqlParameter)parameter).UdtTypeName = "GEOGRAPHY";
parameter.Value = new SqlServerBytesWriter() { IsGeography = true }.Write(value);
}
}
Successful Location processing with WKBReader
I have focused on getting the spatial queries to work and will stress/performance test my implementations in a future post. I will also revisit the /Spatial/NearbyGeography method to see if I can get it to work without “Location.Serialize() as Location”.
CREATE TABLE [dbo].[Neighbourhood](
[id] [bigint] IDENTITY(1,1) NOT NULL,
[NeighbourhoodUID] [uniqueidentifier] NOT NULL,
[name] [nvarchar](50) NOT NULL,
[neighbourhood_url] [nvarchar](100) NOT NULL,
[boundary] [geography] NOT NULL,
CONSTRAINT [PK_Neighbourhood] PRIMARY KEY CLUSTERED
(
[id] ASC
)
-- Then create a spatial index on GEOGRAPHY which contains the boundary polygon(s)
CREATE SPATIAL INDEX [ISX_NeighbourhoodBoundary] ON [dbo].[Neighbourhood]
(
[boundary]
)
I added a GEOGRAPHY column to the Listing table, populated it using the Latitude and Longitudes of the Listings then added a spatial index.
-- Use latitude and longitude to populate Location GEOGRAPHY column
UPDATE listing
SET Listing.Location = geography::Point(latitude, longitude, 4326)
-- Then index Location column after changing to NOT NULL
CREATE SPATIAL INDEX [IXS_ListingByLocation] ON [dbo].[listing]
(
[Location]
)
The first spatial search uses the latitude and longitude (most probably extracted from image metadata) to get a Listing’s neighbourhood.
Testing listing in Neighbourhood SQL
It uses the STContains method to find the neighbourhood polygon (if there is one) which the listing location is inside.
const string ListingInNeighbourhoodSQL = @"SELECT neighbourhoodUID, name, neighbourhood_url as neighbourhoodUrl FROM Neighbourhood WHERE Neighbourhood.Boundary.STContains(geography::Point(@Latitude, @Longitude, 4326)) = 1";
...
app.MapGet("/Spatial/Neighbourhood", async (double latitude, double longitude, [FromServices] IDapperContext dapperContext) =>
{
Model.NeighbourhoodSearchDto neighbourhood;
using (var connection = dapperContext.ConnectionCreate())
{
neighbourhood = await connection.QuerySingleOrDefaultWithRetryAsync<Model.NeighbourhoodSearchDto>(ListingInNeighbourhoodSQL, new { latitude, longitude });
}
if (neighbourhood is null)
{
return Results.Problem($"Neighbourhood for Latitude:{latitude} Longitude:{longitude} not found", statusCode: StatusCodes.Status404NotFound);
}
return Results.Ok(neighbourhood);
})
.Produces<IList<Model.NeighbourhoodSearchDto>>(StatusCodes.Status200OK)
.Produces(StatusCodes.Status404NotFound )
.WithOpenApi();
Testing listing Neighbourhood lookup with Swagger user interface
In a future blog post I will compare the performance of STContains vs. STWithin with a load testing application.
Testing listings near a location SQL
The second search simulates a customer looking for listing(s) within a specified distance of a point of interest.
const string ListingsNearbySQL = @"DECLARE @Origin AS GEOGRAPHY = geography::Point(@Latitude, @Longitude, 4326);
DECLARE @Circle AS GEOGRAPHY = @Origin.STBuffer(@distance);
--DECLARE @Circle AS GEOGRAPHY = @Origin.BufferWithTolerance(@distance, 0.09,true);
SELECT uid as ListingUID, Name, listing_url as ListingUrl,
@Origin.STDistance(Listing.Location) as Distance
FROM [listing]
WHERE Listing.Location.STWithin(@Circle) = 1 ORDER BY Distance";
...
app.MapGet("/Spatial/NearbyText", async (double latitude, double longitude, double distance, [FromServices] IDapperContext dapperContext) =>
{
using (var connection = dapperContext.ConnectionCreate())
{
return await connection.QueryWithRetryAsync<Model.ListingNearbyListDto>(ListingsNearbySQL, new { latitude, longitude, distance });
}
})
.Produces<IList<Model.ListingNearbyListDto>>(StatusCodes.Status200OK)
.WithOpenApi();
The STBuffer command returns a geography object that represents represent a circle centered on @Location with a radius of @distance.
Testing listings near a location with Swagger user interface
The third and final search simulates a customer looking for listing(s) within a specified distance of a point of interest with the latitude and longitude of the listing included in the results.
Testing listings near a location SQL with latitude & Longitude
const string ListingsNearbyLatitudeLongitudeSQL = @"DECLARE @Location AS GEOGRAPHY = geography::Point(@Latitude, @longitude,4326)
DECLARE @Circle AS GEOGRAPHY = @Location.STBuffer(@distance);
SELECT UID as ListingUID
,[Name]
,listing_url as ListingUrl
,Listing.Location.STDistance(@Location) as Distance
,latitude
,longitude
FROM [listing]
WHERE Listing.Location.STWithin(@Circle) = 1
ORDER BY Distance";
app.MapGet("/Spatial/NearbyLatitudeLongitude", async (double latitude, double longitude, double distance, [FromServices] IDapperContext dapperContext) =>
{
using (var connection = dapperContext.ConnectionCreate())
{
return await connection.QueryWithRetryAsync<Model.ListingNearbyListLatitudeLongitudeDto>(ListingsNearbyLatitudeLongitudeSQL, new { latitude, longitude, distance });
}
})
.Produces<IList<Model.ListingNearbyListLatitudeLongitudeDto>>(StatusCodes.Status200OK)
.WithOpenApi();
Testing listings near a location with latitude & Longitude with Swagger user interface
The next couple of posts will use the third-party libraries Geo and NetTopolgySuite
using System.Text.Json;
using System.Data;
using System.Data.SqlClient;
using Microsoft.SqlServer.Types;
using Dapper;
string jsonString = File.ReadAllText("your geoJSON file");
using (IDbConnection connection = new SqlConnection("This is not the connection string you are looking for"))
{
connection.Open();
var neighbourHoods = JsonSerializer.Deserialize<GeoJSON.Text.Feature.FeatureCollection>(jsonString)!;
Console.WriteLine($"Features:{neighbourHoods.Features.Count}");
foreach (var feature in neighbourHoods.Features)
{
string neighbourhood = feature.Properties["neighbourhood"].ToString();
Console.WriteLine($"Neightbourhood:{neighbourhood}");
var geometery = (GeoJSON.Text.Geometry.MultiPolygon)feature.Geometry;
var s = new SqlGeographyBuilder();
s.SetSrid(4326);
s.BeginGeography(OpenGisGeographyType.MultiPolygon);
s.BeginGeography(OpenGisGeographyType.Polygon); // A
Console.WriteLine($"Polygon cordinates:{geometery.Coordinates.Count}");
foreach (var coordinates in geometery.Coordinates)
{
//s.BeginGeography(OpenGisGeographyType.Polygon); // B
Console.WriteLine($"Linestring cordinates:{coordinates.Coordinates.Count}");
foreach (var c in coordinates.Coordinates)
{
Console.WriteLine($"Point cordinates:{c.Coordinates.Count}");
s.BeginFigure(c.Coordinates[0].Latitude, c.Coordinates[0].Longitude, null, null);
for (int i = 1; i < c.Coordinates.Count; i++)
{
s.AddLine(c.Coordinates[i].Latitude, c.Coordinates[i].Longitude);
Console.Write('.');
}
Console.WriteLine();
s.EndFigure();
}
//s.EndGeography(); //B
}
s.EndGeography(); //A
s.EndGeography(); // OpenGisGeographyType.MultiPolygon
connection.Execute("INSERT INTO Neighbourhood (Name, Boundary) VALUES( @Neighbourhood, geography::STMPolyFromText(@boundary, 4326))", new { neighbourhood, boundary = s.ConstructedGeography.ToString()});
Console.WriteLine();
}
}
Console.WriteLine("loaded press <enter> to exit");
Console.ReadLine();
The neighbourhood feature loader utility is pretty “nasty” and was built for my specific scenario
CREATE TABLE [dbo].[Neighbourhood](
[NeighbourhoodUID] [uniqueidentifier] NOT NULL,
[Name] [nvarchar](50) NOT NULL,
[Boundary] [geography] NOT NULL,
CONSTRAINT [PK_Neighbourhood] PRIMARY KEY CLUSTERED
(
[NeighbourhoodUID] ASC
)
I used Dapper to insert rows into the Neighbourhood table
DECLARE @PointGeography AS GEOGRAPHY
SET @PointGeography = geography::Point(51.512837,-0.2894983, 4326) --Flat
--SET @PointGeography = geography::Point(51.5053469,-0.0262693,4326) -- Canary Wharf
--SET @PointGeography = geography::Point(51.476853,0.0,4326) -- Greenwich
--SET @PointGeography = geography::Point(51.501476,-0.140634, 4326) -- Buckingham palace
--SET @PointGeography = geography::Point(51.533611, -0.318889, 4326) -- Hoover factory
--SET @PointGeography = geography::Point(51.5045, -0.0865, 4326) -- Shard
--SET @PointGeography = geography::Point(51.5145683288574,-0.0194199997931719, 4326) -- D0FD60C0-CC45-4517-91F6-00161E710F28 Tower Hamlets
--SET @PointGeography = geography::Point(51.5553092956543,0.00039999998989515, 4326) -- 80264AED-BC74-4150-B393-02D42711E2E6 Waltham Forest
--SET @PointGeography = geography::Point(51.4925193786621,-0.192310005426407, 4326) -- D36E4D1C-4A35-4B6E-B9A7-01E8D732FD3B Kensington and Chelsea
SET @PointGeography = geography::Point(51.5185317993164,-0.199739992618561, 4326) -- 2CAE3CAE-5E43-4F20-9550-01B86D7EF6FF Westminster
SELECT Name, @PointGeography.STWithin(Neighbourhood.Boundary) as 'STWithin', Neighbourhood.Boundary.STContains(@PointGeography) as 'STContains'
FROM Neighbourhood
ORDER BY Name
To test the neighbourhood geography I built a test harness with some “known” locations
SQL Server Management Studio Query “inverted” results
The query results were the inverse of what I was expecting.
SQL Server Management Studio Query “reorientated” results
The Method Callback Delegate has different parameters, so I had to update the downlink formatter interface and update all of the sample downlink payload formatters.
public interface IFormatterDownlink
{
public byte[] Evaluate(string terminalId, string methodName, JObject? payloadJson, byte[] payloadBytes);
}
How direct methods will be processed is configured in the application settings. For each direct method name the downlink payload formatter to be invoked and an optional Javascript Object Notation(JSON) payload can be configured.
If there is no configuration for the direct method name, the payload formatter specified in Myriota device “DownlinkDefault” Attribute is used, and if that is not configured the default formatter in the payloadFormatters section of the application settings is used.
namespace devMobile.IoT.MyriotaAzureIoTConnector.Connector
{
internal class IoTHubDownlink(ILogger<IoTHubDownlink> _logger, IOptions<Models.AzureIoT> azureIoTSettings, IPayloadFormatterCache _payloadFormatterCache, IMyriotaModuleAPI _myriotaModuleAPI) : IIoTHubDownlink
{
private readonly Models.AzureIoT _azureIoTSettings = azureIoTSettings.Value;
public async Task<MethodResponse> IotHubMethodHandler(MethodRequest methodRequest, object userContext)
{
// DIY request identifier so processing progress can be tracked in Application Insights
string requestId = Guid.NewGuid().ToString();
Models.DeviceConnectionContext context = (Models.DeviceConnectionContext)userContext;
try
{
_logger.LogInformation("Downlink- TerminalId:{TerminalId} RequestId:{requestId} Name:{Name}", context.TerminalId, requestId, methodRequest.Name);
// Lookup payload formatter name, none specified use context one which is from device attributes or the default in configuration
string payloadFormatterName;
if (_azureIoTSettings.IoTHub.Methods.TryGetValue(methodRequest.Name, out Models.AzureIoTHubMethod? method) && !string.IsNullOrEmpty(method.Formatter))
{
payloadFormatterName = method.Formatter;
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TermimalId} RequestID:{requestId} Method formatter:{payloadFormatterName} ", context.TerminalId, requestId, payloadFormatterName);
}
else
{
payloadFormatterName = context.PayloadFormatterDownlink;
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TermimalId} RequestID:{requestId} Context formatter:{payloadFormatterName} ", context.TerminalId, requestId, payloadFormatterName);
}
// Display methodRequest.Data as Hex
if (methodRequest.Data is not null)
{
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} Data:{Data}", context.TerminalId, requestId, BitConverter.ToString(methodRequest.Data));
}
else
{
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} Data:null", context.TerminalId, requestId);
}
JObject? requestJson = null;
if ((method is not null) && !string.IsNullOrWhiteSpace(method.Payload))
{
// There is a matching method with a possible JSON payload
string payload = method.Payload.Trim();
if ((payload.StartsWith('{') && payload.EndsWith('}')) || (payload.StartsWith('[') && payload.EndsWith(']')))
{
// The payload is could be JSON
try
{
requestJson = JObject.Parse(payload);
}
catch (JsonReaderException jex)
{
_logger.LogWarning(jex, "Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} Method Payload is not valid JSON", context.TerminalId, requestId);
return new MethodResponse(Encoding.ASCII.GetBytes($"{{\"message\":\"RequestID:{requestId} Method payload is not valid JSON.\"}}"), (int)HttpStatusCode.UnprocessableEntity);
}
}
else
{
// The payload couldn't be JSON
_logger.LogWarning("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} Method Payload is definitely not valid JSON", context.TerminalId, requestId);
return new MethodResponse(Encoding.ASCII.GetBytes($"{{\"message\":\"RequestID:{requestId} Method payload is definitely not valid JSON.\"}}"), (int)HttpStatusCode.UnprocessableEntity);
}
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} Method Payload:{requestJson}", context.TerminalId, requestId, JsonConvert.SerializeObject(requestJson, Formatting.Indented));
}
else
{
// If there was not matching method or the payload was "empty" see if the method request payload is valid
if (!string.IsNullOrWhiteSpace(methodRequest.DataAsJson))
{
string payload = methodRequest.DataAsJson.Trim();
if ((payload.StartsWith('{') && payload.EndsWith('}')) || (payload.StartsWith('[') && payload.EndsWith(']')))
{
// The payload is could be JSON
try
{
requestJson = JObject.Parse(payload);
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} DataAsJson:{requestJson}", context.TerminalId, requestId, JsonConvert.SerializeObject(requestJson, Formatting.Indented));
}
catch (JsonReaderException jex)
{
_logger.LogInformation(jex, "Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} DataAsJson is not valid JSON", context.TerminalId, requestId);
}
}
}
}
// This "shouldn't" fail, but it could for invalid path to blob, timeout retrieving blob, payload formatter syntax error etc.
IFormatterDownlink payloadFormatter = await _payloadFormatterCache.DownlinkGetAsync(payloadFormatterName);
if ( requestJson is null )
{
requestJson = new JObject();
}
// This also "shouldn't" fail, but the payload formatters can throw runtime exceptions like null reference, divide by zero, index out of range etc.
byte[] payloadBytes = payloadFormatter.Evaluate(context.TerminalId, methodRequest.Name, requestJson, methodRequest.Data);
// Validate payload before calling Myriota control message send API method
if (payloadBytes is null)
{
_logger.LogWarning("Downlink- IoT Hub TerminalID:{TerminalId} Request:{requestId} Evaluate returned null", context.TerminalId, requestId);
return new MethodResponse(Encoding.ASCII.GetBytes($"{{\"message\":\"RequestID:{requestId} payload evaluate returned null.\"}}"), (int)HttpStatusCode.UnprocessableEntity);
}
if ((payloadBytes.Length < Constants.DownlinkPayloadMinimumLength) || (payloadBytes.Length > Constants.DownlinkPayloadMaximumLength))
{
_logger.LogWarning("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} PayloadBytes:{payloadBytes} length:{Length} invalid, must be {DownlinkPayloadMinimumLength} to {DownlinkPayloadMaximumLength} bytes", context.TerminalId, requestId, BitConverter.ToString(payloadBytes), payloadBytes.Length, Constants.DownlinkPayloadMinimumLength, Constants.DownlinkPayloadMaximumLength); ;
return new MethodResponse(Encoding.ASCII.GetBytes($"{{\"message\":\"RequestID:{requestId} payload evaluation length invalid.\"}}"), (int)HttpStatusCode.UnprocessableEntity);
}
// Finally send Control Message to device using the Myriota API
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestID} PayloadBytes:{payloadBytes} Length:{Length} sending", context.TerminalId, requestId, BitConverter.ToString(payloadBytes), payloadBytes.Length);
string messageId = await _myriotaModuleAPI.SendAsync(context.TerminalId, payloadBytes);
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} Myriota MessageID:{messageId} sent", context.TerminalId, requestId, messageId);
}
catch (Exception ex)
{
_logger.LogError(ex, "Downlink- IoT Hub TerminalID:{TerminalId} RequestID:{requestId} IotHubMethodHandler processing failed", context.TerminalId, requestId);
return new MethodResponse(Encoding.ASCII.GetBytes($"{{\"message\":\"TerminalID:{context.TerminalId} RequestID:{requestId} method handler failed.\"}}"), (int)HttpStatusCode.InternalServerError);
}
return new MethodResponse(Encoding.ASCII.GetBytes($"{{\"message\":\"TerminalID:{context.TerminalId} RequestID:{requestId} Message sent successfully.\"}}"), (int)HttpStatusCode.OK);
}
}
}
public class FormatterDownlink : PayloadFormatter.IFormatterDownlink
{
public byte[] Evaluate(string terminalId, string methodName, JObject payloadJson, byte[] payloadBytes)
{
byte? status = payloadJson.GetValue("FanSpeed", StringComparison.OrdinalIgnoreCase)?.Value<byte>();
if (!status.HasValue)
{
return new byte[] { };
}
return new byte[] { 1, status.Value };
}
}
The FanSpeed.cs payload formatter extracts the FanSpeed value from the JSON payload and returns a two byte array containing the message type and speed of the fan.
Azure Function application displaying Diagnostic information for control message
Each logging message starts with the TerminalID (to simplify searching for all the direct methods invoked on a device) and the requestId a Globally Unique Identifier (GUID) to simplify searching for all the “steps” associated with sending a message) with the rest of the logging message containing “step” specific diagnostic information.
Azure Application Insights displaying information diagnostic information
Myriota Device manager control message history displaying pending control message
The Azure IoT Explorer payload for an empty message contained two ” characters which is a bit odd. I will have to build a test application which uses the Azure IoT Hub C2D direct method API to see if this is a “feature”.
Each logging message starts with the TerminalID (to simplify searching for all the messages sent to a device) and the message LockToken (to simplify searching for all the “steps” associated with sending a message) with the rest of the logging message containing “step” specific diagnostic information.
Successful Azure IoT Explorer C2D JSON Message
If there is no PayloadFormatter attribute the default in the PayloadFormatters section of the function configuration is used.
using System;
using System.Collections.Generic;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
public class FormatterDownlink : PayloadFormatter.IFormatterDownlink
{
public byte[] Evaluate(IDictionary<string, string> properties, string terminalId, JObject payloadJson, byte[] payloadBytes)
{
byte? status = payloadJson.Value<byte?>("FanSpeed");
if (!status.HasValue)
{
return new byte[] { };
}
return new byte[] { 1, status.Value };
}
}
The FanSpeed.cs payload formatter extracts the FanSpeed value from the JSON payload and returns a two byte array containing the message type and speed of the fan.
Azure IoT Function running waiting for a C2D message
After re-reading the SetMethodHandlerAync documentation I refactored the code (back to the approach used a couple of branches ago) with the “using” wrapping the try/catch.
public async Task AzureIoTHubMessageHandler(Message message, object userContext)
{
Models.DeviceConnectionContext context = (Models.DeviceConnectionContext)userContext;
_logger.LogInformation("Downlink- IoT Hub TerminalId:{TermimalId} LockToken:{LockToken}", context.TerminalId, message.LockToken);
using (message) // https://learn.microsoft.com/en-us/dotnet/api/microsoft.azure.devices.client.deviceclient.setreceivemessagehandlerasync?view=azure-dotnet
{
try
{
// Replace default formatter with message specific formatter if configured.
if (!message.Properties.TryGetValue(Constants.IoTHubDownlinkPayloadFormatterProperty, out string? payloadFormatterName) || string.IsNullOrEmpty(payloadFormatterName))
{
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TermimalId} LockToken:{LockToken} Context formatter:{payloadFormatterName} ", context.TerminalId, message.LockToken, payloadFormatterName);
payloadFormatterName = context.PayloadFormatterDownlink;
}
else
{
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TermimalId} LockToken:{LockToken} Property formatter:{payloadFormatterName} ", context.TerminalId, message.LockToken, payloadFormatterName);
}
// If GetBytes fails payload really badly broken
byte[] messageBytes = message.GetBytes();
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} Message bytes:{messageBytes}", context.TerminalId, message.LockToken, BitConverter.ToString(messageBytes));
// Try converting the bytes to text then to JSON
JObject? messageJson = null;
try
{
// These will fail for some messages, payload formatter gets bytes only
string messageText = Encoding.UTF8.GetString(messageBytes);
try
{
messageJson = JObject.Parse(messageText);
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} JSON:{messageJson}", context.TerminalId, message.LockToken, JsonConvert.SerializeObject(messageJson, Formatting.Indented));
}
catch (JsonReaderException jex)
{
_logger.LogInformation(jex, "Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} not valid JSON", context.TerminalId, message.LockToken);
}
}
catch (ArgumentException aex)
{
_logger.LogInformation(aex, "Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} message bytes not valid text", context.TerminalId, message.LockToken);
}
// This shouldn't fail, but it could for invalid path to blob, timeout retrieving blob, payload formatter syntax error etc.
IFormatterDownlink payloadFormatter = await _payloadFormatterCache.DownlinkGetAsync(payloadFormatterName);
// This will fail if payload formatter throws runtime exceptions like null reference, divide by zero, index out of range etc.
byte[] payloadBytes = payloadFormatter.Evaluate(message.Properties, context.TerminalId, messageJson, messageBytes);
// Validate payload before calling Myriota control message send API method
if (payloadBytes is null)
{
_logger.LogWarning("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} payload formatter:{payloadFormatter} Evaluate returned null", context.TerminalId, message.LockToken, payloadFormatterName);
await context.DeviceClient.RejectAsync(message);
return;
}
if ((payloadBytes.Length < Constants.DownlinkPayloadMinimumLength) || (payloadBytes.Length > Constants.DownlinkPayloadMaximumLength))
{
_logger.LogWarning("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} PayloadBytes:{payloadBytes} length:{Length} invalid must be {DownlinkPayloadMinimumLength} to {DownlinkPayloadMaximumLength} bytes", context.TerminalId, message.LockToken, Convert.ToHexString(payloadBytes), payloadBytes.Length, Constants.DownlinkPayloadMinimumLength, Constants.DownlinkPayloadMaximumLength);
await context.DeviceClient.RejectAsync(message);
return;
}
// Finally send Control Message to device using the Myriota API
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} PayloadBytes:{payloadBytes} Length:{Length} sending", context.TerminalId, message.LockToken, BitConverter.ToString(payloadBytes), payloadBytes.Length);
string messageId = await _myriotaModuleAPI.SendAsync(context.TerminalId, payloadBytes);
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} MessageID:{messageId} sent", context.TerminalId, message.LockToken, messageId);
await context.DeviceClient.CompleteAsync(message);
}
catch (Exception ex)
{
_logger.LogError(ex, "Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} MessageHandler processing failed", context.TerminalId, message.LockToken);
await context.DeviceClient.RejectAsync(message);
}
}
}
...
// Finally send Control Message to device using the Myriota API
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} PayloadBytes:{payloadBytes} Length:{Length} sending", context.TerminalId, message.LockToken, BitConverter.ToString(payloadBytes), payloadBytes.Length);
string messageId = await _myriotaModuleAPI.SendAsync(context.TerminalId, payloadBytes);
...
Azure IoT Function successfully sending downlink message.
The Encoding.UTF8.GetString and JObject.Parse are processed in a single Try with a specialised catch for when the payload cannot be converted to text. If the payload cannot be converted to JSON only the payloadBytes parameter of payload formatter is populated.
The Azure IoT Hub downlink message handler was a partial class and part of implementation of the IDeviceConnectionCache which was a hangover from one of the initial versions.
The myriotaAzure IoT Hub Cloud Identity Translation Gateway uplink message handler Azure Storage Queue Trigger Function wasn’t processing “transient” vs. “permanent” failures well. Sometimes a “permanent” failure message would be retried multiple times by the function runtime before getting moved to the poison queue.
After some experimentation using an Azure Storage Queue Function Output binding to move messages to the poison queue looked like a reasonable approach. (Though, returning null to indicate the message should be removed from the queue was not obvious from the documentation)
[Function("UplinkMessageProcessor")]
[QueueOutput(queueName: "uplink-poison", Connection = "UplinkQueueStorage")]
public async Task<Models.UplinkPayloadQueueDto> UplinkMessageProcessor([QueueTrigger(queueName: "uplink", Connection = "UplinkQueueStorage")] Models.UplinkPayloadQueueDto payload, CancellationToken cancellationToken)
{
...
// Process each packet in the payload. Myriota docs say only one packet per payload but just incase...
foreach (Models.QueuePacket packet in payload.Data.Packets)
{
// Lookup the device client in the cache or create a new one
Models.DeviceConnectionContext context;
try
{
context = await _deviceConnectionCache.GetOrAddAsync(packet.TerminalId, cancellationToken);
}
catch (DeviceNotFoundException dnfex)
{
_logger.LogError(dnfex, "Uplink- PayloadId:{0} TerminalId:{1} terminal not found", payload.Id, packet.TerminalId);
return payload;
}
catch (Exception ex) // Maybe just send to poison queue or figure if transient error?
{
_logger.LogError(ex, "Uplink- PayloadId:{0} TerminalId:{1} ", payload.Id, packet.TerminalId);
throw;
}
...
// Proccessing successful, message can be deleted by QueueTrigger plumbing
return null;
}
After building and testing an Azure Storage Queue Function Output binding implementation I’m not certain that it is a good approach. The code is a bit “chunky” and I have had to implement more of the retry process logic.
The myriotaAzure IoT Hub Cloud Identity Translation Gateway downlink message handler was getting a bit “chunky”. So, I started by stripping the code back to the absolute bare minimum that would “work”.
Then the code was then extended so it worked for “sunny day” scenarios. The payload formatter was successfully retrieved from the configured Azure Storage Blob, CS-Script successfully compiled the payload formatter, the message payload was valid text, the message text was valid Javascript Object Notation(JSON), the JSON was successfully processed by the compiled payload formatter, and finally the payload was accepted by the Myriota Cloud API.
Then finally the code was modified to gracefully handle broken payloads returned by the payload formatter evaluation, some comments were added, and the non-managed resources of the DeviceClient.Message disposed.
public async Task AzureIoTHubMessageHandler(Message message, object userContext)
{
Models.DeviceConnectionContext context = (Models.DeviceConnectionContext)userContext;
_logger.LogInformation("Downlink- IoT Hub TerminalId:{termimalId} LockToken:{LockToken}", context.TerminalId, message.LockToken);
// Use default formatter and replace with message specific formatter if configured.
string payloadFormatter;
if (!message.Properties.TryGetValue(Constants.IoTHubDownlinkPayloadFormatterProperty, out payloadFormatter) || string.IsNullOrEmpty(payloadFormatter))
{
payloadFormatter = context.PayloadFormatterDownlink;
}
_logger.LogInformation("Downlink- IoT Hub TerminalID:{termimalId} LockToken:{LockToken} Payload formatter:{payloadFormatter} ", context.TerminalId, message.LockToken, payloadFormatter);
try
{
// If this fails payload broken
byte[] messageBytes = message.GetBytes();
// This will fail for some messages, payload formatter gets bytes only
string messageText = string.Empty;
try
{
messageText = Encoding.UTF8.GetString(messageBytes);
}
catch (ArgumentException aex)
{
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} messageBytes:{2} not valid Text", context.TerminalId, message.LockToken, BitConverter.ToString(messageBytes));
}
// This will fail for some messages, payload formatter gets bytes only
JObject? messageJson = null;
try
{
messageJson = JObject.Parse(messageText);
}
catch ( JsonReaderException jex)
{
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} messageText:{2} not valid json", context.TerminalId, message.LockToken, BitConverter.ToString(messageBytes));
}
// This shouldn't fail, but it could for lots of diffent reasons, invalid path to blob, syntax error, interface broken etc.
IFormatterDownlink payloadFormatterDownlink = await _payloadFormatterCache.DownlinkGetAsync(payloadFormatter);
// This shouldn't fail, but it could for lots of different reasons, null references, divide by zero, out of range etc.
byte[] payloadBytes = payloadFormatterDownlink.Evaluate(message.Properties, context.TerminalId, messageJson, messageBytes);
// Validate payload before calling Myriota control message send API method
if (payloadBytes is null)
{
_logger.LogWarning("Downlink- IoT Hub TerminalID:{terminalId} LockToken:{LockToken} payload formatter:{payloadFormatter} Evaluate returned null", context.TerminalId, message.LockToken, payloadFormatter);
await context.DeviceClient.RejectAsync(message);
return;
}
if ((payloadBytes.Length < Constants.DownlinkPayloadMinimumLength) || (payloadBytes.Length > Constants.DownlinkPayloadMaximumLength))
{
_logger.LogWarning("Downlink- IoT Hub TerminalID:{terminalId} LockToken:{LockToken} payloadData length:{Length} invalid must be {DownlinkPayloadMinimumLength} to {DownlinkPayloadMaximumLength} bytes", context.TerminalId, message.LockToken, payloadBytes.Length, Constants.DownlinkPayloadMinimumLength, Constants.DownlinkPayloadMaximumLength);
await context.DeviceClient.RejectAsync(message);
return;
}
// This shouldn't fail, but it could few reasons mainly connectivity & message queuing etc.
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} PayloadData:{payloadData} Length:{Length} sending", context.TerminalId, message.LockToken, Convert.ToHexString(payloadBytes), payloadBytes.Length);
// Finally send the message using Myriota API
string messageId = await _myriotaModuleAPI.SendAsync(context.TerminalId, payloadBytes);
_logger.LogInformation("Downlink- IoT Hub TerminalID:{TerminalId} LockToken:{LockToken} MessageID:{messageId} sent", context.TerminalId, message.LockToken, messageId);
await context.DeviceClient.CompleteAsync(message);
_logger.LogInformation("Downlink- IoT Hub TerminalID:{terminalId} LockToken:{LockToken} MessageID:{messageId} sent", context.TerminalId, message.LockToken, messageId);
}
catch (Exception ex)
{
await context.DeviceClient.RejectAsync(message);
_logger.LogError(ex, "Downlink- IoT Hub TerminalID:{terminalId} LockToken:{LockToken} failed", context.TerminalId, message.LockToken);
}
finally
{
// Mop up the non managed resources of message
message.Dispose();
}
}
As the code was being extended, I tested different failures to make sure the Application Insights logging messages were useful. The first failure mode tested was the Azure Storage Blob, path was broken or the blob was missing.
Visual Studio 2022 Debugger blob not found exception message
Application Insights blob not found exception logging
Then a series of “broken” payload formatters were created to test CS-Script compile time failures.
// Broken interface implementation
using System;
using System.Collections.Generic;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
public class FormatterDownlink : PayloadFormatter.IFormatterDownlink
{
public byte[] Evaluate(IDictionary<string, string> properties, string terminalId, byte[] payloadBytes)
{
return payloadBytes;
}
}
Visual Studio 2022 Debugger interface implementation broken exception message
// Broken syntax
using System;
using System.Collections.Generic;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
public class FormatterDownlink : PayloadFormatter.IFormatterDownlink
{
public byte[] Evaluate(IDictionary<string, string> properties, string terminalId, JObject payloadJson, byte[] payloadBytes)
{
return payloadBytes
}
}
Visual Studio 2022 Debugger syntax error exception message
The final test was sending a downlink message which was valid JSON, contained the correct information for the specified payload formatter and was successfully processed by the Myriota Cloud API.
Azure IoT Explorer with valid JSON payload and payload formatter name
Azure function output of successful downlink message
devMobile's blog 