Skip to content

Commit 2ab9d89

Browse files
codepitbullcodepitbull
authored
Merge pull request #599 from hivemq/fix/modbus-rework
Fix/modbus rework
2 parents 970a0f6 + 63e5d6e commit 2ab9d89

23 files changed

+349
-303
lines changed

‎modules/hivemq-edge-module-modbus/src/main/java/com/hivemq/edge/adapters/modbus/ModbusProtocolAdapter.java‎

Lines changed: 86 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,6 @@
1616
package com.hivemq.edge.adapters.modbus;
1717

1818
import com.hivemq.adapter.sdk.api.ProtocolAdapterInformation;
19-
import com.hivemq.adapter.sdk.api.config.PollingContext;
2019
import com.hivemq.adapter.sdk.api.data.DataPoint;
2120
import com.hivemq.adapter.sdk.api.discovery.NodeTree;
2221
import com.hivemq.adapter.sdk.api.discovery.NodeType;
@@ -34,6 +33,8 @@
3433
import com.hivemq.adapter.sdk.api.state.ProtocolAdapterState;
3534
import com.hivemq.edge.adapters.modbus.config.AddressRange;
3635
import com.hivemq.edge.adapters.modbus.config.ModbusAdapterConfig;
36+
import com.hivemq.edge.adapters.modbus.config.ModbusAdu;
37+
import com.hivemq.edge.adapters.modbus.config.ModbusDataType;
3738
import com.hivemq.edge.adapters.modbus.config.ModbusToMqttMapping;
3839
import com.hivemq.edge.adapters.modbus.impl.ModbusClient;
3940
import com.hivemq.edge.adapters.modbus.model.ModBusData;
@@ -50,16 +51,16 @@
5051
import java.util.concurrent.CompletableFuture;
5152

5253
import static com.hivemq.adapter.sdk.api.state.ProtocolAdapterState.ConnectionStatus.CONNECTED;
54+
import static com.hivemq.adapter.sdk.api.state.ProtocolAdapterState.ConnectionStatus.DISCONNECTED;
5355

