Mqtt

Dotnet

Angular

Event Driven Architecture

Typescript

Build Real-Time Pub/Sub Messaging with MQTT, EMQX, .NET, and Angular

20 min readMar 16, 2026

A hands-on guide to building client-server Pub/Sub messaging with MQTT and EMQX across ASP.NET, VB.NET, and Angular. We explore the basics of MQTT, QoS levels, durable subscriptions, subtopics, and practical patterns for building reliable, message-driven applications.

What is MQTT?

Message Queuing Telemetry Transport (or MQTT for short) is a standard publish/subscribe messaging protocol for the Internet of Things (IoT). It is designed as an extremely lightweight messaging transport, ideal for connecting remote devices to servers, web, or mobile applications.

While MQTT is not suited for large payloads, complex routing patterns (such as those in RabbitMQ), or long‑term message retention (as provided by Kafka), its simplicity, efficiency, and broad device support have made it a leading choice for IoT messaging. More on MQTT.

Setting up MQTT

There are several online services that provide MQTT hosting, such as AmazonMQ and AWS IoT, but for this blog, I decided to go with an MQTT-specific platform. There are a few popular choices:

The broker I’ll be using is EMQX to integrate a few ASP.NET APIs with a VB.NET and Angular frontend. There will be multiple publishers and subscribers:

Publishers: ASP.NET WEB API, Angular web app
Subscribers: Another ASP.NET Web API, Angular web app, VB.NET desktop app

Publishers and subscribers communicate via messaging topics; a message published to one topic is immediately received by all active subscribers.

MQTT broker setup via EMQX

Go to the official EMQX site and create an account. If you’re completely new, you can read their official introduction article.

Once you log into the EMQX console and choose to create a new project, you’ll be prompted to enter its name and the note:

I named mine MQTT Starter.

On the next screen, you’ll pick the price plan. Every tier requires you to add a payment method. The Serverless tier will not charge you unless you exceed its limits.

Another important part of this screen is the configuration tab in the sidebar (to the right). It shows you usage and tier limits, uptime, and the ports you’ll connect to.

The serverless region defaults to North America

MQTT uses port 8883 to establish connections between servers and desktop and mobile apps.
Web browsers do not support the MQTT port. WebSocket port (8084) provided by EMQX is used to establish a connection between the broker and the web client (e.g., EMQX and Angular web app).

The image illustrates use of protocols on various devices

In addition, each tier requires a TLS/SSL secure connection. To connect to EMQ, you’ll use the provided SSL certificate, along with the broker connection string and user credentials.

Once you add the payment method, the sidebar button will change text to Deploy.

The broker is up and running.

If you click on the broker, it will take you to the dashboard page, where you’ll find usage analytics of your broker and, more importantly, how to connect to it:

The address is your broker URI
Ports (MQTT & WebSocket) exposed on the broker
CA Certificate is a downloadable TLS/SSL certificate required to connect to the broker.

Broker User

Expand the Access Control side nav and click Authentication. This is where you’ll add a username and password of the user who will connect to the broker.

Next, click on the Authorization option. Head over to the All users tab and click Add. This is where you’ll create a topic and grant all users write-read (pub-sub) access to it.

Testing connection

To test the connection before writing the code, head over to the Online Test (on the side nav) and connect to the broker using the previously created broker user.

Enter the name of the topic you’ve previously created (in the Authorization tab) that you’ll publish to, and use the same topic name on the subscriber.

Once you publish the message, it will appear on the Messages screen.

When you’re done with the online test, click on the Disconnect button.

Creating an MQTT Publisher via ASP.NET Web API

I created a new .NET WEB API project through Visual Studio (2022). I’m using .NET 8.
Open the NuGet Package Manager and install the latest compatible version of MQTTnet.

Create a service interface. You’ll invoke this interface from the controller,

 public interface IMQTTService
 {
  Task<string?> PublishToTopic<T>(T blog);
 }

Next, create a service class that implements this interface.

 public class MQTTService : IMQTTService
 {
  private readonly string _broker_uri = "<YOUR-EMQX-BROKER-URI>";
  private readonly int _mqtt_port = 8883;
  private readonly string _clientId = "Publisher-" + Guid.NewGuid().ToString("N");

  private readonly string _username = "USERNAME";
  private readonly string _password = "PASSWORD";
  private readonly string _cert_file = "CA/Serverless/emqxsl-ca.crt";

  private IMqttClient _mqttClient;
  private MqttClientConnectResult _mqttClientConnectResult;

  public MQTTService()
  {
   ConnectToBroker();
  }
}

