Why send two 'Binary Message' when connect, can I configure that just send one 'Binary Message'?

[ghzjtian]

Related versions

MQTT.js library version: 4.3.7
Frontend Vue version: 3.2.47
browser version: Chrome Version 124.0.6367.62
MQTT broker version: 3.1.1

Issue

I searched online for a long time but couldn't find the answer.

I use the library "mqtt": "^4.3.7" to develop my Vue3 APP, found that when connect to the MQTT broker, the library will send double 'Binary Message' when connect, and that self host MQTT broker will not allow to connect because the format of the connect is not correct. So the connect is fail.

But if I implement the MQTT connect logic by myself( Just send one 'Binary Message' when connect), and send data use websocket, it was success. Below is my test code:

<template>
  <div>
    test
  </div>
</template>

<script setup>

function hexToUint8Array(hexString) {
  const bytes = [];
  for (let i = 0; i < hexString.length; i += 2) {
    bytes.push(parseInt(hexString.substr(i, 2), 16));
  }
  return new Uint8Array(bytes);
}

const hexData = "1076xxxxx"; // xxx contain sensitive data.

const socket = new WebSocket('wss://xxx.com:11801/mqtt');
socket.onmessage = (event) => {
      // Handle incoming messages
      console.log(`event: ${event}`);
    };
    
    socket.onopen = () => {
      // WebSocket connection established
      console.log(`onopen`);

      const uint8Array = hexToUint8Array(hexData);
      socket.send(uint8Array);
    };
    socket.onclose = (event) => {
      // WebSocket connection established
      console.log(`onclose: ${event}`);
    };
    
    socket.onerror = (error) => {
      // Handle WebSocket error
      console.log(`onerror: ${error}`);
    };

    
</script>

<style lang="scss" scoped>

</style>

But it is strange that if i use the broker of emqx, it can successfully connect although it send two 'Binary Message ' when connect. Below are the code and some screenshot.

<script setup lang="ts">

// vue 3 + vite use MQTT.js refer to https://github.com/mqttjs/MQTT.js/issues/1269
import * as mqtt from "mqtt/dist/mqtt.min";
import { reactive, ref } from "vue";


// https://github.com/mqttjs/MQTT.js#qos
const qosList = [0, 1, 2];

const connection = reactive({
  // ws or wss
  protocol: "wss",
  host: "broker.emqx.io",
  // ws -> 8083; wss -> 8084
  port: 8084,
  clientId: "emqx_vue3_" + Math.random().toString(16).substring(2, 8),
  /**
   * By default, EMQX allows clients to connect without authentication.
   * https://docs.emqx.com/en/enterprise/v4.4/advanced/auth.html#anonymous-login
   */
  username: "emqx_test",
  password: "emqx_test",

  clean: true,
  connectTimeout: 4 * 1000, // ms
  reconnectPeriod: 1 * 1000, // ms
  // for more options and details, please refer to https://github.com/mqttjs/MQTT.js#mqttclientstreambuilder-options
  protocolId: 'MQTT',
  protocolVersion: 4,
  retain: 0,
  qos: 0,
  rejectUnauthorized: false,
});


// topic & QoS for MQTT subscribing
const subscription = ref({
  topic: "topic/mqttx",
  qos: 0 as mqtt.QoS,
});

// topic, QoS & payload for publishing message
const publish = ref({
  topic: "topic/browser",
  qos: 0 as mqtt.QoS,
  payload: '{ "msg": "Hello, I am browser." }',
});

let client = ref({
  connected: false,
} as mqtt.MqttClient);
const receivedMessages = ref("");
const subscribedSuccess = ref(false);
const btnLoadingType = ref("");
const retryTimes = ref(0);

const initData = () => {
  client.value = {
    connected: false,
  } as mqtt.MqttClient;
  retryTimes.value = 0;
  btnLoadingType.value = "";
  subscribedSuccess.value = false;
};

const handleOnReConnect = () => {
  retryTimes.value += 1;
  if (retryTimes.value > 1) { // 5
    try {
      client.value.end();
      initData();
      console.log("connection maxReconnectTimes limit, stop retry");
    } catch (error) {
      console.log("handleOnReConnect catch error:", error);
    }
  }
};

// create MQTT connection
const createConnection = () => {
  try {
    btnLoadingType.value = "connect";
    const { protocol, host, port, ...options } = connection;
    const connectUrl = `${protocol}://${host}:${port}/mqtt`;

    client.value = mqtt.connect(connectUrl, options);
    if (client.value.on) {
      // https://github.com/mqttjs/MQTT.js#event-connect
      client.value.on("connect", () => {
        btnLoadingType.value = "";
        console.log("connection successful");
      });

      // https://github.com/mqttjs/MQTT.js#event-reconnect
      client.value.on("reconnect", handleOnReConnect);

      // https://github.com/mqttjs/MQTT.js#event-error
      client.value.on("error", (error) => {
        console.log("connection error:", error);
      });

      // https://github.com/mqttjs/MQTT.js#event-message
      client.value.on("message", (topic: string, message) => {
        receivedMessages.value = receivedMessages.value.concat(
          message.toString()
        );
        console.log(`received message: ${message} from topic: ${topic}`);
      });
    }
  } catch (error) {
    btnLoadingType.value = "";
    console.log("mqtt.connect error:", error);
  }
};

