After getting MQ Telemetry Transport (MQTT) Device to Cloud (D2C) messaging working for AdaFruit.IO I have also got Cloud to Device (C2D) messaging working as well.
The MQTT broker, username, API key, client ID, optional group name (to keep MQTT aligned with REST API terminology), command topic and feed name are command line options.
The Adafruit IO MQTT documentation suggests an approach for naming topics which allows a bit more structure for feed (D2C and C2D) names than the REST API (which only does D2C).
class Program
{
private static IMqttClient mqttClient = null;
private static IMqttClientOptions mqttOptions = null;
private static string server;
private static string username;
private static string password;
private static string clientId;
private static string commandTopic;
private static string groupname;
private static string feedname;
static void Main(string[] args)
{
MqttFactory factory = new MqttFactory();
mqttClient = factory.CreateMqttClient();
if ((args.Length != 6) && (args.Length != 7))
{
Console.WriteLine("[MQTT Server] [UserName] [Password] [ClientID] [CommandTopic] [GroupName] [FeedName]");
Console.WriteLine("[MQTT Server] [UserName] [Password] [ClientID] [CommandTopic] [FeedName]");
Console.WriteLine("Press <enter> to exit");
Console.ReadLine();
return;
}
server = args[0];
username = args[1];
password = args[2];
clientId = args[3];
commandTopic = args[4];
if (args.Length == 6)
{
feedname = args[5].ToLower();
Console.WriteLine($"MQTT Server:{server} Username:{username} ClientID:{clientId} CommandTopic:{commandTopic} Feedname:{feedname}");
}
if (args.Length == 7)
{
groupname = args[5].ToLower();
feedname = args[6].ToLower();
Console.WriteLine($"MQTT Server:{server} Username:{username} ClientID:{clientId} CommandTopic:{commandTopic} Groupname:{groupname} Feedname:{feedname}");
}
mqttOptions = new MqttClientOptionsBuilder()
.WithTcpServer(server)
.WithCredentials(username, password)
.WithClientId(clientId)
.WithTls()
.Build();
mqttClient.Disconnected += MqttClient_Disconnected;
mqttClient.ConnectAsync(mqttOptions).Wait();
mqttClient.ApplicationMessageReceived += MqttClient_ApplicationMessageReceived;
// Adafruit.IO format for topics which are called feeds
string topic = string.Empty;
if (args.Length == 6)
{
topic = $"{username}/feeds/{feedname}";
}
if (args.Length == 7)
{
topic = $"{username}/feeds/{groupname}.{feedname}";
}
mqttClient.SubscribeAsync(commandTopic, MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce).GetAwaiter().GetResult();
while (true)
{
string value = "22." + DateTime.UtcNow.Millisecond.ToString();
Console.WriteLine($"Topic:{topic} Value:{value}");
var message = new MqttApplicationMessageBuilder()
.WithTopic(topic)
.WithPayload(value)
.WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce)
.WithRetainFlag()
.Build();
Console.WriteLine("PublishAsync start");
mqttClient.PublishAsync(message).Wait();
Console.WriteLine("PublishAsync finish");
Thread.Sleep(30100);
}
}
private static void MqttClient_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
{
Console.WriteLine($"ClientId:{e.ClientId} Topic:{e.ApplicationMessage.Topic} Payload:{e.ApplicationMessage.ConvertPayloadToString()}");
}
private static async void MqttClient_Disconnected(object sender, MqttClientDisconnectedEventArgs e)
{
Debug.WriteLine("Disconnected");
await Task.Delay(TimeSpan.FromSeconds(5));
try
{
await mqttClient.ConnectAsync(mqttOptions);
}
catch (Exception ex)
{
Debug.WriteLine("Reconnect failed {0}", ex.Message);
}
}
}
I configured a slider on the dashboard for my home called “setpoint” (yet again I was tripped up “automatically” camel casing the name because I’m a C# developer) which my MQTT client subscribed to.
After figuring out the format of the command topic I found that when the slider was moved the MQTT client subscription event fired reliably.
Overall the process went pretty well, though the manual configuration of the subscriptions to AdaFruit.IO feeds could become a bit of a problem at scale.
devMobile's blog 