MQTT client sample

The MQTT client sample uses the MQTTnet nuget package. You'll find further information about this library here.

Prerequisites

In order for the MQTT sample to run it is necessary to start the Mosquitto MQTT broker.

General information

The MQTT driver extension sample publishes and subscribes to topics on the MQTT broker. For the topic, the symbolic address of the variable is used. Because of the nature of MQTT this implies, that all values which are published to the broker are also sent back from the broker to the client.

Initialisation

The configuration of the MQTT client happens in the InitializeAsync method of the driver extension. If a configuration file is provided, the MQTT server (broker) IP address is taken from there. If not, the sample uses the loopback adress 127.0.0.1 and the default unencrypted MQTT port 1883. The clientId is myLocalClientId if not defined otherwise.

After the options are set, the client connects to the broker.

      var configuration = GetConfiguration(configFilePath);
      var serverAddress = configuration["MqttServerAddress"] ?? "127.0.0.1";
      var clientId = configuration["ClientId"] ?? "myLocalClientId";

var options = new MqttClientOptionsBuilder()
    .WithClientId(clientId)
    .WithTcpServer(serverAddress, 1883)
    .WithCleanSession()
    .Build();

await _mqttClient.ConnectAsync(options, CancellationToken.None);

Further in the initialisation two handlers are defined:

• _mqttClient.UseDisconnectedHandler(...)
• _mqttClient.UseApplicationMessageReceivedHandler(...)

As an argument these handlers get a lamdba expression (anonymous function). This function is executed, in case of a disconnect or a value is received for a subscribed topic.

On a disconnect, the driver extension writes all last known values to zenon. Then it waits for 5 seconds and tries to reconnect. If the reconnect is successfull, the subscriptions for all advised variables are renewed.

      _mqttClient.UseDisconnectedHandler(async e =>
      {
        if (!_mqttClient.IsConnected)
        {
          foreach (var subscription in _subscriptions)
          {
            _valueCallback.SetValue(subscription, StatusBits.Invalid);
          }
        }

        logger.Warn("### DISCONNECTED FROM SERVER ###");

        await Task.Delay(TimeSpan.FromSeconds(5));

        try
        {
          await _mqttClient.ReconnectAsync();

          foreach (var symbolicAddress in _subscriptions)
          {
            await _mqttClient.SubscribeAsync(
                      new TopicFilterBuilder().WithTopic(symbolicAddress).Build());
          }
        }
        catch
        {
          logger.Error("### RECONNECTING FAILED ###");
        }
      });

The message received handler is called every time the MQTT client driver extension receives a value from the broker. The lambda expression provided to the handler extracts the value and timestamp from the message payload. The value is immediately updated in zenon by means of the _valueCallback.SetValue(...) method. Please remark, that the internal list holding all the advised variables is not checked.

      _mqttClient.UseApplicationMessageReceivedHandler(args =>
      {
        var payload = Encoding.UTF8.GetString(args.ApplicationMessage.Payload);
        var t = JsonConvert.DeserializeObject<SensorPayload>(payload);
        _valueCallback.SetValue(args.ApplicationMessage.Topic, t.Value, t.LastChangeDateTime);
      });

Subscribing to - Advising - a variable

Whenever a variable is subscribed (also advised) by zenon to the driver extension, the symbolic address is used as topic for the MQTT client to subscribe at the broker.

public async Task<bool> SubscribeAsync(string symbolicAddress)
    {
      _subscriptions.Add(symbolicAddress);
      // Subscribe to a topic
      await _mqttClient.SubscribeAsync(new TopicFilterBuilder()
        .WithTopic(symbolicAddress)
        .Build()
      );

Writing a numeric value

If zenon writes a numeric value to the driver extension, the MQTT client is used to publish the value unter the topic of the symbolic address of the variable.

public async Task<bool> WriteNumericAsync(string symbolicAddress, double value, DateTime dateTime, StatusBits status)
    {
      var sensorPayload = new SensorPayload() { Value = value, LastChangeDateTime = dateTime };
      var payloadString = JsonConvert.SerializeObject(sensorPayload);

      var message = new MqttApplicationMessageBuilder()
        .WithTopic(symbolicAddress)
        .WithPayload(payloadString)
        .WithExactlyOnceQoS()
        .WithRetainFlag()
        .Build();

      try
      {
        await _mqttClient.PublishAsync(message);
        return true;
      }
      catch (Exception e)
      {
        _logger.Error("Error while submitting value: " + e.Message);
      }

      return false;
    }

Other methods

Other methods of the driver extension are quite simple and can be well understood by reading and interpreting the code. Please remark, that not all methods provide a useful example. Like the WriteStringAsync(...) method is more or less undefined and just returns a false.