// subscribe topic
// https://github.com/mqttjs/MQTT.js#mqttclientsubscribetopictopic-arraytopic-object-options-callback
const doSubscribe = () => {
  btnLoadingType.value = "subscribe";
  const { topic, qos } = subscription.value;
  client.value.subscribe(
    topic,
    { qos },
    (error: Error, granted: mqtt.ISubscriptionGrant[]) => {
      btnLoadingType.value = "";
      if (error) {
        console.log("subscribe error:", error);
        return;
      }
      subscribedSuccess.value = true;
      console.log("subscribe successfully:", granted);
    }
  );
};

// unsubscribe topic
// https://github.com/mqttjs/MQTT.js#mqttclientunsubscribetopictopic-array-options-callback
const doUnSubscribe = () => {
  btnLoadingType.value = "unsubscribe";
  const { topic, qos } = subscription.value;
  client.value.unsubscribe(topic, { qos }, (error) => {
    btnLoadingType.value = "";
    subscribedSuccess.value = false;
    if (error) {
      console.log("unsubscribe error:", error);
      return;
    }
    console.log(`unsubscribed topic: ${topic}`);
  });
};

// publish message
// https://github.com/mqttjs/MQTT.js#mqttclientpublishtopic-message-options-callback
const doPublish = () => {
  btnLoadingType.value = "publish";
  const { topic, qos, payload } = publish.value;
  client.value.publish(topic, payload, { qos }, (error) => {
    btnLoadingType.value = "";
    if (error) {
      console.log("publish error:", error);
      return;
    }
    console.log(`published message: ${payload}`);
  });
};

// disconnect
// https://github.com/mqttjs/MQTT.js#mqttclientendforce-options-callback
const destroyConnection = () => {
  if (client.value.connected) {
    btnLoadingType.value = "disconnect";
    try {
      client.value.end(false, () => {
        initData();
        console.log("disconnected successfully");
      });
    } catch (error) {
      btnLoadingType.value = "";
      console.log("disconnect error:", error);
    }
  }
};

const handleProtocolChange = (value: string) => {
  connection.port = value === "wss" ? 8084 : 8083;
};
</script>

<template>
  <div class="mqtt-demo">
    <el-card>
      <h1>Configuration</h1>
      <el-form label-position="top" :model="connection">
        <el-row :gutter="20">
          <el-col :span="8">
            <el-form-item prop="protocol" label="Protocol">
              <el-select
                v-model="connection.protocol"
                @change="handleProtocolChange"
              >
                <el-option label="ws://" value="ws"></el-option>
                <el-option label="wss://" value="wss"></el-option>
              </el-select>
            </el-form-item>
          </el-col>
          <el-col :span="8">
            <el-form-item prop="host" label="Host">
              <el-input v-model="connection.host"></el-input>
            </el-form-item>
          </el-col>
          <el-col :span="8">
            <el-form-item prop="port" label="Port">
              <el-input
                v-model.number="connection.port"
                type="number"
                placeholder="8083/8084"
              ></el-input>
            </el-form-item>
          </el-col>
          <el-col :span="8">
            <el-form-item prop="clientId" label="Client ID">
              <el-input v-model="connection.clientId"> </el-input>
            </el-form-item>
          </el-col>
          <el-col :span="8">
            <el-form-item prop="username" label="Username">
              <el-input v-model="connection.username"></el-input>
            </el-form-item>
          </el-col>
          <el-col :span="8">
            <el-form-item prop="password" label="Password">
              <el-input v-model="connection.password"></el-input>
            </el-form-item>
          </el-col>

          <el-col :span="24">
            <el-button
              type="primary"
              :disabled="client.connected"
              @click="createConnection"
              :loading="btnLoadingType === 'connect'"
            >
              {{ client.connected ? "Connected" : "Connect" }}
            </el-button>

            <el-button
              v-if="client.connected"
              type="danger"
              @click="destroyConnection"
              :loading="btnLoadingType === 'disconnect'"
            >
              Disconnect
            </el-button>
          </el-col>
        </el-row>
      </el-form>
    </el-card>
    <el-card>
      <h1>Subscribe</h1>
      <el-form label-position="top" :model="subscription">
        <el-row :gutter="20">
          <el-col :span="8">
            <el-form-item prop="topic" label="Topic">
              <el-input
                v-model="subscription.topic"
                :disabled="subscribedSuccess"
              ></el-input>
            </el-form-item>
          </el-col>
          <el-col :span="8">
            <el-form-item prop="qos" label="QoS">
              <el-select
                v-model="subscription.qos"
                :disabled="subscribedSuccess"
              >
                <el-option
                  v-for="qos in qosList"
                  :key="qos"
                  :label="qos"
                  :value="qos"
                ></el-option>
              </el-select>
            </el-form-item>
          </el-col>
          <el-col :span="8">
            <el-button
              type="primary"
              class="sub-btn"
              :loading="btnLoadingType === 'subscribe'"
              :disabled="!client.connected || subscribedSuccess"
              @click="doSubscribe"
            >
              {{ subscribedSuccess ? "Subscribed" : "Subscribe" }}
            </el-button>
            <el-button
              v-if="subscribedSuccess"
              type="primary"
              class="sub-btn"
              :loading="btnLoadingType === 'unsubscribe'"
              :disabled="!client.connected"
              @click="doUnSubscribe"
            >
              Unsubscribe
            </el-button>
          </el-col>
        </el-row>
      </el-form>
    </el-card>
    <el-card>
      <h1>Publish</h1>
      <el-form label-position="top" :model="publish">
        <el-row :gutter="20">
          <el-col :span="8">
            <el-form-item prop="topic" label="Topic">
              <el-input v-model="publish.topic"></el-input>
            </el-form-item>
          </el-col>
          <el-col :span="8">
            <el-form-item prop="payload" label="Payload">
              <el-input v-model="publish.payload"></el-input>
            </el-form-item>
          </el-col>
          <el-col :span="8">
            <el-form-item prop="qos" label="QoS">
              <el-select v-model="publish.qos">
                <el-option
                  v-for="qos in qosList"
                  :key="qos"
                  :label="qos"
                  :value="qos"
                ></el-option>
              </el-select>
            </el-form-item>
          </el-col>
        </el-row>
      </el-form>
      <el-col :span="24" class="text-right">
        <el-button
          type="primary"
          :loading="btnLoadingType === 'publish'"
          :disabled="!client.connected"
          @click="doPublish"
        >
          Publish
        </el-button>
      </el-col>
    </el-card>
    <el-card>
      <h1>Receive</h1>
      <el-col :span="24">
        <el-input
          type="textarea"
          :rows="3"
          v-model="receivedMessages"
          readonly
        ></el-input>
      </el-col>
    </el-card>
  </div>
