Client Not Receiving messages after being disconnnected and then reconnected.

[sameert89]

I have a frontend that subscribes to a topic and listens to messages, in my use-case I establish the connection when the component is shown and end it when it is unloaded. But when the component is reloaded and connect and subscribe happens again, I do not get the components from the queue which were pushed while I was not connected. I only get messages when I am connected. Below is the code from my service:

import { Buffer } from 'buffer';
import mqtt, { IClientOptions, MqttClient } from 'mqtt';
import { Observable } from 'rxjs';
import { environment } from 'src/environments/environment.dev';

import { Injectable } from '@angular/core';

@Injectable({
  providedIn: 'root',
})
export class MqttService {
  public MQTT_SERVICE_OPTIONS: IClientOptions;
  client?: MqttClient;
  public message: string = '';

  constructor() {
    this.MQTT_SERVICE_OPTIONS = {
      hostname: environment.eventGridHostName,
      port: 443,
      path: '/mqtt',
      protocol: 'wss',
      username: "user1",
      clientId: "client1",
      protocolVersion: 5,
      clean: false
    };
  }

  connectMqtt(accessToken: string): Observable<boolean> {
    return new Observable((observer) => {
      mqtt
        .connectAsync({
          ...this.MQTT_SERVICE_OPTIONS,
          properties: {
            authenticationMethod: 'OAUTH2-JWT',
            authenticationData: Buffer.from(accessToken),
          },
        })
        .then((client) => {
          this.client = client;
          this.message = 'Connected';
          observer.next(true);
          observer.complete();
        })
        .catch((error) => {
          observer.error(error);
        });
    });
  }

  disconnectMqtt(): Observable<boolean> {
    return new Observable((observer) => {
      if (this.client) {
        this.client
          .endAsync()
          .then(() => {
            this.message = 'Disconnected';
            observer.next(true);
            observer.complete();
          })
          .catch((error) => {
            observer.error(error);
          });
      } else {
        observer.error(new Error('MQTT client is not connected'));
      }
    });
  }

  subscribeToTopic(topic: string): Observable<EventGridEventDetails> {  // Caller calls this immediately after connect
    if (!this.client) {
      throw new Error('MQTT client is not connected.');
    }

    return new Observable((observer) => {
      this.client!.subscribeAsync(topic, { qos: 1 })
        .then(() => {
          this.client!.on('message', (topic, message) => {
            const eventDetails: EventGridEventDetails = {
              topic: topic,
              content: message.toString(),
            };
            observer.next(eventDetails);
          });
        })
        .catch((error) => {
          observer.error(error);
        });
    });
  }
}

I have set QoS to 1 on the publisher side as well, I don't know what I am doing wrong, any help would be greatly appreciated.

[robertsLando]

In order to do that you should use clean: true

Check: https://www.hivemq.com/blog/mqtt-essentials-part-7-persistent-session-queuing-messages/

Be aware that sometimes what you need is just retained messages and not all messages but I dunno your use case. Using clean true could lead to some issues on broker db expecially if you have lot of clients using clean true with id that changes frequently and nothing that clears offline messages on broker db.

[sameert89]

Clean session/start means a non persistent session which means the queue is discarded by the broker on client disconnect. AFAIK I need a persistent connection so the broker maintains the session:

From the Link You Shared