Inviare dati in tempo reali e storici da un applicativo .NET (server) ad un’applicazione Flutter (client) utilizzando il protocollo MQTT

Pubblicato il di

In questa guida realizzeremo passo passo un’applicazione console .NET (il nostro server) che invierà la temperatura attuale e le ultime rilevazioni effettuate di una serie di sensori ad un’applicazione Flutter Android (il nostro client).

L’applicazione Flutter permetterà di selezionare il sensore da un’elenco e visualizzarne le temperature rilevate. La comunicazione verrà effettuata tramite protocollo MQTT e come ambiente di sviluppo utilizzeremo VSCode.

I valori temperatura saranno numeri interi generati casualmente, non saranno realmente recuperati da sensori fisici. Il concetto sensori temperatura serve solo per rendere l’idea di quale applicazione potrebbe avere un progetto simile.

Il protocollo MQTT

Come già visto in precedenza nel seguente articolo, MQTT è un protocollo di messaggistica publish-subscribe, opera sopra a TCP/IP ed è molto leggero, ideale negli ambiti in cui è richiesto un’utilizzo della banda ridotto ed un basso impatto energetico. Perfetto per un sensore temperatura ad esempio.

La struttura del nostro esempio

Come già anticipato il nostro progetto sarà suddiviso in due parti: l’applicativo console (server) e l’applicazione Flutter (client). Vediamoli più nel dettaglio:

Console Application .NET

L’applicativo console .NET si occuperà di due compiti:

  • Esporre un endpoint API GET localhost:8080/api/sensors che restituirà l’elenco dei sensori temperatura disponibili in formato JSON.
    Oltre ovviamente al nome del sensore, proprietà sensorName, verranno restituiti latestValueTopic e historyTopic rispettivamente ad indicare il topic MQTT per il ricevimento dell’ultimo valore di temperatura e per il ricevimento dello storico valori precedenti.
  • Istanziare un broker MQTT che invii per ogni sensore (ad un’intervallo pre-impostato) un messaggio contenente l’ultima temperatura rilevata (un valore casuale da 0 a 100) ed un messaggio contenente lo storico aggiornato delle rilevazioni precedenti.
    Per questioni di dimensioni lo storico riporterà le ultime 99 rilevazioni per ogni sensore, la centesima sarà l’ultima rilevata.
    I topic saranno così strutturati:
    • sensor1/latest
    • sensor1/history
    • sensor2/latest
    • sensor2/history
    • …

Applicazione Flutter

L’applicazione Flutter una volta avviata mostrerà l’elenco dei sensori disponibili. Selezionandone uno, mostrerà l’ultimo valore di temperatura rilevato con il riferimento data e ora UTC e le precedenti rilevazioni, lo storico appunto.

Il progetto avrà questa struttura cartelle:

- sensors_example
    - broker
        - ...
    - client
        - ...

Iniziamo con la realizzazione.

Il broker

Creiamo una nuova cartella per il progetto con mkdir sensors_example; cd sensors_example.

Lanciamo dotnet new console -o broker per creare il progetto console .NET ed apriamo VSCode sulla cartella sensors_example con code ..

Predisponiamo il task di compilazione e lancio dell’applicativo creando una nuova cartella chiamata .vscode nella root di progetto con all’interno un file tasks.json:

{
    "version": "2.0.0",
    "tasks": [
        {
            "label": "build",
            "command": "dotnet",
            "type": "process",
            "args": [
                "build",
                "${workspaceFolder}/broker/broker.csproj",
                "/property:GenerateFullPaths=true",
                "/consoleloggerparameters:NoSummary"
            ],
            "problemMatcher": "$msCompile"
        }
    ]
}

e successivamente launch.json:

{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "broker",
            "type": "coreclr",
            "request": "launch",
            "preLaunchTask": "build",
            "program": "${workspaceFolder}/broker/bin/Debug/net9.0/broker.dll",
            "args": [],
            "cwd": "${workspaceFolder}/broker/bin/Debug/net9.0",
            "stopAtEntry": false,
            "console": "internalConsole",
            "internalConsoleOptions": "openOnSessionStart",
        }
    ]
}

