Add json write support

README.md

@@ -58,9 +58,9 @@ device support. For now, the supported interfaces are the following:
      [these](https://github.com/eclipse-paho/paho.mqtt.cpp?tab=readme-ov-file#build-the-paho-c-and-paho-c-libraries-together)
      instructions.
 
-  > Note: AutoparamDriver explicitly requires EPICS >= 7.0, see 
-  > [reference](https://github.com/Cosylab/autoparamDriver/blob/main/autoparamDriverSup/src/Makefile#L15). For this reason,
-  > this module requires EPICS 7.0 or later to be built and used.
+> Note: AutoparamDriver explicitly requires EPICS >= 7.0, see
+> [reference](https://github.com/Cosylab/autoparamDriver/blob/main/autoparamDriverSup/src/Makefile#L15). For this
+> reason, this module requires EPICS 7.0 or later to be built and used.
 
 2. Clone this repository:
 
@@ -117,9 +117,12 @@ Where:
 - `<FORMAT>` is the format of the payload: `FLAT` or `JSON`.
 - `<TYPE>` is the general type of the expected value [`INT|FLOAT|DIGITAL|STRING|INTARRAY|FLOATARRAY`].
 - `<TOPIC>` is the MQTT topic to which the record will be subscribed/published.
-- `<FIELD>` is JSON pointer to extract value from a JSON payload (e.g. `/sensor/temperature`). If empty, JSON root is used.
+- `<FIELD>` is JSON pointer to extract value from a JSON payload (e.g. `/sensor/temperature`). If empty, JSON root is
+  used.
 
-> **Note on JSON write support:** Writing to JSON-formatted topics is currently **not supported**. At the moment the driver has no way of knowing the JSON structure expected by the broker ahead of time for write records. For this reason, only `FLAT` format can be used for output records.
+> **Note on JSON write:** JSON write currently publishes the record value as a plain JSON scalar/array. Composing JSON
+> objects within the driver is not yet supported (see [#14](https://github.com/epics-modules/mqtt/issues/14)). To write
+> a structured JSON payload, compose it at the application level using a `FLAT:STRING` record and a `scalcout` record.
 
 **Important: Due to the pub/sub nature of MQTT, ALL input records are expected to be `I/O Intr`.**
 
@@ -166,7 +169,6 @@ Example:
 
 Below are the supported interfaces and their implementation status.
 
-
 | Message type        | Asyn Parameter Type                    | `FORMAT:TYPE` string to use | Direction    | Status    |
 | ------------------- | -------------------------------------- | --------------------------- | ------------ | --------- |
 | Integer             | asynInt32                              | `FLAT:INT`                  | Read / Write | Supported |
@@ -175,22 +177,23 @@ Below are the supported interfaces and their implementation status.
 | Strings             | asynOctetRead/asynOctetWrite           | `FLAT:STRING`               | Read / Write | Supported |
 | Integer Array       | asynInt32ArrayIn/asynInt32ArrayOut     | `FLAT:INTARRAY`             | Read / Write | Supported |
 | Float Array         | asynFloat64ArrayIn/asynFloat64ArrayOut | `FLAT:FLOATARRAY`           | Read / Write | Supported |
-| Integer             | asynInt32                              | `JSON:INT`                  | Read only    | Supported |
-| Float               | asynFloat64                            | `JSON:FLOAT`                | Read only    | Supported |
-| Bit masked          | asynUInt32Digital                      | `JSON:DIGITAL`              | Read only    | Supported |
-| String              | asynOctetRead                          | `JSON:STRING`               | Read only    | Supported |
+| Integer             | asynInt32                              | `JSON:INT`                  | Read / Write | Supported |
+| Float               | asynFloat64                            | `JSON:FLOAT`                | Read / Write | Supported |
+| Bit masked          | asynUInt32Digital                      | `JSON:DIGITAL`              | Read / Write | Supported |
+| String              | asynOctetRead/asynOctetWrite           | `JSON:STRING`               | Read / Write | Supported |
+| Integer Array       | asynInt32ArrayIn/asynInt32ArrayOut     | `JSON:INTARRAY`             | Read / Write | Supported |
+| Float Array         | asynFloat64ArrayIn/asynFloat64ArrayOut | `JSON:FLOATARRAY`           | Read / Write | Supported |
 
 ## Licensing Terms
 
 Copyright (C) 2026 André Favoto
 
-This program is free software: you can redistribute it and/or modify it under the terms of the GNU
-General Public License as published by the Free Software Foundation, either version 3 of the
-License, or (at your option) any later version.
+This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public
+License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later
+version.
 
-This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without
-even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
-General Public License for more details.
+This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied
+warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
 
-You should have received a copy of the GNU General Public License along with this program. If not,
-see <https://www.gnu.org/licenses/>.
+You should have received a copy of the GNU General Public License along with this program. If not, see
+<https://www.gnu.org/licenses/>.

mqttSup/src/drvMqtt.cpp

@@ -258,7 +258,13 @@ void MqttDriver::onMessageCb(Autoparam::Driver* driver, const std::string& topic
     if (addr.format == MqttTopicAddr::JSON) {
       try {
         json root = json::parse(payload);
-        val = to_string(root.at(json::json_pointer(addr.jsonField)));
+        auto jsonVal = root.at(json::json_pointer(addr.jsonField));
+        val = jsonVal.is_string() ? jsonVal.get<std::string>() : to_string(jsonVal);
+      }
+      catch (const json::out_of_range& e) {
+        // there can be multiple records pointing to same topic
+        // topicName is not enough to differentiate (json pointer may differ)
+        continue;
       }
       catch (const std::exception& e) {
         asynPrint(pself->pasynUserSelf, ASYN_TRACE_ERROR,
@@ -557,8 +563,10 @@ WriteResult MqttDriver::integerWrite(DeviceVariable& deviceVar, epicsInt32 value
       status = asynSuccess;
     }
     else if (addr.format == MqttTopicAddr::TopicFormat::JSON) {
-      // TODO: implement JSON support for integer values
-      throw std::logic_error("JSON support not implemented");
+      // TODO: Compose JSON from driver instead of publishing a plain JSON scalar.
+      // See https://github.com/epics-modules/mqtt/issues/14
+      driver->mqttClient.publish(addr.topicName, to_string(value));
+      status = asynSuccess;
     }
   }
   catch (const std::exception& exc) {
@@ -599,8 +607,10 @@ WriteResult MqttDriver::digitalWrite(DeviceVariable& deviceVar, epicsUInt32 cons
       status = asynSuccess;
     }
     else if (addr.format == MqttTopicAddr::TopicFormat::JSON) {
-      // TODO: implement JSON support for digital values
-      throw std::logic_error("JSON support not implemented");
+      // TODO: Compose JSON from driver instead of publishing a plain JSON scalar.
+      // See https://github.com/epics-modules/mqtt/issues/14
+      driver->mqttClient.publish(addr.topicName, to_string(value));
+      status = asynSuccess;
     }
   }
   catch (const std::exception& exc) {
@@ -626,8 +636,10 @@ WriteResult MqttDriver::floatWrite(DeviceVariable& deviceVar, epicsFloat64 value
       status = asynSuccess;
     }
     else if (addr.format == MqttTopicAddr::TopicFormat::JSON) {
-      // TODO: implement JSON support for float values
-      throw std::logic_error("JSON support not implemented");
+      // TODO: Compose JSON from driver instead of publishing a plain JSON scalar.
+      // See https://github.com/epics-modules/mqtt/issues/14
+      driver->mqttClient.publish(addr.topicName, to_string(value));
+      status = asynSuccess;
     }
   }
   catch (const std::exception& exc) {
@@ -649,10 +661,9 @@ WriteResult MqttDriver::arrayWrite(DeviceVariable& deviceVar, Array<epicsDataTyp
   std::string topicName = addr.topicName;
   MqttDriver* driver = static_cast<MqttTopicVariable&>(deviceVar).driver;
   try {
+    const epicsDataType* arrayData = reinterpret_cast<const epicsDataType*>(value.data());
+    size_t count = value.size();
     if (addr.format == MqttTopicAddr::TopicFormat::FLAT) {
-      const epicsDataType* arrayData = reinterpret_cast<const epicsDataType*>(value.data());
-      size_t count = value.size();
-
       std::ostringstream oss;
       for (size_t i = 0; i < count; ++i) {
         if (i > 0) oss << ",";
@@ -662,8 +673,14 @@ WriteResult MqttDriver::arrayWrite(DeviceVariable& deviceVar, Array<epicsDataTyp
       status = asynSuccess;
     }
     else if (addr.format == MqttTopicAddr::TopicFormat::JSON) {
-      // TODO: implement JSON support for array values
-      throw std::logic_error("JSON support not implemented");
+      // TODO: Compose JSON from driver instead of publishing a plain JSON scalar.
+      // See https://github.com/epics-modules/mqtt/issues/14
+      json data = json::array();
+      for (size_t i = 0; i < count; ++i) {
+        data.push_back(arrayData[i]);
+      }
+      driver->mqttClient.publish(addr.topicName, data.dump());
+      status = asynSuccess;
     }
   }
   catch (const std::exception& exc) {
@@ -683,17 +700,18 @@ WriteResult MqttDriver::stringWrite(DeviceVariable& deviceVar, Octet const& valu
   MqttTopicAddr const& addr = static_cast<MqttTopicAddr const&>(deviceVar.address());
   std::string topicName = addr.topicName;
   MqttDriver* driver = static_cast<MqttTopicVariable&>(deviceVar).driver;
+  std::vector<char> stringData(value.maxSize());
+  value.writeTo(stringData.data(), stringData.size());
   try {
     if (addr.format == MqttTopicAddr::TopicFormat::FLAT) {
-      std::vector<char> stringData(value.maxSize());
-      if (value.writeTo(stringData.data(), stringData.size())) {
-        driver->mqttClient.publish(addr.topicName, stringData.data());
-        status = asynSuccess;
-      }
-    }
-    else if (addr.format == MqttTopicAddr::TopicFormat::JSON) {
-      // TODO: implement JSON support for string values
-      throw std::logic_error("JSON support not implemented");
+      driver->mqttClient.publish(addr.topicName, stringData.data());
+      status = asynSuccess;
+    } else if (addr.format == MqttTopicAddr::TopicFormat::JSON) {
+      // TODO: Compose JSON from driver instead of publishing a plain JSON scalar.
+      // See https://github.com/epics-modules/mqtt/issues/14
+      json j = std::string(stringData.data());
+      driver->mqttClient.publish(addr.topicName, j.dump());
+      status = asynSuccess;
     }
   }
   catch (const std::exception& exc) {

testApp/Db/mqttTest.db

@@ -70,3 +70,102 @@ record(aao, "$(P)$(R)FloatArrayOutput") {
 	field(NELM, "16")
 	field(OUT, "@asyn($(PORT)) FLAT:FLOATARRAY $(TOPIC_ROOT)/floatarray")
 }
+
+record(ai, "$(P)$(R)JsonInt32Input") {
+	field(DESC, "CI Int32 Input")
+	field(DTYP, "asynInt32")
+	field(SCAN, "I/O Intr")
+	field(INP, "@asyn($(PORT)) JSON:INT $(TOPIC_ROOT)/json/int")
+}
+
+record(ao, "$(P)$(R)JsonInt32Output") {
+	field(DESC, "CI Int32 Output")
+	field(DTYP, "asynInt32")
+	field(OUT, "@asyn($(PORT)) JSON:INT $(TOPIC_ROOT)/json/int")
+}
+
+record(ai, "$(P)$(R)JsonFloat64Input") {
+	field(DESC, "CI Float64 Input")
+	field(DTYP, "asynFloat64")
+	field(SCAN, "I/O Intr")
+	field(INP, "@asyn($(PORT)) JSON:FLOAT $(TOPIC_ROOT)/json/float")
+}
+
+record(ao, "$(P)$(R)JsonFloat64Output") {
+	field(DESC, "CI Float64 Output")
+	field(DTYP, "asynFloat64")
+	field(OUT, "@asyn($(PORT)) JSON:FLOAT $(TOPIC_ROOT)/json/float")
+}
+
+record(stringin, "$(P)$(R)JsonStringInput") {
+	field(DESC, "CI String Input")
+	field(DTYP, "asynOctetRead")
+	field(SCAN, "I/O Intr")
+	field(INP, "@asyn($(PORT)) JSON:STRING $(TOPIC_ROOT)/json/string")
+}
+
+record(stringout, "$(P)$(R)JsonStringOutput") {
+	field(DESC, "CI String Output")
+	field(DTYP, "asynOctetWrite")
+	field(OUT, "@asyn($(PORT)) JSON:STRING $(TOPIC_ROOT)/json/string")
+}
+
+record(aai, "$(P)$(R)JsonIntArrayInput") {
+	field(DESC, "CI Int Array Input")
+	field(DTYP, "asynInt32ArrayIn")
+	field(SCAN, "I/O Intr")
+	field(FTVL, "LONG")
+	field(NELM, "16")
+	field(INP, "@asyn($(PORT)) JSON:INTARRAY $(TOPIC_ROOT)/json/intarray")
+}
+
+record(aao, "$(P)$(R)JsonIntArrayOutput") {
+	field(DESC, "CI Int Array Output")
+	field(DTYP, "asynInt32ArrayOut")
+	field(FTVL, "LONG")
+	field(NELM, "16")
+	field(OUT, "@asyn($(PORT)) JSON:INTARRAY $(TOPIC_ROOT)/json/intarray")
+}
+
+record(aai, "$(P)$(R)JsonFloatArrayInput") {
+	field(DESC, "CI Float Array Input")
+	field(DTYP, "asynFloat64ArrayIn")
+	field(SCAN, "I/O Intr")
+	field(FTVL, "DOUBLE")
+	field(NELM, "16")
+	field(INP, "@asyn($(PORT)) JSON:FLOATARRAY $(TOPIC_ROOT)/json/floatarray")
+}
+
+record(aao, "$(P)$(R)JsonFloatArrayOutput") {
+	field(DESC, "CI Float Array Output")
+	field(DTYP, "asynFloat64ArrayOut")
+	field(FTVL, "DOUBLE")
+	field(NELM, "16")
+	field(OUT, "@asyn($(PORT)) JSON:FLOATARRAY $(TOPIC_ROOT)/json/floatarray")
+}
+
+record(ai, "$(P)$(R)JsonAltAddressInput") {
+	field(DESC, "CI Alt Address Input")
+	field(DTYP, "asynInt32")
+	field(SCAN, "I/O Intr")
+	field(INP, "@asyn($(PORT)) JSON:INT $(TOPIC_ROOT)/json/altAddress")
+}
+
+record(ao, "$(P)$(R)JsonAltAddressOutput") {
+	field(DESC, "CI Alt Address Output")
+	field(DTYP, "asynInt32")
+	field(OUT, "@asyn($(PORT)) JSON:INT $(TOPIC_ROOT)/json/altAddress")
+}
+
+record(ai, "$(P)$(R)JsonTemplateInput") {
+	field(DESC, "CI Template Input")
+	field(DTYP, "asynInt32")
+	field(SCAN, "I/O Intr")
+	field(INP, "@asyn($(PORT)) JSON:INT $(TOPIC_ROOT)/json/writeTemplate")
+}
+
+record(ao, "$(P)$(R)JsonTemplateOutput") {
+	field(DESC, "CI Template Output")
+	field(DTYP, "asynInt32")
+	field(OUT, "@asyn($(PORT)) JSON:INT $(TOPIC_ROOT)/json/writeTemplate")
+}

tests/README.md

@@ -1,21 +1,34 @@
 # MQTT Round-Trip Tests
 
-This folder contains an integration test for the EPICS MQTT driver using `pytest` + `mosquitto` + `p4p`.
+This folder contains an integration test for the EPICS MQTT driver using `pytest` + `mosquitto` +
+`p4p`.
 
 ## What this test does
 
 - Starts a local Mosquitto broker in port **18830**.
 - Starts the test IOC by running `./st.cmd` in `iocBoot/ioctest`, pointed at the local broker.
   - Uses a unique MQTT client ID and topic root per test session to avoid crosstalk.
-- Writes values to output PVs and waits for the corresponding input PVs to receive the same value back.
-  - This ensures the basic functionality of the driver, including MQTT connectivity, topic subscription, and record
-    updates, is working correctly.
+- Writes values to output PVs and waits for the corresponding input PVs to receive the same value
+  back.
+  - This ensures the basic functionality of the driver, including MQTT connectivity, topic
+    subscription, and record updates, is working correctly.
 
 ## Run locally
 
-To run the tests locally, from the root of the repository:
+To run the tests locally, install the required dependencies:
 
 ```bash
+sudo apt-install mosquitto
 pip3 install p4p pytest
+```
+
+> By default, `mosquitto` creates a broker service in systemd that runs on `1883`, which is why the
+> test uses `18830` to avoid conflicts. You may disable the system service if you want to use the
+> default port by running `sudo systemctl disable --now mosquitto`.
+
+Then, from the root of the repository:
+
+```bash
+make # build the epics module
 pytest -q tests
 ```

tests/test_mqtt_roundtrip.py

@@ -73,7 +73,14 @@ def _put_and_wait(context, output_pv, input_pv, value, timeout=10.0):
         ("mqtt:test:StringOutput", "mqtt:test:StringInput", "epicsMQTT-ci"),
         ("mqtt:test:IntArrayOutput", "mqtt:test:IntArrayInput", [1, 2, 3, 4, 5]),
         ("mqtt:test:FloatArrayOutput", "mqtt:test:FloatArrayInput", [1.1, 2.2, 3.3, 4.4, 5.5]),
-    ],
+        ("mqtt:test:JsonInt32Output", "mqtt:test:JsonInt32Input", 42),
+        ("mqtt:test:JsonFloat64Output", "mqtt:test:JsonFloat64Input", 3.14159),
+        ("mqtt:test:JsonStringOutput", "mqtt:test:JsonStringInput", "epicsMQTT-ci"),
+        ("mqtt:test:JsonIntArrayOutput", "mqtt:test:JsonIntArrayInput", [1, 2, 3, 4, 5]),
+        ("mqtt:test:JsonFloatArrayOutput", "mqtt:test:JsonFloatArrayInput", [1.1, 2.2, 3.3, 4.4, 5.5]),
+        ("mqtt:test:JsonAltAddressOutput", "mqtt:test:JsonAltAddressInput", 43),
+        ("mqtt:test:JsonTemplateOutput", "mqtt:test:JsonTemplateInput", 44),
+     ],
 )
 def test_round_trip_via_broker(pva_context, output_pv, input_pv, value):
     _put_and_wait(pva_context, output_pv, input_pv, value)