As you can see, you need to provide the connection string and the credentials you created on the EMQX console. In addition, there is the path to the certificate file, which I pasted in the solution.

Connect to the broker

Next, create a method that connects to the broker,

  private void ConnectToBroker()
  {
   try
   {
    // Create a MQTT client factory
    var factory = new MqttClientFactory();
    _mqttClient = factory.CreateMqttClient();

    // Load the certificate file
    var caCertBytes = File.ReadAllBytes(_cert_file);
    var caCert = new X509Certificate2(caCertBytes);

    var options = new MqttClientOptionsBuilder()
     .WithTcpServer(_broker_uri, _mqtt_port)
     .WithCredentials(_username, _password)
     .WithClientId(_clientId) // Unique id
     .WithCleanSession() // - No previous subscriptions restored
     .WithTlsOptions(o =>
     {
      o.UseTls(true);
      o.WithSslProtocols(System.Security.Authentication.SslProtocols.Tls12);

      o.WithCertificateValidationHandler(ctx =>
      {
       // Convert server certificate to X509Certificate2
       var serverCert = new X509Certificate2(ctx.Certificate);

       // Add EMQX CA to the chain
       ctx.Chain.ChainPolicy.ExtraStore.Add(caCert);

       // Allow unknown CA (because EMQX uses custom CA)
       ctx.Chain.ChainPolicy.VerificationFlags =
        X509VerificationFlags.AllowUnknownCertificateAuthority;

       // Disable revocation check (EMQX CA doesn't support it)
       ctx.Chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck;

       // Validate the chain
       return ctx.Chain.Build(serverCert);
      });
     })
     .Build();

    // Save the result in the reusable property
    _mqttClientConnectResult = _mqttClient.ConnectAsync(options).Result;
   }
   catch (Exception ex)
   {
    _connectionError = ex.Message;
    throw;
   }
  }

Publish the message

Create a static method that builds a message based on the provided topic and payload:

  private static MqttApplicationMessage BuildMessage(string topic, string payload)
  {
   return new MqttApplicationMessageBuilder()
     .WithTopic(topic) // push to provided destination
     .WithPayload(payload) // payload is prepared
     .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce) // fire and forget mode
     .WithRetainFlag(false) // message will not be retained for first time subscribers
     .Build();
  }

Implement the interface by creating the PublishToTopic() method:

  public async Task<string?> PublishToTopic<T>(T blog)
  {
   try
   {
    // reload the connection if the broker disconnects
    if (_mqttClient is null || !_mqttClient.IsConnected)
    {
     ConnectToBroker();
     return "Reconnecting... Try again!";
    }

    var payload = JsonSerializer.Serialize(blog);
    var topic = "blogs";

    var blogMessaage = BuildMessage(topic, payload);
    await _mqttClient.PublishAsync(blogMessaage);

    return null;
   }
   catch (Exception ex)
   {
    return ex.Message;
   }
  }

Create an entry point

If this were a console application, you could publish a message from the CLI input, but since I created a WEB API, I’ll add a new controller endpoint that invokes the MQTT service.

Go to Program.cs and register the service:

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddSingleton<IMQTTService, MQTTService>(); // 👈👈

builder.Services.AddControllers();

//...

Then, create a new API controller that invokes the service method:

 [Route("api/[controller]")]
 [ApiController]
 public class BlogsController(IMQTTService mqttService) : ControllerBase
 {

  [HttpPost("CreateBlog")]
  public async Task<ActionResult<Blog>> CreateBlog([FromBody] NewBlogDTO newBlog)
  {
   var errorMessage = await mqttService.PublishToTopic(newBlog);

   return errorMessage is null ?
     Ok("The message was published successfully!") :
     StatusCode(500, errorMessage);
  }

 }
}

If you go to the EMQX project console, click metrics, then clients, you’ll see your published items in the connected clients table.

Where are the published messages now? Are they sitting in some queue waiting to be consumed?
- No. These messages are gone forever.

Unlike queues, topics do not retain messages for future subscribers. That’s why you need to create an active subscription to the topic to see the published messages immediately.

ASP.NET Subscriber