Tramite F5 lanciamo quindi la nuova configurazione broker e verifichiamo che funzioni correttamente.

All’interno del file di progetto broker.csproj andiamo a sostituire Microsoft.NET.Sdk con Microsoft.NET.Sdk.Web (prima riga del file), ed aggiungiamo tutti i pacchetti necessari:

<ItemGroup>
    <PackageReference Include="Microsoft.AspNetCore.Hosting" Version="2.3.0" />
    <PackageReference Include="Microsoft.AspNetCore.Server.Kestrel" Version="2.3.0" />
    <PackageReference Include="Microsoft.Extensions.Hosting" Version="9.0.5" />
    <PackageReference Include="Microsoft.AspNetCore.Http" Version="2.3.0" />
    <PackageReference Include="MQTTnet" Version="5.0.1.1416" />
    <PackageReference Include="MQTTnet.Server" Version="5.0.1.1416" />
    <PackageReference Include="Serilog.Extensions.Hosting" Version="9.0.0" />
    <PackageReference Include="Serilog.Settings.Configuration" Version="9.0.0" />
    <PackageReference Include="Serilog.Sinks.Console" Version="6.0.0" />
</ItemGroup>

Dato che il nostro applicativo istanzierà, oltre ad un server MQTT, anche un endpoint API tramite Kestrel, bisogna utilizzare l’SDK Web.

Per il broker MQTT utilizziamo la libreria MQTTnet e per il logging la libreria Serilog.

Creiamo ora il file appsettings.json che conterrà la configurazione Serilog e la proprietà SensorHistoryItemSize (proprietà che indicherà il numero massimo di elementi salvati nello storico).

{
    "AppSettings": {
        "SensorHistoryItemSize": 100
    },
    "Serilog": {
        "MinimumLevel": {
            "Default": "Debug",
            "Override": {
                "Microsoft": "Fatal"
            }
        },
        "WriteTo": [
            {
                "Name": "Console"
            }
        ],
        "Enrich": [
            "FromLogContext",
            "WithExceptionDetails"
        ],
        "Properties": {
            "ApplicationName": "broker",
            "Environment": "Int"
        }
    }
}

Così configurato, Serilog scriverà sulla console.

Impostiamo CopyToOutputDirectory per il nuovo file di configurazione nel file di progetto broker.csproj

<ItemGroup>
    <None Update="appsettings.json">
      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
    </None>
</ItemGroup>

Creiamo la classe AppSettings.cs per la proprietà SensorHistoryItemSize

namespace broker
{
    public class AppSettings
    {
        public int SensorHistoryItemSize { get; set; }
    }
}

Aggiungiamo le classi dei modelli Sensor e Payload (Models.cs)

namespace broker
{
    class Sensor
    {
        public required string SensorName { get; set; }
        public required string LatestValueTopic { get; set; }
        public required string HistoryTopic { get; set; }
    }

    class Payload
    {
        public DateTime TimestampUtc { get; set; }
        public required string SensorName { get; set; }
        public int Value { get; set; }
    }
}

e il file Cns.cs per le costanti

namespace broker
{
    internal class Cns
    {
        internal const string AppSettingsFile = "appsettings.json";

        internal readonly static IReadOnlyList<Sensor> Sensors = [
            new Sensor{
                SensorName = "sensor1",
                LatestValueTopic = "sensor1/latest",
                HistoryTopic = "sensor1/history",
            },
            new Sensor{
                SensorName = "sensor2",
                LatestValueTopic = "sensor2/latest",
                HistoryTopic = "sensor2/history",
            },
            new Sensor{
                SensorName = "sensor3",
                LatestValueTopic = "sensor3/latest",
                HistoryTopic = "sensor3/history",
            },
        ];
    }
}

Iniziamo ora ad implementare il WebHost principale dell’applicazione.

Modifichiamo il file Program.cs come segue

using broker;
using Serilog;
using System.Text.Json;