</template>

<style>
.mqtt-demo {
  max-width: 1200px;
  margin: 32px auto 0 auto;
}

h1 {
  font-size: 16px;
  margin-top: 0;
}

.el-card {
  margin-bottom: 32px;
}
.el-card__body {
  padding: 24px;
}

.el-select {
  width: 100%;
}

.text-right {
  text-align: right;
}

.sub-btn {
  margin-top: 30px;
}
</style>

Specification

According to the MQTT specification( http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html#_Toc442180842 ), the fixed header should be contain with 'MQTT Control Packet type (1)' when connect.

Question

So, my question is , can I configure that just send one 'Binary Message' when connect?

[robertsLando]

@ghzjtian Firstly you are using a very old mqttjs version, latest now is 5.5.3. Try to use that first.

Writes to stream are handled by mqtt-packet, sincerly I have no clue why the connect is splitted in two binary messages. Some references in the code:

1. Create the connect packet

MQTT.js/src/lib/client.ts

Line 833 in e991fa5

this._writePacket(connectPacket)

2. Write to stream

MQTT.js/src/lib/client.ts

Line 833 in e991fa5

this._writePacket(connectPacket)

3. Mqtt Packet connecct: https://github.com/mqttjs/mqtt-packet/blob/master/writeToStream.js#L217

@mcollina Any clue why this could happen?

[mcollina]

Performance when using TLS or TCP. Otherwise we would need to concatenate two buffers into one.

MQTT has its own packets so it treats websockets just as a binary stream.

This could potentially be solved in the underlying layer.

[robertsLando]

This could potentially be solved in the underlying layer.

You mean in mqtt-packet or in websoket (browser) handler? How?

MQTT.js/src/lib/connect/ws.ts

Lines 124 to 156 in e991fa5

	function createBrowserWebSocket(client: MqttClient, opts: IClientOptions) {
	const websocketSubProtocol =
	opts.protocolId === 'MQIsdp' && opts.protocolVersion === 3
	? 'mqttv3.1'
	: 'mqtt'
	const url = buildUrl(opts, client)
	let socket: WebSocket
	if (opts.createWebsocket) {
	socket = opts.createWebsocket(url, [websocketSubProtocol], opts)
	} else {
	socket = new WebSocket(url, [websocketSubProtocol])
	}
	socket.binaryType = 'arraybuffer'
	return socket
	}
	const streamBuilder: StreamBuilder = (client, opts) => {
	debug('streamBuilder')
	const options = setDefaultOpts(opts)
	const url = buildUrl(options, client)
	const socket = createWebSocket(client, url, options)
	const webSocketStream = Ws.createWebSocketStream(
	socket,
	options.wsOptions as DuplexOptions,
	)
	webSocketStream['url'] = url
	socket.on('close', () => {
	webSocketStream.destroy()
	})
	return webSocketStream
	}

[mcollina]

I mean it would need to handled inside ws.

[ghzjtian]

@robertsLando , thanks you, i upload the mqtt.js version to ^5.5.3, it just send one 'Binary Message' and connect success.