Create a new .NET Web API project that subscribes to the blogs topic and handles incoming messages. You need to use a .NET background service class to keep the subscriber alive during the lifetime of the service.

To do so, create a custom class that inherits the BackgroundService abstract class that .NET provides. Be sure to also install MQTTnet in this solution.

namespace MQTTSubscriber.Services
{
 public class BlogSubsciber : BackgroundService
 {

  private readonly string _broker_uri = "<YOUR-EMQ-BROKER-URI>";
  private readonly int _mqtt_port = 8883;
  private readonly string _clientId = "Blog-Subscriber-2026";
  private readonly string _username = "USERNAME";
  private readonly string _password = "PASSWORD";
  private readonly string _cert_file = "CA/Serverless/emqxsl-ca.crt";

  private IMqttClient _mqttClient;

  protected override Task ExecuteAsync(CancellationToken stoppingToken)
  {
   throw new NotImplementedException();
  }
 }
}

The ExecuteAsync() The method is responsible for running the background process. But you also need to connect to the broker and handle the incoming messages. For that reason, I separated the logic into two lifecycle methods:

StartAsync()
ExecuteAsync()

C# Lifecycle methods

The StartAsync() starts first and connects to the broker:

  public override async Task StartAsync(CancellationToken cancellationToken)
  {
   await ConnectToBroker();
   await base.StartAsync(cancellationToken);
  }

The ExecuteAsync() method keeps the background service running indefinitely.

  protected override async Task ExecuteAsync(CancellationToken stoppingToken)
  {
   try
   {
    while (!stoppingToken.IsCancellationRequested)
    {
     await Task.Delay(Timeout.Infinite, stoppingToken);
    }
   }
   catch (OperationCanceledException)
   {
    logger.LogInformation("Operation was gracefully canceled.");
   }
   catch (Exception ex)
   {
    logger.LogError(ex, "Fatal Exception!");
   }
  }

So, where is this StartAsync() method coming from, and how do you know the lifecycle method order? It comes from the IHostedService. The BackgroundService class you inherit already implements this interface,

namespace Microsoft.Extensions.Hosting
{
    /// <summary>
    /// Base class for implementing a long running <see cref="IHostedService"/>.
    /// </summary>
    public abstract class BackgroundService : IHostedService, IDisposable
    {
        protected abstract Task ExecuteAsync(CancellationToken stoppingToken);

        public virtual Task StartAsync(CancellationToken cancellationToken) {}
    }
}

The StartAsync() is triggered when the application host is ready to start the service, before the worker loop begins (ExecuteAsync()). In most cases ExecuteAsync() is sufficient, hence why you can just inherit the BackgroundService instead of implementing the IHostedService.

Connect to the broker

This is where you pass the credentials, load up the certificate, and connect to the client:

  private async Task ConnectToBroker()
  {
   try
   {
    // Create factory
    var factory = new MqttClientFactory();
    _mqttClient = factory.CreateMqttClient();

    // Load CA file
    var caCertBytes = File.ReadAllBytes(_cert_file);
    var caCert = new X509Certificate2(caCertBytes);

    var options = new MqttClientOptionsBuilder()
     .WithTcpServer(_broker_uri, _mqtt_port)
     .WithCredentials(_username, _password)
     .WithClientId(_clientId)
     .WithCleanSession(true) // Don't preserve previous messages
     .WithTlsOptions(o =>
     {
      o.UseTls(true);
      o.WithSslProtocols(System.Security.Authentication.SslProtocols.Tls12);

      o.WithCertificateValidationHandler(ctx =>
      {
       var serverCert = new X509Certificate2(ctx.Certificate);
       ctx.Chain.ChainPolicy.ExtraStore.Add(caCert);
       ctx.Chain.ChainPolicy.VerificationFlags =
        X509VerificationFlags.AllowUnknownCertificateAuthority;
       ctx.Chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck;
       return ctx.Chain.Build(serverCert);
      });
     })
     .Build();

     // Connects to EMQX
     await _mqttClient.ConnectAsync(options);

    /* TODO subscribe to incoming messages */

   }
   catch (Exception ex)
   {
    logger.LogError(ex, "Fatal Exception!");
    throw;
   }

The MQTT client provides handlers for connecting, disconnecting, and subscribing to incoming messages. This means you do not have to write these manually.

// Inside ConnectToBroker in between .Build() and the end of the try block

   // Resubscribe on reconnect
   _mqttClient.ConnectedAsync += async e =>
    {
     logger.LogInformation("Connected. Subscribing to durable topic...");
     await _mqttClient.SubscribeAsync("blogs");
    };

    // Auto reconnect
    _mqttClient.DisconnectedAsync += async e =>
    {
     logger.LogWarning("Disconnected. Reconnecting in 5 seconds...");
     await Task.Delay(5000);
     await _mqttClient.ConnectAsync(options);
    };

    // Handle the incoming message
    _mqttClient.ApplicationMessageReceivedAsync += e =>
    {
     var topic = e.ApplicationMessage.Topic;
     var payloadString = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);

     // Read the message
     var blog = JsonSerializer.Deserialize<Blog>(payloadString);
     logger.LogDebug(blog?.Title);

     // Send ACK
     return Task.CompletedTask;
    };

    // Connects to EMQX
    await _mqttClient.ConnectAsync(options);

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddControllers();
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();