var configuration = new ConfigurationBuilder()
    .SetBasePath(Directory.GetCurrentDirectory())
    .AddJsonFile(Cns.AppSettingsFile)
    .Build();

var hostBuilder = new HostBuilder()
    .ConfigureWebHostDefaults(webBuilder =>
    {
        webBuilder.UseUrls("https://localhost:8080");
        webBuilder.UseKestrel();

        webBuilder.Configure(app =>
        {
            app.UseHttpsRedirection();

            app.Map("/api/sensors", appBuilder =>
            {
                appBuilder.Run(async context =>
                {
                    context.Response.ContentType = "application/json";

                    var options = new JsonSerializerOptions
                    {
                        PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
                        WriteIndented = true
                    };

                    await context.Response.WriteAsync(JsonSerializer.Serialize(Cns.Sensors, options));
                });
            });

            app.Run(async context =>
            {
                context.Response.ContentType = "text/plain";
                await context.Response.WriteAsync("Welcome to MQTT Sensor Example, visit /api/sensors for available sensor list");
            });
        });
    })
    .ConfigureLogging(logging =>
    {
        Log.Logger = new LoggerConfiguration()
                    .ReadFrom.Configuration(configuration)
                    .CreateLogger();
    });

await hostBuilder.RunConsoleAsync();

Vediamo il codice sopra più nel dettaglio:

var configuration = new ConfigurationBuilder()
  .SetBasePath(Directory.GetCurrentDirectory())
  .AddJsonFile(Cns.AppSettingsFile)
  .Build();
  • Carica la configurazione contenuta nel file appsettings.json per iniettarla successivamente nella creazione del logger:
Log.Logger = new LoggerConfiguration()
                    .ReadFrom.Configuration(configuration)
                    .CreateLogger();

webBuilder.UseUrls("https://localhost:8080");
webBuilder.UseKestrel();
...
app.UseHttpsRedirection();
  • Abilita Kestrel in ascolto sulla porta 8080 e re-dirige il traffico HTTP verso HTTPS
app.Run(async context =>
{
  context.Response.ContentType = "text/plain";
  await context.Response.WriteAsync("Welcome to MQTT Sensor Example, visit /api/sensors for available sensor list");
});
  • Permette di restituire il messaggio informativo indicato se non è richiesta l’API impostata /api/sensors seguente
app.Map("/api/sensors", appBuilder =>
{
  appBuilder.Run(async context =>
  {
      context.Response.ContentType = "application/json";

      var options = new JsonSerializerOptions
      {
          PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
          WriteIndented = true
      };

      await context.Response.WriteAsync(JsonSerializer.Serialize(Cns.Sensors, options));
  });
});

Visitando https://localhost:8080 otterremo

mentre visitando https://localhost:8080/api/sensors

Per la parte MQTT aggiungeremo un HostedService aggiungendo il file ConsoleHostedService.cs con il seguente contenuto:

using System.Text.Json;
using broker;
using Microsoft.Extensions.Options;
using MQTTnet;
using MQTTnet.Server;
using Serilog;

