bridge: the discovery answer is published through send()

The answer on <root>/discover_r called the publisher directly; a publisher
that threw or rejected was only caught by the handler meant for listeners
of "discover". It now takes the path of every device message: send()
resolves, and a failure goes to the "error" event when somebody listens.
A test covers a publish that fails, throws and rejects, with and without
an "error" listener: receive() resolves each time and "discover" is not
emitted.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

dist/cjs/Alex2Node.js

@@ -174,12 +174,11 @@ class Alex2MQTT extends events_1.EventEmitter {
         this.log("Discovery request received, getting device json...");
         const deviceArray = this.describeDevices();
         this.lastDiscoveryAt = new Date().toISOString();
-        const result = await this.publisher.publish(topics.discoverReply(this.rootTopic), deviceArray);
-        if (!result.ok) {
-            this.fail(result.error);
+        // send(), as for every message of a device: a publisher that throws or rejects is reported like one that fails
+        const topic = await (0, transport_js_1.send)(this.publisher, topics.discoverReply(this.rootTopic), () => deviceArray, (err) => this.fail(err));
+        if (topic === "")
             return;
-        }
-        this.log(`Discovery payloads published to ${result.topic}`, deviceArray);
+        this.log(`Discovery payloads published to ${topic}`, deviceArray);
         this.emit("discover", deviceArray.length);
     }
     // The endpoint objects of a discovery answer. What check() says about a device is logged, each line once. A device

dist/esm/Alex2Node.js

@@ -7,7 +7,7 @@ import { DisplayCategory } from "./compat/enums.js";
 import { DeclarationError } from "./registry/types.js";
 import { Dispatcher } from "./dispatcher.js";
 import * as topics from "./topics.js";
-import { asError, MqttPublisher } from "./transport.js";
+import { asError, MqttPublisher, send } from "./transport.js";
 export const DEFAULT_HOST = "mqtt://Alex2MQTT.stormysdream.club:1883";
 // What addDevice() copies from the definition to the device
 const DESCRIBED = [
@@ -135,12 +135,11 @@ class Alex2MQTT extends EventEmitter {
         this.log("Discovery request received, getting device json...");
         const deviceArray = this.describeDevices();
         this.lastDiscoveryAt = new Date().toISOString();
-        const result = await this.publisher.publish(topics.discoverReply(this.rootTopic), deviceArray);
-        if (!result.ok) {
-            this.fail(result.error);
+        // send(), as for every message of a device: a publisher that throws or rejects is reported like one that fails
+        const topic = await send(this.publisher, topics.discoverReply(this.rootTopic), () => deviceArray, (err) => this.fail(err));
+        if (topic === "")
             return;
-        }
-        this.log(`Discovery payloads published to ${result.topic}`, deviceArray);
+        this.log(`Discovery payloads published to ${topic}`, deviceArray);
         this.emit("discover", deviceArray.length);
     }
     // The endpoint objects of a discovery answer. What check() says about a device is logged, each line once. A device

src/Alex2Node.ts

@@ -10,7 +10,7 @@ import { DeclarationError } from "./registry/types.js";
 import { Dispatcher } from "./dispatcher.js";
 import type { Timers } from "./dispatcher.js";
 import * as topics from "./topics.js";
-import { asError, MqttPublisher } from "./transport.js";
+import { asError, MqttPublisher, send } from "./transport.js";
 import type { Publisher } from "./transport.js";
 
 /** Optional settings for the bridge (1.5.1). Everything has the 1.4.0 behaviour as its default. */
@@ -176,9 +176,10 @@ class Alex2MQTT extends EventEmitter {
     this.log("Discovery request received, getting device json...");
     const deviceArray = this.describeDevices();
     this.lastDiscoveryAt = new Date().toISOString();
-    const result = await this.publisher.publish(topics.discoverReply(this.rootTopic), deviceArray);
-    if (!result.ok) { this.fail(result.error); return; }
-    this.log(`Discovery payloads published to ${result.topic}`, deviceArray);
+    // send(), as for every message of a device: a publisher that throws or rejects is reported like one that fails
+    const topic = await send(this.publisher, topics.discoverReply(this.rootTopic), () => deviceArray, (err) => this.fail(err));
+    if (topic === "") return;
+    this.log(`Discovery payloads published to ${topic}`, deviceArray);
     this.emit("discover", deviceArray.length);
   }
 

test/transport.test.js

@@ -63,3 +63,29 @@ test("unregisterDevice and clearDevices take the publisher from the device", ()
   bridge.clearDevices();
   assert.equal(plug.publisher, null);
 });
+
+test("a discovery answer that cannot be published: receive() resolves and the \"error\" event says why", async () => {
+  const gone = new MemoryPublisher();
+  gone.failWith = new Error("broker gone");
+  const publishers = {
+    "fails": gone,
+    "throws": { publish() { throw new Error("publisher threw"); } },
+    "rejects": { publish: () => Promise.reject(new Error("publisher rejected")) },
+  };
+  const expected = { "fails": "broker gone", "throws": "publisher threw", "rejects": "publisher rejected" };
+  for (const [how, publisher] of Object.entries(publishers)) {
+    const bridge = new Alex2MQTT("user", "password", "root", false, { publisher });
+    bridge.addDevice({ endpointId: "lamp-1", name: "Lamp", categories: ["LIGHT"] }).add(PowerController);
+    const errors = [];
+    const discovered = [];
+    bridge.on("error", (err) => errors.push(err.message));
+    bridge.on("discover", (n) => discovered.push(n));
+    await bridge.receive("root/discover", "");
+    assert.deepEqual(errors, [expected[how]], how);
+    assert.deepEqual(discovered, [], how);
+
+    // Without a listener the error has nowhere to go, and still nothing is thrown or rejected
+    bridge.removeAllListeners("error");
+    await bridge.receive("root/discover", "");
+  }
+});