builder.Services.AddHostedService<BlogSubscriber>(); // 👈👈

var app = builder.Build();

Run the app. Your subscriber should appear in the EMQX console:

Send a message from one point and observe the other.

ASP.NET Publisher => ASP.NET Subscriber demo

The subscriber receives messages as expected.

VB .NET Client

This part demonstrates how to set up MQTT in the desktop application built with Visual Basic .NET. The VB client will serve only as the topic subscriber, while the publisher will remain the ASP .NET WEB API.

Create a new empty application using Visual Studio 2022 (.NET 8):

<!-- vb.net vbproj -->

<Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
    <OutputType>WinExe</OutputType>
    <TargetFramework>net8.0-windows</TargetFramework>
    <StartupObject>Sub Main</StartupObject>
    <UseWindowsForms>true</UseWindowsForms>
    <MyType>WindowsForms</MyType>
  </PropertyGroup>
 ...

</Project>

Then install the MQTTnet via NuGet. The compatible version is 4.3.0:

<!-- vb.net vbproj -->

<ItemGroup>
    <PackageReference Include="MQTTnet" Version="4.3.0.858" />
</ItemGroup>

In addition to installing, you’ll need to include the SSL certificate. I added mine in the directory Solution/CA/emqxsl-ca.crt:

<!-- vb.net vbproj -->

  <ItemGroup>
    <None Update="CA\emqxsl-ca.crt">
      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
    </None>
    <None Update="My Project\Application.myapp">
      <Generator>MyApplicationCodeGenerator</Generator>
      <LastGenOutput>Application.Designer.vb</LastGenOutput>
    </None>
  </ItemGroup>

  <ItemGroup>
    <Folder Include="CA\" />
  </ItemGroup>

Create a Subscriber

Start off by setting up the broker credentials in your main form file:

Imports MQTTnet
Imports MQTTnet.Client
Imports System.Security.Cryptography.X509Certificates
Imports System.Text

Public Class Form1

 Private mqttClient As IMqttClient

 Private broker = "<YOUR-EMQX-BROKER-URI>"
 Private port = 8883
 Private username = "USERNAME"
 Private password = "PASSWORD"
 Private topic = "blogs"
 Private clientId = "vb-client-" & Guid.NewGuid().ToString("N")

End Class

I created a Blog entity that will arrive from the topic:

Public Class Blog
    Public Property AuthorID As Integer
    Public Property Title As String
    Public Property Content As String
End Class

I also added the button on the screen that connects to EMQX when clicked:

 Private Async Sub Button1_Click(sender As Object, e As EventArgs) Handles Button1.Click
  Await ConnectToBroker()
  MessageBox.Show("Connected to MQTT!")
 End Sub