internal sealed class ConsoleHostedService(IOptions<AppSettings> appSettings,
                            IHostApplicationLifetime appLifetime) : IHostedService
{
    private readonly AppSettings _appSettings = appSettings.Value;
    private readonly IHostApplicationLifetime _appLifetime = appLifetime;
    private MqttServer? _mqttServer;
    private readonly List<Payload> _history = [];

    public Task StartAsync(CancellationToken cancellationToken)
    {
        _appLifetime.ApplicationStarted.Register(OnStarted);
        _appLifetime.ApplicationStopping.Register(OnStopping);
        _appLifetime.ApplicationStopped.Register(OnStopped);

        return Task.CompletedTask;
    }

    private async void OnStarted()
    {
        try
        {
            Log.Logger.Information("Starting broker service");

            var options = new MqttServerOptionsBuilder()
                .WithDefaultEndpoint()
                .WithDefaultEndpointPort(1883)
                .Build();

            var mqttFactory = new MqttServerFactory();
            _mqttServer = mqttFactory.CreateMqttServer(options);

            _mqttServer.ClientConnectedAsync += e =>
            {
                Log.Logger.Information("Client {clientId} connected", e.ClientId);
                return Task.CompletedTask;
            };

            _mqttServer.ClientDisconnectedAsync += e =>
            {
                Log.Logger.Information("Client {clientId} disconnected", e.ClientId);
                return Task.CompletedTask;
            };

            await _mqttServer.StartAsync();

            var rnd = new Random();

            while (true)
            {
                foreach (var sensor in Cns.Sensors)
                {
                    try
                    {
                        // random int from 0 to 100
                        var value = rnd.Next(101);

                        var payload = new Payload
                        {
                            TimestampUtc = DateTime.UtcNow,
                            SensorName = sensor.SensorName,
                            Value = value,
                        };

                        var jsonOptions = new JsonSerializerOptions
                        {
                            PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
                            WriteIndented = true
                        };

                        _history.Add(payload);
                        if (_history.Where(t => t.SensorName == sensor.SensorName).Count() > _appSettings.SensorHistoryItemSize)
                            _history.Remove(_history.First(t => t.SensorName == sensor.SensorName));

                        var latestValueMessage = new MqttApplicationMessageBuilder()
                            .WithTopic(sensor.LatestValueTopic)
                            .WithPayload(JsonSerializer.Serialize(payload, jsonOptions))
                            .WithRetainFlag(true)
                            .Build();

                        await _mqttServer.InjectApplicationMessage(new InjectedMqttApplicationMessage(latestValueMessage), CancellationToken.None);

                        Log.Logger.Information("Message sended for sensor {sensorName} on topic {topic} with value {value}", sensor.SensorName, sensor.LatestValueTopic, value);

                        var sensorHistory = _history.Where(t => t.SensorName == sensor.SensorName).ToList();

                        var historyMessage = new MqttApplicationMessageBuilder()
                            .WithTopic(sensor.HistoryTopic)
                            .WithPayload(JsonSerializer.Serialize(sensorHistory.SkipLast(1), jsonOptions))
                            .WithRetainFlag(true)
                            .Build();

                        await _mqttServer.InjectApplicationMessage(new InjectedMqttApplicationMessage(historyMessage), CancellationToken.None);

                        Log.Logger.Information("Updated history for sensor {sensorName} on topic {topic} (history size {historySize})", sensor.SensorName, sensor.HistoryTopic, sensorHistory.Count);
                    }
                    finally
                    {
                        Thread.Sleep(1000);
                    }
                }

                Thread.Sleep(5000);
            }
        }
        catch (Exception ex)
        {
            Log.Logger.Error(ex, "Unhandled exception!");
            _appLifetime.StopApplication();
        }
    }

    private void OnStopping()
    {
        Log.Logger.Information("Stopping broker service");
        _mqttServer?.Dispose();
    }

    private void OnStopped()
    {
        Log.Logger.Information("Stopped broker service");
    }


    public Task StopAsync(CancellationToken cancellationToken)
    {
        return Task.CompletedTask;
    }
}

Successivamente aggiungiamo il nuovo HostedService al WebHost (Program.cs), appena dopo la configurazione del logger ed iniettiamo la classe per la configurazione (AppSettings).

...
.ConfigureServices((hostContext, services) =>
{
    services.AddOptions<AppSettings>().Bind(configuration.GetSection("AppSettings"));
    services.AddHostedService<ConsoleHostedService>();
});

Vediamo il codice della logica MQTT più nel dettaglio:

  • Il broker MQTT viene istanzato con le impostazioni di default utilizzando la porta 1883.
  • I due eventi ClientConnectedAsync e ClientDisconnectedAsync ci forniranno informazioni in merito ai client che vengono connessi ed eventualmente disconnessi
var options = new MqttServerOptionsBuilder()
  .WithDefaultEndpoint()
  .WithDefaultEndpointPort(1883)
  .Build();

var mqttFactory = new MqttServerFactory();
_mqttServer = mqttFactory.CreateMqttServer(options);
  
