[modbus] Modbus transport API simplification (#7994)

* [modbus] connection closing behaviour finetuned

The binding closes TCP/serial connections connections as per user
configuration. It is either
- every time, immediately after a transaction (reconnectAfterMillis=0)
- *after* a read/write transaction, if the connection has been open "too
  long" (configurable with reconnectAfterMillis)

We have an obvious downside to this simple logic -- the connection
can remain open "indefinitely" if there are no transactions occurring.
With Modbus we are quite often dealing with PLCs and other embedded
systems with limited resources, and "reserving" the connection is not
always something we want.

Previously (2.5.x branch) connections were also closed when
- a regular poll task was unregistered (i.e. we stop reading regularly
from a modbus server ("slave"). Since most users have regular polling in
place, so this ensured that connections do get closed.

In this commit, the behaviour is adjusted such that connections are
closed when last communication interface pointing to the server/serial
port (i.e. "endpoint") is close()'d.

With modbus binding this basically means when the tcp or serial thing
is removed / disabled, the connections is closed/freed, but only if it
is the last thing pointing to that server or serial port.

Since modbus.sunspec binding reuses modbus serial & tcp endpoint things,
the same note applies for modbus.sunspec binding.

This is change in functional behaviour but in a way is logical. We can
further introduce to have "delayed"/"deferred"/"debounce" connection
closing connections as per the setting reconnectAfterMillis even in
situations where communication interface is still open.

* [modbus] Check and disconnect idle connections without transactions
* [modbus] mvn spotless:apply
* [modbubs] Fixed log message
* [modbus] Race condition fix

The CountDownLatch was used as a guard (latch.await) in many tests to
wait for callbacks to be called before proceeding with assertions.

Since the latch was countDown() beginning of the callback, we introduced
a race condition with the subsequent assertions and updating the other
counters used in the subsequent assertions.

This commit updates the CountDownLatch as the last step of the callback,
resolving the race condition.

* [modbus] small test fix
* [modbus] readcallback changed to nonnull
* [modbus] Refactored ModbusCommunicationInterface to have seperate callback for result and failure

However I had to dig deep to reach all the affected parts.
Also there is a new callback, and a new "result" type to communicate
the failures.
* [modbus] SmokeTest refactored to new api
* [modbus] Modbus bundle refactored to use the new api
* [modbus][sunspec] refactored sunspec bundle to use the new modbus API
* [modbus] refactor modbus tests to use the new api
* [modbus] Removed ModbusWriteCallback interface from ModbusDataThingHandler
Also reset ModbusDataThingHandler testWriteHandling generic to it's original form
* [modbus] ModbusDataThingHandler does not implement ModbusReadCallback anymore
Instead it has a public onReadResult method. ModbusPollerThingHandler changed to
work with ModbusDataThingHandler children.
* [modbus] Fixed caching in ModbusPollerThingHandler
* [modbus] read modbus data as Optionals
* [modbus] toString for PollResult
* [modbus] fixed confusing variable name
* [modbus] Disallow null callbacks
* [modbus] mvn spotless:apply
* [modbus] Removing Nullable decorations
* [modbus] submitOneTimeWrite simplification
* [modbus] Less verbose logging
* [modbus] submitOneTimePoll simplification
* [modbus] Less verbose logging
* [modbus] Many null warnings removed
* [modbus] Fix: no need for a @nonnull annotation because it is default
* [modbus] Removing unneeded Nullable, using final in immutables
* [modbus] Explicit functional interface
* [modbus] @nullable and @NonNullByDefault aligned with coding conventions
* [modbus] Collections.emptyMap instead of allocating new map every time.

Signed-off-by: Sami Salonen <ssalonen@gmail.com>

Co-authored-by: Nagy Attila Gábor <mrbig@sneaker.hu>

bundles/org.openhab.binding.modbus.sunspec/src/main/java/org/openhab/binding/modbus/sunspec/internal/SunSpecHandlerFactory.java

@@ -23,10 +23,7 @@
 import org.eclipse.smarthome.core.thing.binding.ThingHandlerFactory;
 import org.openhab.binding.modbus.sunspec.internal.handler.InverterHandler;
 import org.openhab.binding.modbus.sunspec.internal.handler.MeterHandler;
-import org.openhab.io.transport.modbus.ModbusManager;
-import org.osgi.service.component.annotations.Activate;
 import org.osgi.service.component.annotations.Component;
-import org.osgi.service.component.annotations.Reference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -45,23 +42,6 @@ public class SunSpecHandlerFactory extends BaseThingHandlerFactory {
  */
  private final Logger logger = LoggerFactory.getLogger(SunSpecHandlerFactory.class);
 
- /**
- * Reference to the modbus manager
- */
- private ModbusManager manager;
-
- /**
- * This factory needs a reference to the ModbusManager wich is provided
- * by the org.openhab.io.transport.modbus bundle. Please make
- * sure it's installed and enabled before using this bundle
- *
- * @param manager reference to the ModbusManager. We use this for modbus communication
- */
- @Activate
- public SunSpecHandlerFactory(@Reference ModbusManager manager) {
- this.manager = manager;
- }
-
  @Override
  public boolean supportsThingType(ThingTypeUID thingTypeUID) {
  return SUPPORTED_THING_TYPES_UIDS.containsValue(thingTypeUID);
@@ -75,12 +55,12 @@ public boolean supportsThingType(ThingTypeUID thingTypeUID) {
  || thingTypeUID.equals(THING_TYPE_INVERTER_SPLIT_PHASE)
  || thingTypeUID.equals(THING_TYPE_INVERTER_THREE_PHASE)) {
  logger.debug("New InverterHandler created");
- return new InverterHandler(thing, manager);
+ return new InverterHandler(thing);
  } else if (thingTypeUID.equals(THING_TYPE_METER_SINGLE_PHASE)
  || thingTypeUID.equals(THING_TYPE_METER_SPLIT_PHASE) || thingTypeUID.equals(THING_TYPE_METER_WYE_PHASE)
  || thingTypeUID.equals(THING_TYPE_METER_DELTA_PHASE)) {
  logger.debug("New MeterHandler created");
- return new MeterHandler(thing, manager);
+ return new MeterHandler(thing);
  }
 
  return null;

bundles/org.openhab.binding.modbus.sunspec/src/main/java/org/openhab/binding/modbus/sunspec/internal/discovery/SunspecDiscoveryProcess.java

@@ -32,18 +32,14 @@
 import org.openhab.binding.modbus.sunspec.internal.dto.CommonModelBlock;
 import org.openhab.binding.modbus.sunspec.internal.dto.ModelBlock;
 import org.openhab.binding.modbus.sunspec.internal.parser.CommonModelParser;
-import org.openhab.io.transport.modbus.BasicModbusReadRequestBlueprint;
-import org.openhab.io.transport.modbus.BasicPollTaskImpl;
-import org.openhab.io.transport.modbus.BitArray;
+import org.openhab.io.transport.modbus.AsyncModbusFailure;
 import org.openhab.io.transport.modbus.ModbusBitUtilities;
+import org.openhab.io.transport.modbus.ModbusCommunicationInterface;
 import org.openhab.io.transport.modbus.ModbusConstants.ValueType;
-import org.openhab.io.transport.modbus.ModbusReadCallback;
 import org.openhab.io.transport.modbus.ModbusReadFunctionCode;
 import org.openhab.io.transport.modbus.ModbusReadRequestBlueprint;
 import org.openhab.io.transport.modbus.ModbusRegisterArray;
-import org.openhab.io.transport.modbus.ModbusSlaveErrorResponseException;
-import org.openhab.io.transport.modbus.PollTask;
-import org.openhab.io.transport.modbus.endpoint.ModbusSlaveEndpoint;
+import org.openhab.io.transport.modbus.exception.ModbusSlaveErrorResponseException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -68,11 +64,6 @@ public class SunspecDiscoveryProcess {
  */
  private final ModbusEndpointThingHandler handler;
 
- /**
- * The endpoint where we can reach the device
- */
- private final ModbusSlaveEndpoint endpoint;
-
  /**
  * Listener for the discovered devices. We get this
  * from the main discovery service, and it is used to
@@ -116,6 +107,11 @@ public class SunspecDiscoveryProcess {
  */
  private @Nullable CommonModelBlock lastCommonBlock = null;
 
+ /**
+ * Communication interface to the endpoint
+ */
+ private ModbusCommunicationInterface comms;
+
  /**
  * New instances of this class should get a reference to the handler
  *
@@ -125,9 +121,9 @@ public SunspecDiscoveryProcess(ModbusEndpointThingHandler handler, ModbusDiscove
  throws EndpointNotInitializedException {
  this.handler = handler;
 
- ModbusSlaveEndpoint endpoint = this.handler.asSlaveEndpoint();
- if (endpoint != null) {
- this.endpoint = endpoint;
+ ModbusCommunicationInterface localComms = handler.getCommunicationInterface();
+ if (localComms != null) {
+ this.comms = localComms;
  } else {
  throw new EndpointNotInitializedException();
  }
@@ -158,30 +154,13 @@ public void detectModel() {
  baseAddress = possibleAddresses.poll();
  logger.trace("Beginning scan for SunSpec device at address {}", baseAddress);
 
- BasicModbusReadRequestBlueprint request = new BasicModbusReadRequestBlueprint(slaveId,
+ ModbusReadRequestBlueprint request = new ModbusReadRequestBlueprint(slaveId,
  ModbusReadFunctionCode.READ_MULTIPLE_REGISTERS, baseAddress, // Start address
  SUNSPEC_ID_SIZE, // number or words to return
  maxTries);
 
- PollTask task = new BasicPollTaskImpl(endpoint, request, new ModbusReadCallback() {
-
- @Override
- public void onRegisters(ModbusReadRequestBlueprint request, ModbusRegisterArray registers) {
- headerReceived(registers);
- }
-
- @Override
- public void onError(ModbusReadRequestBlueprint request, Exception error) {
- handleError(error);
- }
-
- @Override
- public void onBits(@Nullable ModbusReadRequestBlueprint request, @Nullable BitArray bits) {
- // don't care, we don't expect this result
- }
- });
-
- handler.getManagerRef().get().submitOneTimePoll(task);
+ comms.submitOneTimePoll(request, result -> result.getRegisters().ifPresent(this::headerReceived),
+ this::handleError);
  }
 
  /**
@@ -210,30 +189,13 @@ private void headerReceived(ModbusRegisterArray registers) {
  */
  private void lookForModelBlock() {
 
- BasicModbusReadRequestBlueprint request = new BasicModbusReadRequestBlueprint(slaveId,
+ ModbusReadRequestBlueprint request = new ModbusReadRequestBlueprint(slaveId,
  ModbusReadFunctionCode.READ_MULTIPLE_REGISTERS, baseAddress, // Start address
  MODEL_HEADER_SIZE, // number or words to return
  maxTries);
 
- PollTask task = new BasicPollTaskImpl(endpoint, request, new ModbusReadCallback() {
-
- @Override
- public void onRegisters(ModbusReadRequestBlueprint request, ModbusRegisterArray registers) {
- modelBlockReceived(registers);
- }
-
- @Override
- public void onError(ModbusReadRequestBlueprint request, Exception error) {
- handleError(error);
- }
-
- @Override
- public void onBits(@Nullable ModbusReadRequestBlueprint request, @Nullable BitArray bits) {
- // don't care, we don't expect this result
- }
- });
-
- handler.getManagerRef().get().submitOneTimePoll(task);
+ comms.submitOneTimePoll(request, result -> result.getRegisters().ifPresent(this::modelBlockReceived),
+ this::handleError);
  }
 
  /**
@@ -280,30 +242,13 @@ private void modelBlockReceived(ModbusRegisterArray registers) {
  * @param block
  */
  private void readCommonBlock(ModelBlock block) {
- BasicModbusReadRequestBlueprint request = new BasicModbusReadRequestBlueprint(slaveId,
+ ModbusReadRequestBlueprint request = new ModbusReadRequestBlueprint(slaveId,
  ModbusReadFunctionCode.READ_MULTIPLE_REGISTERS, block.address, // Start address
  block.length, // number or words to return
  maxTries);
 
- PollTask task = new BasicPollTaskImpl(endpoint, request, new ModbusReadCallback() {
-
- @Override
- public void onRegisters(ModbusReadRequestBlueprint request, ModbusRegisterArray registers) {
- parseCommonBlock(registers);
- }
-
- @Override
- public void onError(ModbusReadRequestBlueprint request, Exception error) {
- handleError(error);
- }
-
- @Override
- public void onBits(@Nullable ModbusReadRequestBlueprint request, @Nullable BitArray bits) {
- // don't care, we don't expect this result
- }
- });
-
- handler.getManagerRef().get().submitOneTimePoll(task);
+ comms.submitOneTimePoll(request, result -> result.getRegisters().ifPresent(this::parseCommonBlock),
+ this::handleError);
  }
 
  /**
@@ -367,25 +312,22 @@ private void parsingFinished() {
  /**
  * Handle errors received during communication
  */
- private void handleError(Exception error) {
- String msg = "";
- String cls = "";
-
- if (blocksFound > 1 && error instanceof ModbusSlaveErrorResponseException) {
- int code = ((ModbusSlaveErrorResponseException) error).getExceptionCode();
+ private void handleError(AsyncModbusFailure<ModbusReadRequestBlueprint> failure) {
+ if (blocksFound > 1 && failure.getCause() instanceof ModbusSlaveErrorResponseException) {
+ int code = ((ModbusSlaveErrorResponseException) failure.getCause()).getExceptionCode();
  if (code == ModbusSlaveErrorResponseException.ILLEGAL_DATA_ACCESS
  || code == ModbusSlaveErrorResponseException.ILLEGAL_DATA_VALUE) {
  // It is very likely that the slave does not report an end block (0xffff) after the main blocks
  // so we treat this situation as normal.
  logger.debug(
- "Seems like slave device does not report an end block. Continouing with the dectected blocks");
+ "Seems like slave device does not report an end block. Continuing with the dectected blocks");
  parsingFinished();
  return;
  }
  }
 
- cls = error.getClass().getName();
- msg = error.getMessage();
+ String cls = failure.getCause().getClass().getName();
+ String msg = failure.getCause().getMessage();
 
  logger.warn("Error with read at address {}: {} {}", baseAddress, cls, msg);