The following code connects to the broker, loads the certificate file, subscribes to the topic, and handles the incoming messages:

 Private Async Function ConnectToBroker() As Task
  Dim factory = New MqttFactory()
  mqttClient = factory.CreateMqttClient()

  ' Read certificate file
  Dim caCertBytes = IO.File.ReadAllBytes("CA/emqxsl-ca.crt")
  Dim caCert = New X509Certificate2(caCertBytes)

  ' Load CA certificate
  Dim tls = New MqttClientOptionsBuilderTlsParameters() With {
   .UseTls = True,
   .SslProtocol = Security.Authentication.SslProtocols.Tls12,
   .AllowUntrustedCertificates = True,
   .IgnoreCertificateChainErrors = True,
   .IgnoreCertificateRevocationErrors = True,
   .CertificateValidationHandler =
    Function(ctx)
     Dim serverCert As New X509Certificate2(ctx.Certificate.GetRawCertData())

     ctx.Chain.ChainPolicy.ExtraStore.Add(caCert)
     ctx.Chain.ChainPolicy.VerificationFlags =
      X509VerificationFlags.AllowUnknownCertificateAuthority
     ctx.Chain.ChainPolicy.RevocationMode =
      X509RevocationMode.NoCheck

     Return ctx.Chain.Build(serverCert)
    End Function
  }

  ' Connect to the broker
  Dim options = New MqttClientOptionsBuilder().
   WithTcpServer(broker, port).
   WithCredentials(username, password).
   WithClientId(clientId).
   WithCleanSession().
   WithTls(tls).
   Build()

  ' Connect to the client
  Await mqttClient.ConnectAsync(options)

  ' Subscribe to topic
  Await mqttClient.SubscribeAsync(
   New MqttTopicFilterBuilder().
    WithTopic(topic).
    WithAtMostOnceQoS().
    Build()
  )

  ' Handle incoming messages
  AddHandler mqttClient.ApplicationMessageReceivedAsync,
   Function(e)
    ' Deserialize incoming messages into Blog entity
    Dim payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload)
    Dim blog As Blog = System.Text.Json.JsonSerializer.Deserialize(Of Blog)(payload)

    MessageBox.Show($"Title: {blog.Title}")

    Return Task.CompletedTask
   End Function
 End Function

Angular MQTT Client

This chapter focuses on setting up an MQTT publisher and subscriber in Angular. The versions I’m using:

Node.js: v22.21.1
Angular: v21.0.4
NPM: v8.6.0

Install dependencies

Install the MQTT client from NPM:

> npm install ngx-mqtt --save

After installation, the package should appear in the package.json file:

{
   "dependencies": {
     ...
     "ngx-mqtt": "^17.0.0",
   },
}

The Ngx-MQTT package is quite outdated and only supports module-based components. So if you’re using newer versions of Angular (and standalone components), you’ll need to switch to the old architecture.

To make use of the old component style, you need to install the following package:

"@angular/platform-browser-dynamic": "^21.1.5", // because I'm on v21

And use it to bootstrap the project in the main.ts file:

import { platformBrowserDynamic } from '@angular/platform-browser-dynamic';
import { AppModule } from './app/app.module';

platformBrowserDynamic().bootstrapModule(AppModule)
  .catch(err => console.error(err));

MQTT Config

Create a new file that will hold all the MQTT configuration:

// model/mqtt-connection.ts

import { IMqttServiceOptions } from 'ngx-mqtt';

export const mqttConnectionOptions: IMqttServiceOptions = {
  protocol: 'wss', // WebSocket protocol
  hostname: '<YOUR-EMQX-BROKER-URI>',
  port: 8084, // default wss port
  path: '/mqtt', // default path (set by the broker)
  clean: true, // subscriber is not durable
  connectTimeout: 4000, // default
  reconnectPeriod: 4000, // default
  clientId: 'angular-client-2026', // whatever you set
  username: 'USERNAME',
  password: 'PASSWORD',
}

You’ll import this file into the app.module.ts file as well as the service file:

// app.module.ts

import { NgModule } from '@angular/core';
import { BrowserModule } from '@angular/platform-browser';

import { MqttModule } from 'ngx-mqtt';
import { App } from './app';
import { mqttConnectionOptions } from './model/mqtt-connection';

@NgModule({
  declarations: [
    App
  ],
  imports: [
    BrowserModule,
    MqttModule.forRoot(mqttConnectionOptions)
  ],
  providers: [],
  bootstrap: [App]
})
export class AppModule { }

Angular Service

This is where you write business logic regarding the MQTT broker:

// services/connect-to-mqtt.service.ts

import { Injectable } from '@angular/core';
import { IMqttMessage, IPublishOptions, MqttService } from 'ngx-mqtt';
import { Observable } from 'rxjs';
import { mqttConnectionOptions } from '../model/mqtt-connection';
import { QosLevel } from '../model/qos';

@Injectable({
  providedIn: 'root',
})
export class ConnectToMqttService {

  constructor(private mqttService: MqttService) {
    this.createConnection();
  }