5456
public class ModbusProtocolAdapter implements PollingProtocolAdapter<ModbusToMqttMapping> {
5557
private static final Logger log = LoggerFactory.getLogger(ModbusProtocolAdapter.class);
56-
private final @NotNull Object lock = new Object();
5758
private final @NotNull ProtocolAdapterInformation adapterInformation;
5859
private final @NotNull ModbusAdapterConfig adapterConfig;
5960
private final @NotNull ProtocolAdapterState protocolAdapterState;
6061
private final @NotNull AdapterFactories adapterFactories;
6162

62-
private volatile @Nullable ModbusClient modbusClient;
63+
private final @Nullable ModbusClient modbusClient;
6364
private final @NotNull Map<ModbusToMqttMapping, List<DataPoint>> lastSamples = new HashMap<>();
6465

6566
public ModbusProtocolAdapter(
@@ -70,31 +71,35 @@ public ModbusProtocolAdapter(
7071
this.adapterConfig = adapterConfig;
7172
this.protocolAdapterState = input.getProtocolAdapterState();
7273
this.adapterFactories = input.adapterFactories();
74+
this.modbusClient = new ModbusClient(adapterConfig, adapterFactories.dataPointFactory());
7375
}
7476

7577
@Override
76-
public void start(
77-
@NotNull final ProtocolAdapterStartInput input, @NotNull final ProtocolAdapterStartOutput output) {
78-
try {
79-
initConnection();
80-
output.startedSuccessfully();
81-
} catch (final Exception e) {
82-
output.failStart(e, "Exception during setup of Modbus client.");
83-
}
84-
78+
public void start(@NotNull final ProtocolAdapterStartInput input, @NotNull final ProtocolAdapterStartOutput output) {
79+
modbusClient
80+
.connect()
81+
.whenComplete((unused, throwable) -> {
82+
if (throwable == null) {
83+
output.startedSuccessfully();
84+
protocolAdapterState.setConnectionStatus(CONNECTED);
85+
} else {
86+
output.failStart(throwable, "Exception during setup of Modbus client.");
87+
}});
8588
}
8689

8790
@Override
8891
public void stop(@NotNull final ProtocolAdapterStopInput input, @NotNull final ProtocolAdapterStopOutput output) {
89-
try {
90-
if (modbusClient != null) {
91-
modbusClient.disconnect();
92-
}
93-
} catch (final Exception e) {
94-
output.failStop(e, "Error encountered closing connection to Modbus device.");
95-
return;
96-
}
97-
output.stoppedSuccessfully();
92+
modbusClient
93+
.disconnect()
94+
.whenComplete((unused,t) -> {
95+
if(t == null) {
96+
output.stoppedSuccessfully();
97+
protocolAdapterState.setConnectionStatus(DISCONNECTED);
98+
} else {
99+
output.failStop(t, "Error encountered closing connection to Modbus device.");
100+
}
101+
}
102+
);
98103
}
99104

100105
@Override
@@ -103,25 +108,15 @@ public void poll(
103108

104109
//-- If a previously linked job has terminally disconnected the client
105110
//-- we need to ensure any orphaned jobs tidy themselves up properly
106-
try {
107-
if (modbusClient != null) {
108-
if (!modbusClient.isConnected()) {
109-
modbusClient.connect().thenRun(() -> protocolAdapterState.setConnectionStatus(CONNECTED)).get();
111+
112+
readRegisters(pollingInput.getPollingContext(), modbusClient)
113+
.whenComplete((modbusdata, throwable) -> {
114+
if (throwable != null) {
115+
pollingOutput.fail(throwable, null);
116+
} else {
117+
this.captureDataSample(modbusdata, pollingOutput);
110118
}
111-
CompletableFuture.supplyAsync(() -> readRegisters(pollingInput.getPollingContext()))
112-
.whenComplete((modbusdata, throwable) -> {
113-
if (throwable != null) {
114-
pollingOutput.fail(throwable, null);
115-
} else {
116-
this.captureDataSample(modbusdata, pollingOutput);
117-
}
118-
});
119-
} else {
120-
pollingOutput.fail(new IllegalStateException("client not initialised"),"The client is not initialised.");
121-
}
122-
} catch (final Exception e) {
123-
pollingOutput.fail(e, null);
124-
}
119+
});
125120
}
126121

127122
@Override
@@ -139,21 +134,6 @@ public int getMaxPollingErrorsBeforeRemoval() {
139134
return adapterConfig.getModbusToMQTTConfig().getMaxPollingErrorsBeforeRemoval();
140135
}
141136

142-
143-
private @NotNull ModbusClient initConnection() {
144-
if (modbusClient == null) {
145-
synchronized (lock) {
146-
if (modbusClient == null) {
147-
if (log.isTraceEnabled()) {
148-
log.trace("Creating new instance of Modbus Client with {}.", adapterConfig);
149-
}
150-
modbusClient = new ModbusClient(adapterConfig, adapterFactories.dataPointFactory());
151-
}
152-
}
153-
}
154-
return modbusClient;
155-
}
156-
157137
@Override
158138
public @NotNull String getId() {
159139
return adapterConfig.getId();
@@ -218,21 +198,62 @@ private void calculateDelta(@NotNull final ModBusData modBusData, @NotNull final
218198
}
219199
}
220200

221-
protected @NotNull ModBusData readRegisters(final @NotNull ModbusToMqttMapping modbusToMqttMapping) {
222-
try {
223-
final AddressRange addressRange = modbusToMqttMapping.getAddressRange();
224-
final DataPoint dataPoint = modbusClient.readHoldingRegisters(addressRange.startIdx,
225-
addressRange.nrRegistersToRead, modbusToMqttMapping.getDataType());
226-
final ModBusData data = new ModBusData(modbusToMqttMapping);
227-
data.addDataPoint(dataPoint);
228-
return data;
229-
} catch (final Exception e) {
230-
throw new RuntimeException(e);
201+
protected static @NotNull CompletableFuture<ModBusData> readRegisters(
202+
final @NotNull ModbusToMqttMapping modbusToMqttMapping,
203+
final @NotNull ModbusClient modbusClient) {
204+
final AddressRange addressRange = modbusToMqttMapping.getAddressRange();
205+
206+
return doRead(addressRange.startIdx, addressRange.unitId, addressRange.flipRegisters, modbusToMqttMapping.getDataType(), addressRange.readType, modbusClient)
207+
.thenApply(dataPoint -> {
208+
final ModBusData data = new ModBusData(modbusToMqttMapping);
209+
data.addDataPoint(dataPoint);
210+
return data;
211+
});
212+
}
213+
214+
protected static CompletableFuture<DataPoint> doRead(
215+
final int startIdx,
216+
final int unitId,
217+
final boolean flipRegisters,
218+
final @NotNull ModbusDataType dataType,
219+
final @NotNull ModbusAdu readType,
220+
final @NotNull ModbusClient modbusClient) {
221+
switch (readType) {
222+
case HOLDING_REGISTERS:
223+
return modbusClient
224+
.readHoldingRegisters(
225+
startIdx,
226+
dataType,
227+
unitId,
228+
flipRegisters);
229+
case INPUT_REGISTERS:
230+
return modbusClient
231+
.readInputRegisters(
232+
startIdx,
233+
dataType,
234+
unitId,
235+
flipRegisters);
236+
case COILS:
237+
return modbusClient
238+
.readCoils(
239+
startIdx,
240+
unitId);
241+
case DISCRETE_INPUT:
242+
return modbusClient
243+
.readDiscreteInput(
244+
startIdx,
245+
unitId);
246+
default:
247+
return CompletableFuture.failedFuture(new Exception("Unknown read type " + readType));
231248
}
232249
}
233250

234251
private static void addAddresses(
235-
@NotNull final NodeTree tree, @NotNull final String parent, final int startIdx, final int count, final int groupIdx) {
252+
@NotNull final NodeTree tree,
253+
@NotNull final String parent,
254+
final int startIdx,
255+
final int count,
256+
final int groupIdx) {
236257

237258
String parentNode = parent;
238259
if (groupIdx < count) {

‎modules/hivemq-edge-module-modbus/src/main/java/com/hivemq/edge/adapters/modbus/ModbusProtocolAdapterFactory.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import com.hivemq.adapter.sdk.api.config.ProtocolAdapterConfig;
2424
import com.hivemq.edge.adapters.modbus.config.AddressRange;
2525
import com.hivemq.edge.adapters.modbus.config.ModbusAdapterConfig;
26+
import com.hivemq.edge.adapters.modbus.config.ModbusAdu;
2627
import com.hivemq.edge.adapters.modbus.config.ModbusDataType;
2728
import com.hivemq.edge.adapters.modbus.config.ModbusToMqttConfig;
2829
import com.hivemq.edge.adapters.modbus.config.ModbusToMqttMapping;
@@ -104,7 +105,7 @@ public ModbusProtocolAdapterFactory(final boolean writingEnabled) {
104105
context.getIncludeTimestamp(),
105106
context.getIncludeTagNames(),
106107
context.getLegacyProperties(),
107-
new AddressRange(context.getAddressRange().startIdx, context.getAddressRange().endIdx - context.getAddressRange().startIdx),
108+
new AddressRange(context.getAddressRange().startIdx, ModbusAdu.HOLDING_REGISTERS, 0, false),
108109
ModbusDataType.INT_32))
109110
.collect(Collectors.toList());
110111

‎modules/hivemq-edge-module-modbus/src/main/java/com/hivemq/edge/adapters/modbus/config/AddressRange.java‎

Lines changed: 22 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -30,18 +30,32 @@ public class AddressRange {
3030
required = true)
3131
public final int startIdx;
3232

33-
@JsonProperty(value = "nrRegistersToRead", required = true)
34-
@ModuleConfigField(title = "# of registers to read",
35-
description = "Number of registers to read",
36-
numberMin = 1,
37-
numberMax = ModbusAdapterConfig.PORT_MAX,
33+
@JsonProperty(value = "readType", required = true)
34+
@ModuleConfigField(title = "The way the register range should be read",
35+
description = "Type of read to performe on the registers",
36+
required = true)
37+
public final ModbusAdu readType;
38+
39+
@JsonProperty(value = "unitId", required = true)
40+
@ModuleConfigField(title = "The id of the unit to access",
41+
description = "Id of the unit to access on the modbus",
3842
required = true)
39-
public final int nrRegistersToRead;
43+
public final int unitId;
44+
45+
@JsonProperty(value = "flipRegisters", defaultValue = "false")
46+
@ModuleConfigField(title = "Indicates if registers should be evaluated in reverse order",
47+
description = "Registers and their contents are normally written/read as big endian, some implementations decided to write the content as big endian but to order the actual registers as little endian.",
48+
defaultValue = "false")
49+
public final boolean flipRegisters;
4050

4151
public AddressRange(
4252
@JsonProperty(value = "startIdx", required = true) final int startIdx,
43-
@JsonProperty(value = "nrRegistersToRead", required = true) final int nrRegistersToRead) {
53+
@JsonProperty(value = "readType", required = true) final ModbusAdu readType,
54+
@JsonProperty(value = "unitId", required = true) final int unitId,
55+
@JsonProperty(value = "flipRegisters", defaultValue = "false") final boolean flipRegisters) {
4456
this.startIdx = startIdx;
45-
this.nrRegistersToRead = nrRegistersToRead;
57+
this.readType = readType;
58+
this.unitId = unitId;
59+
this.flipRegisters = flipRegisters;
4660
}
4761
}

‎modules/hivemq-edge-module-modbus/src/main/java/com/hivemq/edge/adapters/modbus/config/ModbusAdu.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
package com.hivemq.edge.adapters.modbus.config;
2+
3+
public enum ModbusAdu {
4+
COILS,
5+
DISCRETE_INPUT,
6+
INPUT_REGISTERS,
7+
HOLDING_REGISTERS,
8+
}

‎modules/hivemq-edge-module-modbus/src/main/java/com/hivemq/edge/adapters/modbus/config/ModbusDataType.java‎

Lines changed: 41 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,45 @@
11
package com.hivemq.edge.adapters.modbus.config;
22

3+
import java.util.Collections;
4+
import java.util.HashMap;
5+
import java.util.Map;
6+
37
public enum ModbusDataType {
4-
INT_16,
5-
UINT_16,
6-
INT_32,
7-
UINT_32,
8-
INT_64,
9-
FLOAT_32,
10-
UTF_8
8+
BOOL("BOOL", 1),
9+
INT_16("INT_16", 1),
10+
UINT_16("UINT_16", 1),
11+
INT_32("INT_32", 2),
12+
UINT_32("UINT_32", 2),
13+
INT_64("INT_64", 4),
14+
FLOAT_32("FLOAT_32", 2),
15+
FLOAT_64("FLOAT_64", 4),
16+
UTF_8("UTF_8", 4);
17+
18+
private static final Map<String, ModbusDataType> BY_LABEL;
19+
20+
static {
21+
final Map<String, ModbusDataType> temp = new HashMap<>();
22+
for (ModbusDataType e : values()) {
23+
temp.put(e.label, e);
24+
}
25+
BY_LABEL = Collections.unmodifiableMap(temp);
26+
}
27+
28+
public final String label;
29+
public final int nrOfRegistersToRead;
30+
31+
ModbusDataType(String label, int nrOfRegistersToRead) {
32+
this.label = label;
33+
this.nrOfRegistersToRead = nrOfRegistersToRead;
34+
}
35+
36+
37+
public static ModbusDataType valueOfLabel(String label) {
38+
return BY_LABEL.get(label);
39+
}
40+
41+
@Override
42+
public String toString() {
43+
return this.label;
44+
}
1145
}

‎modules/hivemq-edge-module-modbus/src/main/java/com/hivemq/edge/adapters/modbus/config/ModbusToMqttMapping.java‎

Lines changed: 0 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -104,42 +104,6 @@ public ModbusToMqttMapping(
104104
this.addressRange = addressRange;
105105
this.userProperties = requireNonNullElseGet(userProperties, List::of);
106106
this.dataType = requireNonNullElse(dataType, ModbusDataType.INT_16);
107-
108-
final int registerCount = addressRange.nrRegistersToRead;
109-
switch (this.dataType) {
110-
case INT_16:
111-
case UINT_16:
112-
if (registerCount != 1) {
113-
throw new IllegalArgumentException("The data type " +
114-
this.dataType +
115-
" needs exactly 1 register, but " +
116-
registerCount +
117-
" registers were configured.");
118-
}
119-
break;
120-
case INT_32:
121-
case UINT_32:
122-
case FLOAT_32:
123-
if (registerCount != 2) {
124-
throw new IllegalArgumentException("The data type " +
125-
this.dataType +
126-
" needs exactly 2 registers, but " +
127-
registerCount +
128-
" registers were configured.");
129-
}
130-
break;
131-
case INT_64:
132-
if (registerCount != 4) {
133-
throw new IllegalArgumentException("The data type " +
134-
this.dataType +
135-
" needs exactly 4 registers, but " +
136-
registerCount +
137-
" registers were configured.");
138-
}
139-
break;
140-
case UTF_8:
141-
default:
142-
}
143107
}
144108

145109
public @NotNull AddressRange getAddressRange() {

0 commit comments

Comments
 (0)