git.basschouten.com Git - openhab-addons.git/blob

   1 /**
   2  * Copyright (c) 2010-2020 Contributors to the openHAB project
   3  *
   4  * See the NOTICE file(s) distributed with this work for additional
   5  * information.
   6  *
   7  * This program and the accompanying materials are made available under the
   8  * terms of the Eclipse Public License 2.0 which is available at
   9  * http://www.eclipse.org/legal/epl-2.0
  10  *
  11  * SPDX-License-Identifier: EPL-2.0
  12  */
  13 package org.openhab.io.transport.modbus.internal.pooling;
  14
  15 import java.net.InetAddress;
  16 import java.net.UnknownHostException;
  17 import java.util.Map;
  18 import java.util.concurrent.ConcurrentHashMap;
  19 import java.util.function.Function;
  20
  21 import org.apache.commons.pool2.BaseKeyedPooledObjectFactory;
  22 import org.apache.commons.pool2.PooledObject;
  23 import org.apache.commons.pool2.impl.DefaultPooledObject;
  24 import org.eclipse.jdt.annotation.NonNullByDefault;
  25 import org.eclipse.jdt.annotation.Nullable;
  26 import org.openhab.io.transport.modbus.endpoint.EndpointPoolConfiguration;
  27 import org.openhab.io.transport.modbus.endpoint.ModbusIPSlaveEndpoint;
  28 import org.openhab.io.transport.modbus.endpoint.ModbusSerialSlaveEndpoint;
  29 import org.openhab.io.transport.modbus.endpoint.ModbusSlaveEndpoint;
  30 import org.openhab.io.transport.modbus.endpoint.ModbusSlaveEndpointVisitor;
  31 import org.openhab.io.transport.modbus.endpoint.ModbusTCPSlaveEndpoint;
  32 import org.openhab.io.transport.modbus.endpoint.ModbusUDPSlaveEndpoint;
  33 import org.slf4j.Logger;
  34 import org.slf4j.LoggerFactory;
  35
  36 import net.wimpi.modbus.net.ModbusSlaveConnection;
  37 import net.wimpi.modbus.net.SerialConnection;
  38 import net.wimpi.modbus.net.TCPMasterConnection;
  39 import net.wimpi.modbus.net.UDPMasterConnection;
  40
  41 /**
  42  * ModbusSlaveConnectionFactoryImpl responsible of the lifecycle of modbus slave connections
  43  *
  44  * The actual pool uses instance of this class to create and destroy connections as-needed.
  45  *
  46  * The overall functionality goes as follow
  47  * - create: create connection object but do not connect it yet
  48  * - destroyObject: close connection and free all resources. Called by the pool when the pool is being closed or the
  49  * object is invalidated.
  50  * - activateObject: prepare connection to be used. In practice, connect if disconnected
  51  * - passivateObject: passivate connection before returning it back to the pool. Currently, passivateObject closes all
  52  * IP-based connections every now and then (reconnectAfterMillis). Serial connections we keep open.
  53  * - wrap: wrap created connection to pooled object wrapper class. It tracks usage statistics and last connection time.
  54  *
  55  * Note that the implementation must be thread safe.
  56  *
  57  * @author Sami Salonen - Initial contribution
  58  */
  59 @NonNullByDefault
  60 public class ModbusSlaveConnectionFactoryImpl
  61         extends BaseKeyedPooledObjectFactory<ModbusSlaveEndpoint, ModbusSlaveConnection> {
  62
  63     class PooledConnection extends DefaultPooledObject<ModbusSlaveConnection> {
  64
  65         private volatile long lastConnected;
  66         private volatile @Nullable ModbusSlaveEndpoint endpoint;
  67
  68         public PooledConnection(ModbusSlaveConnection object) {
  69             super(object);
  70         }
  71
  72         public long getLastConnected() {
  73             return lastConnected;
  74         }
  75
  76         public void setLastConnected(ModbusSlaveEndpoint endpoint, long lastConnected) {
  77             this.endpoint = endpoint;
  78             this.lastConnected = lastConnected;
  79         }
  80
  81         /**
  82          *
  83          * Reset connection if it is too old or fulfills some of the other criteria
  84          *
  85          * @param activityName ongoing activity calling this method. For logging
  86          * @return whether connection was reseted
  87          */
  88         public boolean maybeResetConnection(String activityName) {
  89             ModbusSlaveEndpoint localEndpoint = endpoint;
  90             if (localEndpoint == null) {
  91                 // We have not connected yet, abort
  92                 // Without endpoint we have no age parameters available (endpointPoolConfigs &
  93                 // disconnectIfConnectedBefore)
  94                 return false;
  95             }
  96             long localLastConnected = lastConnected;
  97
  98             ModbusSlaveConnection connection = getObject();
  99
 100             @Nullable
 101             EndpointPoolConfiguration configuration = endpointPoolConfigs.get(localEndpoint);
 102             long reconnectAfterMillis = configuration == null ? 0 : configuration.getReconnectAfterMillis();
 103             long connectionAgeMillis = System.currentTimeMillis() - localLastConnected;
 104             long disconnectIfConnectedBeforeMillis = disconnectIfConnectedBefore.getOrDefault(localEndpoint, -1L);
 105             boolean disconnectSinceTooOldConnection = disconnectIfConnectedBeforeMillis < 0L ? false
 106                     : localLastConnected <= disconnectIfConnectedBeforeMillis;
 107             boolean shouldBeDisconnected = (reconnectAfterMillis == 0
 108                     || (reconnectAfterMillis > 0 && connectionAgeMillis > reconnectAfterMillis)
 109                     || disconnectSinceTooOldConnection);
 110             if (shouldBeDisconnected) {
 111                 logger.trace(
 112                         "({}) Connection {} (endpoint {}) age {}ms is over the reconnectAfterMillis={}ms limit or has been connection time ({}) is after the \"disconnectBeforeConnectedMillis\"={} -> disconnecting.",
 113                         activityName, connection, localEndpoint, connectionAgeMillis, reconnectAfterMillis,
 114                         localLastConnected, disconnectIfConnectedBeforeMillis);
 115                 connection.resetConnection();
 116                 return true;
 117             } else {
 118                 logger.trace(
 119                         "({}) Connection {} (endpoint {}) age ({}ms) is below the reconnectAfterMillis ({}ms) limit and connection time ({}) is after the \"disconnectBeforeConnectedMillis\"={}. Keep the connection open.",
 120                         activityName, connection, localEndpoint, connectionAgeMillis, reconnectAfterMillis,
 121                         localLastConnected, disconnectIfConnectedBeforeMillis);
 122                 return false;
 123             }
 124         }
 125     }
 126
 127     private final Logger logger = LoggerFactory.getLogger(ModbusSlaveConnectionFactoryImpl.class);
 128     private volatile Map<ModbusSlaveEndpoint, @Nullable EndpointPoolConfiguration> endpointPoolConfigs = new ConcurrentHashMap<>();
 129     private volatile Map<ModbusSlaveEndpoint, Long> lastPassivateMillis = new ConcurrentHashMap<>();
 130     private volatile Map<ModbusSlaveEndpoint, Long> lastConnectMillis = new ConcurrentHashMap<>();
 131     private volatile Map<ModbusSlaveEndpoint, Long> disconnectIfConnectedBefore = new ConcurrentHashMap<>();
 132     private volatile Function<ModbusSlaveEndpoint, @Nullable EndpointPoolConfiguration> defaultPoolConfigurationFactory = endpoint -> null;
 133
 134     private @Nullable InetAddress getInetAddress(ModbusIPSlaveEndpoint key) {
 135         try {
 136             return InetAddress.getByName(key.getAddress());
 137         } catch (UnknownHostException e) {
 138             logger.error("KeyedPooledModbusSlaveConnectionFactory: Unknown host: {}. Connection creation failed.",
 139                     e.getMessage());
 140             return null;
 141         }
 142     }
 143
 144     @Override
 145     public ModbusSlaveConnection create(ModbusSlaveEndpoint endpoint) throws Exception {
 146         return endpoint.accept(new ModbusSlaveEndpointVisitor<ModbusSlaveConnection>() {
 147             @Override
 148             public @Nullable ModbusSlaveConnection visit(ModbusSerialSlaveEndpoint modbusSerialSlavePoolingKey) {
 149                 SerialConnection connection = new SerialConnection(modbusSerialSlavePoolingKey.getSerialParameters());
 150                 logger.trace("Created connection {} for endpoint {}", connection, modbusSerialSlavePoolingKey);
 151                 return connection;
 152             }
 153
 154             @Override
 155             public @Nullable ModbusSlaveConnection visit(ModbusTCPSlaveEndpoint key) {
 156                 InetAddress address = getInetAddress(key);
 157                 if (address == null) {
 158                     return null;
 159                 }
 160                 EndpointPoolConfiguration config = getEndpointPoolConfiguration(key);
 161                 int connectTimeoutMillis = 0;
 162                 if (config != null) {
 163                     connectTimeoutMillis = config.getConnectTimeoutMillis();
 164                 }
 165                 TCPMasterConnection connection = new TCPMasterConnection(address, key.getPort(), connectTimeoutMillis);
 166                 logger.trace("Created connection {} for endpoint {}", connection, key);
 167                 return connection;
 168             }
 169
 170             @Override
 171             public @Nullable ModbusSlaveConnection visit(ModbusUDPSlaveEndpoint key) {
 172                 InetAddress address = getInetAddress(key);
 173                 if (address == null) {
 174                     return null;
 175                 }
 176                 UDPMasterConnection connection = new UDPMasterConnection(address, key.getPort());
 177                 logger.trace("Created connection {} for endpoint {}", connection, key);
 178                 return connection;
 179             }
 180         });
 181     }
 182
 183     @Override
 184     public PooledObject<ModbusSlaveConnection> wrap(ModbusSlaveConnection connection) {
 185         return new PooledConnection(connection);
 186     }
 187
 188     @Override
 189     public void destroyObject(ModbusSlaveEndpoint endpoint, @Nullable PooledObject<ModbusSlaveConnection> obj) {
 190         if (obj == null) {
 191             return;
 192         }
 193         logger.trace("destroyObject for connection {} and endpoint {} -> closing the connection", obj.getObject(),
 194                 endpoint);
 195         obj.getObject().resetConnection();
 196     }
 197
 198     @Override
 199     public void activateObject(ModbusSlaveEndpoint endpoint, @Nullable PooledObject<ModbusSlaveConnection> obj)
 200             throws Exception {
 201         if (obj == null) {
 202             return;
 203         }
 204         ModbusSlaveConnection connection = obj.getObject();
 205         try {
 206             @Nullable
 207             EndpointPoolConfiguration config = getEndpointPoolConfiguration(endpoint);
 208             if (!connection.isConnected()) {
 209                 tryConnect(endpoint, obj, connection, config);
 210             }
 211
 212             if (config != null) {
 213                 long waited = waitAtleast(lastPassivateMillis.get(endpoint), config.getInterTransactionDelayMillis());
 214                 logger.trace(
 215                         "Waited {}ms (interTransactionDelayMillis {}ms) before giving returning connection {} for endpoint {}, to ensure delay between transactions.",
 216                         waited, config.getInterTransactionDelayMillis(), obj.getObject(), endpoint);
 217             }
 218         } catch (InterruptedException e) {
 219             // Someone wants to cancel us, reset the connection and abort
 220             if (connection.isConnected()) {
 221                 connection.resetConnection();
 222             }
 223         } catch (Exception e) {
 224             logger.error("Error connecting connection {} for endpoint {}: {}", obj.getObject(), endpoint,
 225                     e.getMessage());
 226         }
 227     }
 228
 229     @Override
 230     public void passivateObject(ModbusSlaveEndpoint endpoint, @Nullable PooledObject<ModbusSlaveConnection> obj) {
 231         if (obj == null) {
 232             return;
 233         }
 234         ModbusSlaveConnection connection = obj.getObject();
 235         logger.trace("Passivating connection {} for endpoint {}...", connection, endpoint);
 236         lastPassivateMillis.put(endpoint, System.currentTimeMillis());
 237         ((PooledConnection) obj).maybeResetConnection("passivate");
 238         logger.trace("...Passivated connection {} for endpoint {}", obj.getObject(), endpoint);
 239     }
 240
 241     @Override
 242     public boolean validateObject(ModbusSlaveEndpoint key, @Nullable PooledObject<ModbusSlaveConnection> p) {
 243         boolean valid = p != null && p.getObject().isConnected();
 244         logger.trace("Validating endpoint {} connection {} -> {}", key, p.getObject(), valid);
 245         return valid;
 246     }
 247
 248     /**
 249      * Configure general connection settings with a given endpoint
 250      *
 251      * @param endpoint endpoint to configure
 252      * @param configuration configuration for the endpoint. Use null to reset the configuration to default settings.
 253      */
 254     public void setEndpointPoolConfiguration(ModbusSlaveEndpoint endpoint, @Nullable EndpointPoolConfiguration config) {
 255         if (config == null) {
 256             endpointPoolConfigs.remove(endpoint);
 257         } else {
 258             endpointPoolConfigs.put(endpoint, config);
 259         }
 260     }
 261
 262     /**
 263      * Get general configuration settings applied to a given endpoint
 264      *
 265      * Note that default configuration settings are returned in case the endpoint has not been configured.
 266      *
 267      * @param endpoint endpoint to query
 268      * @return general connection settings of the given endpoint
 269      */
 270     @SuppressWarnings("null")
 271     public @Nullable EndpointPoolConfiguration getEndpointPoolConfiguration(ModbusSlaveEndpoint endpoint) {
 272         @Nullable
 273         EndpointPoolConfiguration config = endpointPoolConfigs.computeIfAbsent(endpoint,
 274                 defaultPoolConfigurationFactory);
 275         return config;
 276     }
 277
 278     /**
 279      * Set default factory for {@link EndpointPoolConfiguration}
 280      *
 281      * @param defaultPoolConfigurationFactory function providing defaults for a given endpoint
 282      */
 283     public void setDefaultPoolConfigurationFactory(
 284             Function<ModbusSlaveEndpoint, @Nullable EndpointPoolConfiguration> defaultPoolConfigurationFactory) {
 285         this.defaultPoolConfigurationFactory = defaultPoolConfigurationFactory;
 286     }
 287
 288     private void tryConnect(ModbusSlaveEndpoint endpoint, PooledObject<ModbusSlaveConnection> obj,
 289             ModbusSlaveConnection connection, @Nullable EndpointPoolConfiguration config) throws Exception {
 290         if (connection.isConnected()) {
 291             return;
 292         }
 293         int tryIndex = 0;
 294         Long lastConnect = lastConnectMillis.get(endpoint);
 295         int maxTries = config == null ? 1 : config.getConnectMaxTries();
 296         do {
 297             try {
 298                 if (config != null) {
 299                     long waited = waitAtleast(lastConnect,
 300                             Math.max(config.getInterConnectDelayMillis(), config.getInterTransactionDelayMillis()));
 301                     if (waited > 0) {
 302                         logger.trace(
 303                                 "Waited {}ms (interConnectDelayMillis {}ms, interTransactionDelayMillis {}ms) before "
 304                                         + "connecting disconnected connection {} for endpoint {}, to allow delay "
 305                                         + "between connections re-connects",
 306                                 waited, config.getInterConnectDelayMillis(), config.getInterTransactionDelayMillis(),
 307                                 obj.getObject(), endpoint);
 308                     }
 309                 }
 310                 connection.connect();
 311                 long curTime = System.currentTimeMillis();
 312                 ((PooledConnection) obj).setLastConnected(endpoint, curTime);
 313                 lastConnectMillis.put(endpoint, curTime);
 314                 break;
 315             } catch (InterruptedException e) {
 316                 logger.error("connect try {}/{} error: {}. Aborting since interrupted. Connection {}. Endpoint {}.",
 317                         tryIndex, maxTries, e.getMessage(), connection, endpoint);
 318                 throw e;
 319             } catch (Exception e) {
 320                 tryIndex++;
 321                 logger.error("connect try {}/{} error: {}. Connection {}. Endpoint {}", tryIndex, maxTries,
 322                         e.getMessage(), connection, endpoint);
 323                 if (tryIndex >= maxTries) {
 324                     logger.error("re-connect reached max tries {}, throwing last error: {}. Connection {}. Endpoint {}",
 325                             maxTries, e.getMessage(), connection, endpoint);
 326                     throw e;
 327                 }
 328                 lastConnect = System.currentTimeMillis();
 329             }
 330         } while (true);
 331     }
 332
 333     /**
 334      * Sleep until <code>waitMillis</code> has passed from <code>lastOperation</code>
 335      *
 336      * @param lastOperation last time operation was executed, or null if it has not been executed
 337      * @param waitMillis
 338      * @return milliseconds slept
 339      * @throws InterruptedException
 340      */
 341     public static long waitAtleast(@Nullable Long lastOperation, long waitMillis) throws InterruptedException {
 342         if (lastOperation == null) {
 343             return 0;
 344         }
 345         long millisSinceLast = System.currentTimeMillis() - lastOperation;
 346         long millisToWaitStill = Math.min(waitMillis, Math.max(0, waitMillis - millisSinceLast));
 347         try {
 348             Thread.sleep(millisToWaitStill);
 349         } catch (InterruptedException e) {
 350             LoggerFactory.getLogger(ModbusSlaveConnectionFactoryImpl.class).debug("wait interrupted", e);
 351             throw e;
 352         }
 353         return millisToWaitStill;
 354     }
 355
 356     /**
 357      * Disconnect returning connections which have been connected before certain time
 358      *
 359      * @param disconnectBeforeConnectedMillis disconnected connections that have been connected before this time
 360      */
 361     public void disconnectOnReturn(ModbusSlaveEndpoint endpoint, long disconnectBeforeConnectedMillis) {
 362         disconnectIfConnectedBefore.put(endpoint, disconnectBeforeConnectedMillis);
 363     }
 364 }