  // Create a connection
  private createConnection() {
    try {
      this.mqttService.connect(mqttConnectionOptions);
    } catch (error) {
      console.error('mqtt.connect error', error);
      return;
    }

    this.mqttService.onConnect.subscribe(() => {
      console.log('Connection succeeded!');
    });

    this.mqttService.onError.subscribe((error: any) => {
      console.error('Connection failed', error);
    });
  }
}

Angular Subscriber

The Angular app will subscribe to the broker and parse incoming messages.
Add a service method that subscribes to the provided topic and QoS level:

// services/connect-to-mqtt.service.ts

subscribeToTopic(topic: string, qos: QosLevel): Observable<IMqttMessage> {
  return this.mqttService
    .observe(topic, { qos } as any);
}

// model/qos.ts

export type QosLevel = 0 | 1 | 2;

Invoke the service method in the App component file:

// app.ts

import { Component, OnInit, signal } from '@angular/core';
import { IMqttMessage } from 'ngx-mqtt';
import { skip } from 'rxjs';
import { IBlog } from './model/blog';
import { ConnectToMqttService } from './services/connect-to-mqtt.service';

@Component({
  selector: 'app-root',
  templateUrl: './app.html',
  styleUrls: ['./app.scss'],
  standalone: false
})
export class App implements OnInit {
  constructor(private mqttService: ConnectToMqttService) {}

  // loads up on app startup
  ngOnInit(): void {
    this.mqttService.subscribeToTopic('blogs', 0)
      .subscribe((message: IMqttMessage) => {
        const jsonPayload = message.payload.toString();
        const blog: IBlog = JSON.parse(jsonPayload);

        alert(`Title: ${blog.Title}`);
      });
  }

}

// model/blog.ts

export interface IBlog {
  AuthorID: number;
  Title: string;
  Description: string;
}

Run the Angular application:

> ng serve

Once the application boots up, you’ll see the message “Connection succeeded!” printed in the console. If you go to the EMQX console, you’ll see your Angular app listed under connected clients.

If I publish the message from the .NET client, it should appear in the Angular web app.

Of course, if you turn on both the ASP.NET subscriber and the Angular subscriber, the message will be delivered to both destinations.

Unsubscribe

You can destroy the MQTT connection by manually calling the disconnect method:

// services/connect-to-mqtt.service.ts

  destroyConnection() {
    try {
      this.mqttService.disconnect(true);
      console.log('Successfully disconnected!');
    } catch (error: any) {
      console.error('Disconnect failed', error.toString());
    }
  }

Angular Publisher —> ASP .NET Subscriber

You can turn the tables by publishing Angular messages to all active subscribers.
Create a service method that receives a topic, payload, and QoS level, and publishes to MQTT.

// services/connect-to-mqtt.service.ts

  sendMessage(topic: string, payload: string, qos: QosLevel = 0) {
    this.mqttService?
      .unsafePublish(topic, payload, { qos } as IPublishOptions);
  }

Here I’m using the unsafePublish() method because my Angular app is running on localhost without a secure connection. Otherwise, I’d use the publish() method.

Call the service method from the component:

// app.ts

  sendMessage(): void {
    const blog  = {
      AuthorID: 123,
      Title: "Summer",
      Description: "Heat"
    } as IBlog

    const payload = JSON.stringify(blog);

    this.mqttService.sendMessage('blogs', payload);
  }

This message will also be received by the Angular app, as it too is a subscriber to the same topic.

SubTopics

Subtopics are used to target only a subset of subscribers for a given topic.

Subscribers who subscribed to the particular subtopic receive only the messages published to that subtopic:

topic: blogs/games — received by the game subscriber, as well as the all-blogs subscriber

// Publisher
    var gamesBlogMessaage = BuildMessage("blogs/games", payload);
    await _mqttClient.PublishAsync(gamesBlogMessaage);

// Subscribers
    await _mqttClient.SubscribeAsync("blogs/games");
    await _mqttClient.SubscribeAsync("blogs/#");

topic: blogs/dev — received by the dev subscriber, as well as the all-blogs subscriber

// Publisher
    var devBlogMessaage = BuildMessage("blogs/dev", payload);
    await _mqttClient.PublishAsync(devBlogMessaage);

// Subscribers
    await _mqttClient.SubscribeAsync("blogs/dev");
    await _mqttClient.SubscribeAsync("blogs/#");

