How I can handle TS types error MQTT.js with RxJs

[Vetnametz]

I have a problem in attempt to use rxjs with mqtt.js

import { connect, IClientOptions, IPublishPacket, Packet } from "mqtt";
import { fromEvent } from "rxjs";
import { bufferTime, filter, map, takeUntil } from "rxjs/operators";

const powerLinesMQTTClient = connect(MQTT_BROKER_URL!, options);

const connect$ = fromEvent(powerLinesMQTTClient, "connect");
const disconnect$ = fromEvent(powerLinesMQTTClient, "disconnect");
const error$ = fromEvent(powerLinesMQTTClient, "error");
const message$ = fromEvent(powerLinesMQTTClient, "message");

connect$.subscribe(onConnect);
disconnect$.subscribe(onDisconnect);
error$.subscribe(onError);
message$
  .pipe(
    filter(([topic]: [string, Buffer, IPublishPacket]) => {
      Logger.log("debug", `[MQTT] PowerLinesMQTTClient subscribed to ${topic}`);
      const { fieldName } = messagesUtils.parseTopic(topic);

      if (Object.values(SupportedTopics).includes(fieldName as SupportedTopics)) {
        return true;
      }
      Logger.log(
        "warn",
        `[MQTT] Warning: this type of messages: ${fieldName} is not handling by PowerLinesMQTTClient`
      );
      return false;
    })
  )
  .pipe(
    map(([topic, payload]): MessageTS => {
      const { fieldName, hostName } = messagesUtils.parseTopic(topic);

and I can`t find out how I can fix it

It seams its a mqtt.js types problem ... any ideas how I can handle this?

ts-config

{
  "compilerOptions": {
    /* Basic Options */
    "target": "es2018",
    "module": "esNext",
    "lib": ["esnext", "dom"],
    "allowJs": true,
    "checkJs": false,
    "jsx": "preserve",
    "incremental": true,
    "noEmit": true,

    /* Strict Type-Checking Options */
    "strict": true,
    "noImplicitAny": true,
    "strictNullChecks": true,

    /* Additional Checks */
    "noUnusedLocals": true,
    "noUnusedParameters": true,
    "noImplicitReturns": false,
    "noFallthroughCasesInSwitch": false,

    /* Module Resolution Options */
    "baseUrl": ".",
    "paths": {
      /* Support absolute /imports/* with a leading '/' */
      "/*": ["*"],
      /* Pull in type declarations for Meteor packages from either zodern:types or @types/meteor packages */
      "meteor/*": ["node_modules/@types/meteor/*", ".meteor/local/types/packages.d.ts"],
      "meteor/mdg:validated-method": [
        "@types/meteor/mdg:validated-method",
        ".meteor/local/types/packages.d.ts"
      ],
      "meteor/hwillson:stub-collections": [
        "@types/meteor/hwillson:stub-collections",
        ".meteor/local/types/packages.d.ts"
      ],
      "meteor/johanbrook:publication-collector": [
        "@types/meteor/johanbrook:publication-collector",
        ".meteor/local/types/packages.d.ts"
      ],
      "meteor/tunguska:reactive-aggregate": [
        "@types/meteor/tunguska:reactive-aggregate",
        ".meteor/local/types/packages.d.ts"
      ],
      "meteor/percolate:migrations": [
        "@types/meteor/percolate:migrations",
        ".meteor/local/types/packages.d.ts"
      ],
      "meteor/cultofcoders:redis-oplog": [
        "@types/meteor/cultofcoders:redis-oplog",
        ".meteor/local/types/packages.d.ts"
      ],
      "*.ts": ["*"]
    },
    "moduleResolution": "node",
    "resolveJsonModule": true,
    "types": ["node", "mocha", "@emotion/react/types/css-prop", "meteor", "@types/meteor-roles"],
    "esModuleInterop": true,
    "preserveSymlinks": true,
    "experimentalDecorators": true,
    "emitDecoratorMetadata": true
  },
  "exclude": ["./node_modules/**", "./.meteor/**", "./packages/**"]
}

Will be appreciate for any hint ...

[robertsLando]

I'm not a fan of RxJS and I sincerly have no clue where the error could come from. The only thing that comes to my mind is that we use a custom TypedEventEmitter implementation that may be the cause of your issues:

https://github.com/mqttjs/MQTT.js/blob/main/src/lib/TypedEmitter.ts

I kindly ask you to do some tests your own and open a PR to fix this if you find a solution

[AlCalzone]

It looks like fromEvent infers the type of event handler argument in a way that does not consider the actual event name, but MqttClient has different handler signatures for each event.
You end up with the arguments for the packetsend event, which is (packet: Packet), when you wanted the arguments for the message event, which is (topic: string, payload: Buffer, packet: IPublishPacket).

I don't see a clean way to fix this, other than overriding the return type:

import { connect, IClientOptions, IPublishPacket, OnMessageCallback, Packet } from "mqtt";
import { fromEvent, type Observable } from "rxjs";
//...
const message$ = fromEvent(powerLinesMQTTClient, "message") as unknown as Observable<Parameters<OnMessageCallback>>;

[Vetnametz]

Appreciate for answers
This approach is better than @ts-expect-error ))