_mqttServer.ClientConnectedAsync += e =>
{
    Log.Logger.Information("Client {clientId} connected", e.ClientId);
    return Task.CompletedTask;
};

_mqttServer.ClientDisconnectedAsync += e =>
{
    Log.Logger.Information("Client {clientId} disconnected", e.ClientId);
    return Task.CompletedTask;
};

await _mqttServer.StartAsync();
  • Successivamente in modo ciclico verranno inviati due messaggi per ogni sensore: uno contenente l’ultimo valore (un intero casuale da 0 a 100)
var latestValueMessage = new MqttApplicationMessageBuilder()
  .WithTopic(sensor.LatestValueTopic)
  .WithPayload(JsonSerializer.Serialize(payload))
  .WithRetainFlag(true)
  .Build();

await _mqttServer.InjectApplicationMessage(new InjectedMqttApplicationMessage(latestValueMessage), CancellationToken.None);
  • ed uno contenente lo storico relativo al sensore corrente escluso l’ultimo valore inviato sul topic latest
var historyMessage = new MqttApplicationMessageBuilder()
  .WithTopic(sensor.HistoryTopic)
  .WithPayload(JsonSerializer.Serialize(sensorHistory.SkipLast(1)))
  .WithRetainFlag(true)
  .Build();

await _mqttServer.InjectApplicationMessage(new InjectedMqttApplicationMessage(historyMessage), CancellationToken.None);
  • Come si può vedere, entrambe i messaggi hanno il flag Retain impostato a TRUE. Questo è strettamente necessario per far si che i client, quando connessi, ricevano sempre l’ultimo messaggio disponibile su entrambe i topic. In questo modo avremo sempre a disposizione l’ultimo valore di temperatura e lo storico, anche se i messaggi sono stati inviati dal broker in un momento precedente rispetto a quando il client ha stabilito la connessione con esso.
  • Nello storico, per ogni sensore, vengono mantenuti N valori. N viene impostato tramite la variabile SensorHistoryItemSize
_history.Add(payload);
if (_history.Where(t => t.SensorName == sensor.SensorName).Count() > _appSettings.SensorHistoryItemSize)
  _history.Remove(_history.First(t => t.SensorName == sensor.SensorName));

A questo punto il nostro server è completato.
Eseguendolo verranno stampate all’interno della console le informazioni di log relative ai messaggi inviati sui vari topic.

Il client

Il client che creeremo, come detto nell’introduzione, sarà un’applicazione Flutter, nello specifico un’applicazione Android. Il framework e le librerie utilizzare sono cross-platform, la guida quindi, rimane valida anche per un’applicazione Flutter iOS.

Nella root di progetto (sensors_example) eseguiamo il comando flutter create -e client --platforms android per creare un progetto vuoto Flutter con il solo template del progetto Android.

Aggiungiamo dentro al file launch.json (cartella .vscode) la configurazione per il l’esecuzione dell’applicativo client

{
    "name": "client",
    "request": "launch",
    "type": "dart",
    "program": "client/lib/main.dart"
}

Eseguendo con F5 otterremo questo risultato

Aggiungiamo al progetto le librerie necessarie

http: ^1.4.0
mqtt_client: ^10.8.0
intl: ^0.20.2

http per la chiamata API dell’elenco dei topic, mqtt_client per connettersi al broker MQTT istanziato tramite il server ed infine intl per lo strumento DateFormat che utilizzeremo per la formattazione della data relativa alla rilevazione della temperatura.

Iniziamo con la creazione dei modelli SensorVM (lib/models/sensor.dart) e PayloadVM (lib/models/payload.dart), rispettivamente per l’elenco dei sensori con le relative informazioni sui topic MQTT ed i messaggi contenenti i valori temperatura rilevati

class SensorVM {
  final String sensorName;
  final String latestValueTopic;
  final String historyTopic;

  SensorVM({
    required this.sensorName,
    required this.latestValueTopic,
    required this.historyTopic,
  });

  Map&lt;String, dynamic> toMap() {
    return {
      'sensorName': sensorName,
      'latestValueTopic': latestValueTopic,
      'historyTopic': historyTopic,
    };
  }