topic: blogs — received by the all-blogs subscriber

// Publisher
    var allBlogsMessage = BuildMessage("blogs", payload);
    await _mqttClient.PublishAsync(allBlogsMessage);

// Subscriber
    await _mqttClient.SubscribeAsync("blogs/#");

topic: ABCDEF — not received by anyone

This works both for Angular and .NET.

Quality of Service

You’ve probably asked yourself, what is this QoS level I mention every so often and perhaps a couple of other questions:

What happens when the message is not delivered?
Is there a retry mechanism?
What happens to messages sent while subscribers were inactive?

All this is handled using the Quality of Service mechanism.

The Quality of Service (or QoS) defines the guarantees for message delivery between producers, brokers, and consumers in distributed systems. There are three QoS levels:

At most once (QoS=0)

This is a fire-and-forget approach: the messages are published, and no confirmation is expected. Messages may be lost if a failure occurs, but they are never redelivered. (We’ve used this thus far)

At least once (QoS=1)

The producer can send the same message multiple times. This is useful for reprocessing messages that weren’t delivered (i.e., weren’t acknowledged).
The producer guarantees delivery but acknowledges that duplicate messages might occur.

Exactly once (QoS=2)

Ensures each message is delivered and processed exactly once, even in the face of failures. This is the highest level of reliability, ensuring no messages are lost or duplicated.

Dedicated broker

To make this work, I created a dedicated MQTT server using the premium model. There is a 14-day free trial. To follow along, head over to the EMQX Cloud Console and create a new project

Head over to the new deployment page and choose a dedicated server. Also, choose the deployment region closest to you.

The new server is up and running.

Click on the box, and it will take you to the project dashboard, where you’ll once again download the certificate, create a username and password, and copy the connection string:

Building a durable subscriber

The durable subscriber retains messages that were sent while the subscriber was inactive. Under the hood, there is a queue between the topic and the subscriber that stores messages.

With this in place, subscribers can go inactive, resubscribe, and pick up any messages they have missed, as well as process incoming messages.

Persistent ASP .NET Publisher

Here are a few tweaks to the publisher to make it persistent.

Update the MqttClientOptionsBuilder to not start a clean session on reconnect:

    var options = new MqttClientOptionsBuilder()
     .WithTcpServer(_broker_uri, _mqtt_port)
     .WithCredentials(_username, _password)
     .WithClientId(_clientId)
     .WithCleanSession(false) // <-- keep the same session
     .WithTlsOptions(o => {...});

2. Increase the Quality of Service level in the message builder:

  private static MqttApplicationMessage BuildMessage(string topic, string payload)
  {
   return new MqttApplicationMessageBuilder()
     .WithTopic(topic)
     .WithPayload(payload)
     .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) // 👈👈
     .WithRetainFlag(false)
     .Build();
  }

Durable ASP .NET Subscriber

Likewise, update the MqttClientOptionsBuilder in the subscriber as well:

    var options = new MqttClientOptionsBuilder()
     .WithTcpServer(_broker_uri, _mqtt_port)
     .WithCredentials(_username, _password)
     .WithClientId(_clientId)
     .WithCleanSession(false) // Keep the old session
     .WithSessionExpiryInterval(uint.MaxValue) // Do not let session expire
     .WithTlsOptions(o =>

The default expiry interval is zero. If you set it to uint.MaxValue like I have, it will never expire.

2. Update the ConnectedAsync() callback function to include the QoS level in the subscription:

    _mqttClient.ConnectedAsync += async e =>
    {
     logger.LogInformation("Connected. Subscribing to durable topic...");

     await _mqttClient.SubscribeAsync(
      new MqttTopicFilterBuilder()
       .WithTopic(_topic)
       .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) // 👈👈
       .Build()
     );
    };

3. Set your client ID as static:

  private readonly string _clientId = "Durable-Subscriber-7072"; // Fixed ID

Run both services. Head over to the EMQX console. Both services should appear in the clients table with:

clean start set to false
expiry interval (for subscriber) set to Never Expire.

Now, turn off the subscriber service and publish a few messages from the publisher service. Then restart the subscriber and observe the previous messages arrive in the order they were published.

Notice, if you shut down the subscriber service, the durable subscriber will remain in the clients table:

If you click on it and head over to the Subscriptions tab, you’ll notice it displays the topic it’s subscribed to (blogs/games) as well as QoS level (AtLeastOnce).

