fix: Race condition when initializing the MqttClientConnection (#144)

Fixes https://github.com/bourquep/mysa2mqtt/issues/41

.all-contributorsrc

@@ -0,0 +1,22 @@
+{
+  "files": ["README.md"],
+  "imageSize": 100,
+  "commit": false,
+  "commitType": "docs",
+  "commitConvention": "angular",
+  "contributors": [
+    {
+      "login": "jagmandan",
+      "name": "jagmandan",
+      "avatar_url": "https://avatars.githubusercontent.com/u/227265405?v=4",
+      "profile": "https://github.com/jagmandan",
+      "contributions": ["code"]
+    }
+  ],
+  "contributorsPerLine": 7,
+  "skipCi": true,
+  "repoType": "github",
+  "repoHost": "https://github.com",
+  "projectName": "mysa-js-sdk",
+  "projectOwner": "bourquep"
+}

.github/workflows/codeql.yml

@@ -0,0 +1,100 @@
+# For most projects, this workflow file will not need changing; you simply need
+# to commit it to your repository.
+#
+# You may wish to alter this file to override the set of languages analyzed,
+# or to provide custom queries or build logic.
+#
+# ******** NOTE ********
+# We have attempted to detect the languages in your repository. Please check
+# the `language` matrix defined below to confirm you have the correct set of
+# supported CodeQL languages.
+#
+name: 'CodeQL Advanced'
+
+on:
+  push:
+    branches: ['main']
+  pull_request:
+    branches: ['main']
+  schedule:
+    - cron: '44 5 * * 5'
+
+jobs:
+  analyze:
+    name: Analyze (${{ matrix.language }})
+    # Runner size impacts CodeQL analysis time. To learn more, please see:
+    #   - https://gh.io/recommended-hardware-resources-for-running-codeql
+    #   - https://gh.io/supported-runners-and-hardware-resources
+    #   - https://gh.io/using-larger-runners (GitHub.com only)
+    # Consider using larger runners or machines with greater resources for possible analysis time improvements.
+    runs-on: ${{ (matrix.language == 'swift' && 'macos-latest') || 'ubuntu-latest' }}
+    permissions:
+      # required for all workflows
+      security-events: write
+
+      # required to fetch internal or private CodeQL packs
+      packages: read
+
+      # only required for workflows in private repositories
+      actions: read
+      contents: read
+
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - language: actions
+            build-mode: none
+          - language: javascript-typescript
+            build-mode: none
+        # CodeQL supports the following values keywords for 'language': 'actions', 'c-cpp', 'csharp', 'go', 'java-kotlin', 'javascript-typescript', 'python', 'ruby', 'rust', 'swift'
+        # Use `c-cpp` to analyze code written in C, C++ or both
+        # Use 'java-kotlin' to analyze code written in Java, Kotlin or both
+        # Use 'javascript-typescript' to analyze code written in JavaScript, TypeScript or both
+        # To learn more about changing the languages that are analyzed or customizing the build mode for your analysis,
+        # see https://docs.github.com/en/code-security/code-scanning/creating-an-advanced-setup-for-code-scanning/customizing-your-advanced-setup-for-code-scanning.
+        # If you are analyzing a compiled language, you can modify the 'build-mode' for that language to customize how
+        # your codebase is analyzed, see https://docs.github.com/en/code-security/code-scanning/creating-an-advanced-setup-for-code-scanning/codeql-code-scanning-for-compiled-languages
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      # Add any setup steps before running the `github/codeql-action/init` action.
+      # This includes steps like installing compilers or runtimes (`actions/setup-node`
+      # or others). This is typically only required for manual builds.
+      # - name: Setup runtime (example)
+      #   uses: actions/setup-example@v1
+
+      # Initializes the CodeQL tools for scanning.
+      - name: Initialize CodeQL
+        uses: github/codeql-action/init@v3
+        with:
+          languages: ${{ matrix.language }}
+          build-mode: ${{ matrix.build-mode }}
+          # If you wish to specify custom queries, you can do so here or in a config file.
+          # By default, queries listed here will override any specified in a config file.
+          # Prefix the list here with "+" to use these queries and those in the config file.
+
+          # For more details on CodeQL's query packs, refer to: https://docs.github.com/en/code-security/code-scanning/automatically-scanning-your-code-for-vulnerabilities-and-errors/configuring-code-scanning#using-queries-in-ql-packs
+          # queries: security-extended,security-and-quality
+
+      # If the analyze step fails for one of the languages you are analyzing with
+      # "We were unable to automatically build your code", modify the matrix above
+      # to set the build mode to "manual" for that language. Then modify this step
+      # to build your code.
+      # ℹ️ Command-line programs to run using the OS shell.
+      # 📚 See https://docs.github.com/en/actions/using-workflows/workflow-syntax-for-github-actions#jobsjob_idstepsrun
+      - if: matrix.build-mode == 'manual'
+        shell: bash
+        run: |
+          echo 'If you are using a "manual" build mode for one or more of the' \
+            'languages you are analyzing, replace this with the commands to build' \
+            'your code, for example:'
+          echo '  make bootstrap'
+          echo '  make release'
+          exit 1
+
+      - name: Perform CodeQL Analysis
+        uses: github/codeql-action/analyze@v3
+        with:
+          category: '/language:${{matrix.language}}'

README.md

@@ -1,5 +1,11 @@
 # Mysa Smart Thermostat JavaScript SDK
 
+<!-- ALL-CONTRIBUTORS-BADGE:START - Do not remove or modify this section -->
+
+[![All Contributors](https://img.shields.io/badge/all_contributors-1-orange.svg?style=flat-square)](#contributors-)
+
+<!-- ALL-CONTRIBUTORS-BADGE:END -->
+
 [![NPM Version](https://img.shields.io/npm/v/mysa-js-sdk)](https://www.npmjs.com/package/mysa-js-sdk)
 [![CodeQL](https://github.com/bourquep/mysa-js-sdk/actions/workflows/github-code-scanning/codeql/badge.svg)](https://github.com/bourquep/mysa-js-sdk/actions/workflows/github-code-scanning/codeql)
 [![CI: lint, build and release](https://github.com/bourquep/mysa-js-sdk/actions/workflows/ci.yml/badge.svg)](https://github.com/bourquep/mysa-js-sdk/actions/workflows/ci.yml)
@@ -229,3 +235,26 @@ For general questions and discussions, join our [Discussion Forum](https://githu
 This library would not be possible without the amazing work by [@dlenski](https://github.com/dlenski) in his
 [mysotherm](https://github.com/dlenski/mysotherm) repository. He's the one who reversed-engineered the Mysa MQTT
 protocol which is being used by this library.
+
+## Contributors ✨
+
+Thanks goes to these wonderful people ([emoji key](https://allcontributors.org/docs/en/emoji-key)):
+
+<!-- ALL-CONTRIBUTORS-LIST:START - Do not remove or modify this section -->
+<!-- prettier-ignore-start -->
+<!-- markdownlint-disable -->
+<table>
+  <tbody>
+    <tr>
+      <td align="center" valign="top" width="14.28%"><a href="https://github.com/jagmandan"><img src="https://avatars.githubusercontent.com/u/227265405?v=4?s=100" width="100px;" alt="jagmandan"/><br /><sub><b>jagmandan</b></sub></a><br /><a href="https://github.com/bourquep/mysa-js-sdk/commits?author=jagmandan" title="Code">💻</a></td>
+    </tr>
+  </tbody>
+</table>
+
+<!-- markdownlint-restore -->
+<!-- prettier-ignore-end -->
+
+<!-- ALL-CONTRIBUTORS-LIST:END -->
+
+This project follows the [all-contributors](https://github.com/all-contributors/all-contributors) specification.
+Contributions of any kind welcome!

example/main.ts

@@ -70,7 +70,7 @@ async function main() {
           `'${device.Name}' status changed: ${status.temperature}°C, ${status.humidity}%, ${watts ?? 'na'}W`
         );
       } catch (error) {
-        rootLogger.error(`Error processing status update for device '${status.deviceId}':`, error);
+        rootLogger.error(error, `Error processing status update for device '${status.deviceId}'`);
       }
     });
 
@@ -79,7 +79,7 @@ async function main() {
         const device = devices.DevicesObj[change.deviceId];
         rootLogger.info(`'${device.Name}' setpoint changed from ${change.previousSetPoint} to ${change.newSetPoint}`);
       } catch (error) {
-        rootLogger.error(`Error processing setpoint update for device '${change.deviceId}':`, error);
+        rootLogger.error(error, `Error processing setpoint update for device '${change.deviceId}'`);
       }
     });
 
@@ -88,17 +88,19 @@ async function main() {
         const device = devices.DevicesObj[change.deviceId];
         rootLogger.info(change, `'${device.Name}' state changed.`);
       } catch (error) {
-        rootLogger.error(`Error processing setpoint update for device '${change.deviceId}':`, error);
+        rootLogger.error(error, `Error processing state update for device '${change.deviceId}'`);
       }
     });
   }
 
-  for (const device of Object.entries(devices.DevicesObj)) {
-    const serial = await client.getDeviceSerialNumber(device[0]);
-    rootLogger.info(`Serial number for device '${device[0]}' (${device[1].Name}): ${serial}`);
+  await Promise.all(
+    Object.entries(devices.DevicesObj).map(async ([deviceId, device]) => {
+      const serial = await client.getDeviceSerialNumber(deviceId);
+      rootLogger.info(`Serial number for device '${deviceId}' (${device.Name}): ${serial}`);
 
-    await client.startRealtimeUpdates(device[0]);
-  }
+      await client.startRealtimeUpdates(deviceId);
+    })
+  );
 }
 
 main().catch((error) => {

package.json

@@ -50,34 +50,34 @@
     "brace-expansion": "^2.0.2"
   },
   "dependencies": {
-    "@aws-sdk/client-iot": "3.830.0",
-    "@aws-sdk/credential-providers": "3.830.0",
+    "@aws-sdk/client-iot": "3.901.0",
+    "@aws-sdk/credential-providers": "3.901.0",
     "amazon-cognito-identity-js": "6.3.15",
-    "aws-iot-device-sdk-v2": "1.21.5",
+    "aws-iot-device-sdk-v2": "1.22.0",
     "dayjs": "1.11.13",
     "lodash": "4.17.21"
   },
   "devDependencies": {
-    "@eslint/js": "9.29.0",
-    "@semantic-release/npm": "12.0.1",
-    "@types/lodash": "4.17.18",
-    "@types/node": "24.0.3",
-    "conventional-changelog-conventionalcommits": "9.0.0",
-    "dotenv": "16.5.0",
-    "eslint": "9.29.0",
-    "eslint-plugin-jsdoc": "51.0.3",
+    "@eslint/js": "9.34.0",
+    "@semantic-release/npm": "12.0.2",
+    "@types/lodash": "4.17.20",
+    "@types/node": "24.3.0",
+    "conventional-changelog-conventionalcommits": "9.1.0",
+    "dotenv": "17.2.1",
+    "eslint": "9.34.0",
+    "eslint-plugin-jsdoc": "54.1.1",
     "eslint-plugin-tsdoc": "0.4.0",
-    "pino": "9.7.0",
+    "pino": "9.13.0",
     "pino-pretty": "13.0.0",
-    "prettier": "3.5.3",
-    "prettier-plugin-jsdoc": "1.3.2",
-    "prettier-plugin-organize-imports": "4.1.0",
-    "semantic-release": "24.2.5",
+    "prettier": "3.6.2",
+    "prettier-plugin-jsdoc": "1.3.3",
+    "prettier-plugin-organize-imports": "4.2.0",
+    "semantic-release": "24.2.7",
     "tsup": "8.5.0",
     "tsx": "4.20.3",
-    "typedoc": "0.28.5",
+    "typedoc": "0.28.13",
     "typedoc-material-theme": "1.4.0",
-    "typescript": "5.8.3",
-    "typescript-eslint": "8.34.0"
+    "typescript": "5.9.2",
+    "typescript-eslint": "8.41.0"
   }
 }

src/api/MysaApiClient.ts

@@ -75,8 +75,8 @@ export class MysaApiClient {
   /** The fetcher function used by the client. */
   private _fetcher: typeof fetch;
 
-  /** The MQTT connection used for real-time updates. */
-  private _mqttConnection?: mqtt.MqttClientConnection;
+  /** A promise that resolves to the MQTT connection used for real-time updates. */
+  private _mqttConnectionPromise?: Promise<mqtt.MqttClientConnection>;
 
   /** The device IDs that are currently being updated in real-time, mapped to their respective timeouts. */
   private _realtimeDeviceIds: Map<string, NodeJS.Timeout> = new Map();
@@ -374,7 +374,7 @@ export class MysaApiClient {
     this._logger.debug(`Sending request to set device state for '${deviceId}'...`);
     const payload = serializeMqttPayload<ChangeDeviceState>({
       msg: InMessageType.CHANGE_DEVICE_STATE,
-      id: now.unix(),
+      id: now.valueOf(),
       time: now.unix(),
       ver: '1.0',
       src: {
@@ -543,11 +543,24 @@ export class MysaApiClient {
    * @returns A promise that resolves to an active MQTT connection.
    * @throws {@link Error} When connection establishment fails.
    */
-  private async getMqttConnection(): Promise<mqtt.MqttClientConnection> {
-    if (this._mqttConnection) {
-      return this._mqttConnection;
+  private getMqttConnection(): Promise<mqtt.MqttClientConnection> {
+    if (!this._mqttConnectionPromise) {
+      this._mqttConnectionPromise = this.createMqttConnection().catch((err) => {
+        this._mqttConnectionPromise = undefined;
+        throw err;
+      });
     }
 
+    return this._mqttConnectionPromise;
+  }
+
+  /**
+   * Creates a new MQTT connection using AWS IoT WebSocket connections with Cognito credentials.
+   *
+   * @returns A promise that resolves to an active MQTT connection.
+   * @throws {@link Error} When connection establishment fails.
+   */
+  private async createMqttConnection(): Promise<mqtt.MqttClientConnection> {
     const session = await this.getFreshSession();
     const credentialsProvider = fromCognitoIdentityPool({
       clientConfig: {
@@ -572,16 +585,25 @@ export class MysaApiClient {
 
     const config = builder.build();
     const client = new mqtt.MqttClient();
-    this._mqttConnection = client.new_connection(config);
+    const connection = client.new_connection(config);
 
-    this._mqttConnection.on('closed', () => {
+    connection.on('connect', () => this._logger.debug('MQTT connect'));
+    connection.on('connection_success', () => this._logger.debug('MQTT connection_success'));
+    connection.on('connection_failure', (e) => this._logger.error('MQTT connection_failure', e));
+    connection.on('interrupt', (e) => this._logger.warn('MQTT interrupt', e));
+    connection.on('resume', (returnCode, sessionPresent) =>
+      this._logger.info(`MQTT resume returnCode=${returnCode} sessionPresent=${sessionPresent}`)
+    );
+    connection.on('error', (e) => this._logger.error('MQTT error', e));
+
+    connection.on('closed', () => {
       this._logger.info('MQTT connection closed');
-      this._mqttConnection = undefined;
+      this._mqttConnectionPromise = undefined;
     });
 
-    await this._mqttConnection.connect();
+    await connection.connect();
 
-    return this._mqttConnection;
+    return connection;
   }
 
   /**

src/lib/PayloadParser.ts

@@ -32,7 +32,7 @@ export function parseMqttPayload(payload: ArrayBuffer): OutPayload {
  * @param payload - The typed payload object to serialize
  * @returns The serialized payload as ArrayBuffer ready for MQTT transmission
  */
-export function serializeMqttPayload<T extends InPayload>(payload: T): ArrayBuffer {
+export function serializeMqttPayload<T extends InPayload>(payload: T): Uint8Array<ArrayBuffer> {
   const jsonString = JSON.stringify(payload);
   const encoder = new TextEncoder();
   return encoder.encode(jsonString);