  factory SensorVM.fromMap(Map&lt;String, dynamic> map) {
    return SensorVM(
      sensorName: map['sensorName'],
      latestValueTopic: map['latestValueTopic'],
      historyTopic: map['historyTopic'],
    );
  }
}


class PayloadVM {
  final DateTime timestampUtc;
  final String sensorName;
  final int value;

  PayloadVM({
    required this.timestampUtc,
    required this.sensorName,
    required this.value,
  });

  Map&lt;String, dynamic> toMap() {
    return {
      'timestampUtc': timestampUtc.toIso8601String(),
      'sensorName': sensorName,
      'value': value,
    };
  }

  factory PayloadVM.fromMap(Map&lt;String, dynamic> map) {
    return PayloadVM(
      timestampUtc: DateTime.parse(map['timestampUtc']),
      sensorName: map['sensorName'],
      value: map['value'],
    );
  }
}

Procediamo con la creazione del provider di navigazione (lib/providers/navigation.dart) per la gestione di un NavigatorState condiviso tra le varie pagine

import 'package:flutter/material.dart';

class NavigationProvider {
  NavigationProvider._privateContructor();

  static final NavigationProvider instance =
      NavigationProvider._privateContructor();

  GlobalKey&lt;NavigatorState> navigatorKey = GlobalKey&lt;NavigatorState>();
}

creiamo poi il provider per la gestione delle chiamate API (lib/providers/api.dart)

import 'dart:async';
import 'dart:convert';

import 'package:http/http.dart' as http;

import '../models/sensor.dart';

class ApiProvider {
  ApiProvider._privateContructor();

  static ApiProvider instance = ApiProvider._privateContructor();

  Future&lt;List&lt;SensorVM>> getSensors() async {
    final response = await http.get(Uri.https('10.0.2.2:8080', '/api/sensors'));
    final itemList = jsonDecode(response.body);
    return List.generate(
      itemList.length,
      (index) => SensorVM.fromMap(itemList[index]),
    );
  }
}

L’indirizzo 10.0.2.2 fa riferimento all’indirizzo localhost della macchina di sviluppo che esegue l’emulatore Android (qui i dettagli).

Siamo ora pronti a creare le due pagine: la lista dei sensori ed il dettaglio del singolo sensore.

Iniziamo con la lista dei sensori lib/sensor_list.dart

import 'package:flutter/material.dart';

import 'models/sensor.dart';
import 'providers/api.dart';

class SensorListPage extends StatefulWidget {
  const SensorListPage({super.key});

  @override
  State&lt;StatefulWidget> createState() {
    return _SensorListPageState();
  }
}

class _SensorListPageState extends State&lt;SensorListPage> {
  late List&lt;SensorVM> _sensorList;

  Future&lt;void> initSensorList() async {
    _sensorList = await ApiProvider.instance.getSensors();
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text("Sensor List")),
      body: SafeArea(
        child: FutureBuilder(
          future: initSensorList(),
          builder: (context, snapshot) {
            if (snapshot.connectionState == ConnectionState.done) {
              return ListView.separated(
                padding: const EdgeInsets.all(8),
                itemBuilder: (context, index) {
                  return ListTile(
                    onTap: () async {
                      await Navigator.pushNamed(
                        context,
                        '/sensors/detail',
                        arguments: {"sensor": _sensorList[index]},
                      );
                    },
                    title: Text(_sensorList[index].sensorName),
                  );
                },
                separatorBuilder: (context, index) {
                  return const Divider();
                },
                itemCount: _sensorList.length,
              );
            } else {
              return const Center(child: Text("No data"));
            }
          },
        ),
      ),
    );
  }
}

La pagina contiene una ListView popolata dal Future initSensorList().

Alla pressione di un elemento in lista viene caricata tramite Navigator la pagina di dettaglio con come argomento il sensore selezionato.

Procediamo con la pagina di dettaglio lib/sensor_detail.dart