Durable Angular subscriber

There are a few things to do to keep the Angular subscriber active:

Change the connection session from clean to not clean:

export const mqttConnectionOptions: IMqttServiceOptions = {
  protocol: 'wss',
  hostname: '<YOUR-EMQX-BROKER-URI>',
  port: 8084,
  path: '/mqtt',
  clean: false, // 👈👈
  connectTimeout: 4000,
  reconnectPeriod: 4000,
  clientId: 'angular-client-2027', // whatever you set
  username: 'USERNAME',
  password: 'PASSWORD',
}

2. Set the QoS level to At Least Once/Exactly Once in the component subscription:

  ngOnInit(): void {
    this.mqttService.subscribeToTopic('blogs/games', 1) // at least once
    .subscribe((message: IMqttMessage) => {
      const jsonPayload = message.payload.toString();
      const blog: IBlog = JSON.parse(jsonPayload);

      alert(`Title: ${blog.Title}`);
    });
  }

After starting the server, you should see the Angular subscriber in the EMQX clients table

Now, shut down the server, the tab, the browser. Whatever you desire.

> ^C

And publish a few messages from the ASP.NET publisher. The messages should appear in the order they were sent once the server is up again.

Every durable subscriber has it’s own queue.

The messages picked up by the Angular subscriber will also arrive at the .NET subscriber when it comes back alive:

Quick Note:

If your subscription isn’t working properly, it could be that you’re using the same subscription across multiple processes. Just rename the client ID to something else, and it will work again.

Retry mechanism

The retry mechanism is enabled for QoS levels greater than zero:

At least once or
Exactly once

By default, the broker acknowledges the message as soon as you return the completed task:

    // Handle the incoming message
    _mqttClient.ApplicationMessageReceivedAsync += e =>
    {
     var topic = e.ApplicationMessage.Topic;
     var payloadString = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);

     logger.LogDebug("Received message on {topic}, {payloadString}", topic, payloadString);

     var blog = JsonSerializer.Deserialize<Blog>(payloadString);

     return Task.CompletedTask; // ACK
    };

If you decide to throw an exception,

    // Handle the incoming message
    _mqttClient.ApplicationMessageReceivedAsync += e =>
    {
     var topic = e.ApplicationMessage.Topic;
     var payloadString = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);

     logger.LogDebug("Received message on {topic}, {payloadString}", topic, payloadString);

     var blog = JsonSerializer.Deserialize<Blog>(payloadString);

     if (blog?.AuthorID < 1)
     {
      throw new Exception("Invalid author id!");
     }

     return Task.CompletedTask; // ACK
    };

The message sent will be returned to the queue and be consumed again every 10 to 15 seconds. The EMQX keeps the message loop repeating until either

Subscriber acknowledges
Session expires
Message expires (TTL)
The broker is restarted.

Clean up

Lastly, once you’re done, don’t forget to shut down EMQX instances you no longer need.

Confirm that the broker has stopped on the project page.

Now, click on the project once again, click on the settings button on the far right and delete the deployment

You can delete a project, too.

Summary

This was a comprehensive guide to MQTT messaging with EMQX and various technologies. For more on EMQX, be sure to visit their official site. For everything else, Messaging, .NET, Angular-related, stick around on my blog.

I’ll include a GitHub repository with all examples in a few days.

Bye for now 👋

Build Real-Time Pub/Sub Messaging with MQTT, EMQX, .NET, and Angular

What is MQTT?

Setting up MQTT

MQTT broker setup via EMQX

Broker User

Testing connection

Creating an MQTT Publisher via ASP.NET Web API

Connect to the broker

Publish the message

Create an entry point

ASP.NET Subscriber

C# Lifecycle methods

Connect to the broker

VB .NET Client

Create a Subscriber

Angular MQTT Client

Install dependencies

MQTT Config

Angular Service

Angular Subscriber

Unsubscribe

Angular Publisher —> ASP .NET Subscriber

SubTopics

Quality of Service

At most once (QoS=0)

At least once (QoS=1)

Exactly once (QoS=2)

Dedicated broker

Building a durable subscriber

Persistent ASP .NET Publisher

Durable ASP .NET Subscriber

Durable Angular subscriber

Quick Note:

Retry mechanism

Clean up

Summary

Written by Mirza Leka