import 'dart:convert';
import 'package:client/models/payload.dart';
import 'package:flutter/material.dart';
import 'package:intl/intl.dart';
import 'package:mqtt_client/mqtt_client.dart';
import 'package:mqtt_client/mqtt_server_client.dart';

import 'models/sensor.dart';

class SensorDetailPage extends StatefulWidget {
  const SensorDetailPage(this.sensor, {super.key});

  final SensorVM sensor;

  @override
  State&lt;StatefulWidget> createState() {
    return _SensorDetailPageState();
  }
}

class _SensorDetailPageState extends State&lt;SensorDetailPage> {
  late MqttServerClient client;

  final ValueNotifier&lt;PayloadVM?> _latest = ValueNotifier(null);
  final ValueNotifier&lt;List&lt;PayloadVM>> _history = ValueNotifier([]);

  final DateFormat formatter = DateFormat('yyyy-MM-dd HH:mm:ss');

  @override
  void initState() {
    initMqtt();
    super.initState();
  }

  @override
  void dispose() {
    client.disconnect();
    super.dispose();
  }

  initMqtt() async {
    client = MqttServerClient('10.0.2.2', UniqueKey().toString());
    client.setProtocolV311();

    await client.connect();

    client.subscribe(widget.sensor.latestValueTopic, MqttQos.atMostOnce);
    client.subscribe(widget.sensor.historyTopic, MqttQos.atMostOnce);

    client.updates!.listen((List&lt;MqttReceivedMessage&lt;MqttMessage?>>? c) {
      final receivedMessage = c![0];
      final message = receivedMessage.payload as MqttPublishMessage;
      if (receivedMessage.topic == widget.sensor.latestValueTopic) {
        final payload = PayloadVM.fromMap(
          jsonDecode(
                MqttPublishPayload.bytesToStringAsString(
                  message.payload.message,
                ),
              )
              as Map&lt;String, dynamic>,
        );
        _latest.value = payload;
      } else if (receivedMessage.topic == widget.sensor.historyTopic) {
        final decodedPayload =
            jsonDecode(
                  MqttPublishPayload.bytesToStringAsString(
                    message.payload.message,
                  ),
                )
                as List&lt;dynamic>;
        final payload = List.generate(decodedPayload.length, (index) {
          return PayloadVM.fromMap(decodedPayload[index]);
        });
        _history.value = payload;
      }
    });
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: Text("Sensor ${widget.sensor.sensorName}")),
      body: SizedBox(
        height: double.infinity,
        child: Column(
          children: [
            Expanded(
              child: ValueListenableBuilder(
                valueListenable: _history,
                builder: (context, value, child) {
                  final reversedList = value.reversed.toList();
                  return ListView.separated(
                    itemBuilder: (context, index) {
                      return ListTile(
                        title: Text(
                          "${formatter.format(reversedList[index].timestampUtc)} | value: ${reversedList[index].value.toString()}",
                        ),
                      );
                    },
                    separatorBuilder: (context, index) => const Divider(),
                    itemCount: value.length,
                    reverse: true,
                  );
                },
              ),
            ),
            ValueListenableBuilder(
              valueListenable: _latest,
              builder: (context, value, child) {
                if (value != null) {
                  return SizedBox(
                    width: double.infinity,
                    child: Padding(
                      padding: EdgeInsets.all(16.0),
                      child: Column(
                        crossAxisAlignment: CrossAxisAlignment.start,
                        spacing: 4.0,
                        children: [
                          Text("latest"),
                          Text(
                            formatter.format(value.timestampUtc),
                            style: TextStyle(fontSize: 20),
                          ),
                          Text(
                            value.value.toString(),
                            style: TextStyle(fontSize: 80),
                          ),
                        ],
                      ),
                    ),
                  );
                }
                return Text("---", style: TextStyle(fontSize: 80));
              },
            ),
          ],
        ),
      ),
    );
  }
}

All’avvio della pagina viene inizializzato il client MQTT (come già indicato in riferimento al provider API, viene sempre utilizzato l’indirizzo 10.0.2.2 che rimanda all’indirizzo localhost della macchina che esegue l’emulatore).

La porta 1883 è preimpostata e non è quindi necessario indicarla.

client = MqttServerClient('10.0.2.2', UniqueKey().toString());
client.setProtocolV311();

await client.connect();

client.subscribe(widget.sensor.latestValueTopic, MqttQos.atMostOnce);
client.subscribe(widget.sensor.historyTopic, MqttQos.atMostOnce);

Lo StreamSubscription listen viene invocato al ricevimento di messaggi nei topic a cui siamo sottoscritti

client.updates!.listen((List&lt;MqttReceivedMessage&lt;MqttMessage?>>? c) {
  final receivedMessage = c![0];
  final message = receivedMessage.payload as MqttPublishMessage;
  if (receivedMessage.topic == widget.sensor.latestValueTopic) {
    final payload = PayloadVM.fromMap(
      jsonDecode(
            MqttPublishPayload.bytesToStringAsString(
              message.payload.message,
            ),
          )
          as Map&lt;String, dynamic>,
    );
    _latest.value = payload;
  } else if (receivedMessage.topic == widget.sensor.historyTopic) {
    final decodedPayload =
        jsonDecode(
              MqttPublishPayload.bytesToStringAsString(
                message.payload.message,
              ),
            )
            as List&lt;dynamic>;
    final payload = List.generate(decodedPayload.length, (index) {
      return PayloadVM.fromMap(decodedPayload[index]);
    });
    _history.value = payload;
  }
});

I due ValueNotifier _latest ed _history vengono utilizzati per aggiornare gli elementi mostrati a schermo.

La pagina presenta due componenti principali: un widget ListView popolato con lo storico valori ed in fondo, alla schermata, due widget che mostrato la data e ora dell’ultima temperatura rilevata insieme al suo valore.

Per concludere non resta che modificare il file d’ingresso lib/main.dart come segue per impostare il routing alle due pagine create sopra.

import 'dart:io';

import 'package:client/sensor_detail.dart';
import 'package:client/sensor_list.dart';
import 'package:flutter/material.dart';

import 'providers/navigation.dart';

class DevHttpOverrides extends HttpOverrides {
  @override
  HttpClient createHttpClient(SecurityContext? context) {
    return super.createHttpClient(context)
      ..badCertificateCallback = (X509Certificate cert, String host, int port) {
        return true;
      };
  }
}

Future&lt;void> main() async {
  WidgetsFlutterBinding.ensureInitialized();
  HttpOverrides.global = DevHttpOverrides();
  runApp(const MainApp());
}

class MainApp extends StatelessWidget {
  const MainApp({super.key});

  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      navigatorKey: NavigationProvider.instance.navigatorKey,
      initialRoute: '/sensors',
      onGenerateRoute: (settings) {
        final args = (settings.arguments ?? {}) as Map&lt;dynamic, dynamic>;
        Widget? pageWidget;
        switch (settings.name) {
          case '/sensors':
            pageWidget = SensorListPage();
            break;
          case '/sensors/detail':
            pageWidget = SensorDetailPage(args["sensor"]);
          default:
            pageWidget = null;
            break;
        }
        return MaterialPageRoute(builder: (context) => pageWidget!);
      },
    );
  }
}

L’estensione di HttpOverrides DevHttpOverrides è necessaria per non avere errori in merito al certificato SSL di sviluppo Kestrel sull’emulatore Android.

In produzione questa estensione è da evitare per questioni di sicurezza, consiglio di abilitarla o meno in base all’ambiente in cui ci si trova, sviluppo o produzione.

Risultato finale

In primis lanciamo il broker posizionandoci sulla cartella broker appunto ed eseguiamo il comando dotnet run.

Come già visto al paragrafo dedicato, se tutto funziona, otterremo un risultato simile al seguente

Lanciamo l’applicazione Flutter sull’emulatore Android selezionando il profilo di lancio client e premendo F5

Ci troveremo davanti la lista dei sensori disponibili. Selezioniamone uno per visualizzarne il dettaglio

GitHub

Il progetto realizzato in questo articolo è disponibile su GitHub.