Implemented autoconnect for transport connectors

Author: dariol83

eu.dariolucia.reatmetric.api/src/main/java/eu/dariolucia/reatmetric/api/transport/AbstractTransportConnector.java

@@ -50,6 +50,7 @@ abstract public class AbstractTransportConnector implements ITransportConnector
     private volatile boolean prepared = false;
     private volatile boolean initialised = false;
     private volatile boolean busy = false;
+    private volatile boolean autoReconnect = false;
 
     protected AbstractTransportConnector(String name, String description) {
         this();
@@ -128,6 +129,20 @@ public boolean isInitialised() {
         return initialised;
     }
 
+    @Override
+    public void setReconnect(boolean autoReconnect) {
+        boolean toNotify = !Objects.equals(autoReconnect, this.autoReconnect);
+        this.autoReconnect = autoReconnect;
+        if(toNotify) {
+            notifySubscribers();
+        }
+    }
+
+    @Override
+    public boolean isReconnect() {
+        return this.autoReconnect;
+    }
+
     @Override
     public void initialise(Map<String, Object> properties) {
         checkPrepared();
@@ -216,7 +231,7 @@ public void dispose() {
     protected abstract void doDispose();
 
     private void notifySubscribers() {
-        TransportStatus status = new TransportStatus(name, connectionStatus, lastTxRate, lastRxRate, lastAlarmState);
+        TransportStatus status = new TransportStatus(name, connectionStatus, lastTxRate, lastRxRate, lastAlarmState, autoReconnect);
         this.subscribers.forEach((s) -> {
             try {
                 if(LOG.isLoggable(Level.FINER)) {

eu.dariolucia.reatmetric.api/src/main/java/eu/dariolucia/reatmetric/api/transport/ITransportConnector.java

@@ -40,35 +40,41 @@ public interface ITransportConnector extends Remote {
      * Return the name of the connector.
      *
      * @return the name of the connector
+     * @throws RemoteException in case of issues on the remote side
      */
     String getName() throws RemoteException;
 
     /**
      * Return the description of the connector.
      *
      * @return the description of the connector
+     * @throws RemoteException in case of issues on the remote side
      */
     String getDescription() throws RemoteException;
 
     /**
      * Return the current connection status.
      *
      * @return the current connection status
+     * @throws RemoteException in case of issues on the remote side
      */
     TransportConnectionStatus getConnectionStatus() throws RemoteException;
 
     /**
      * Ask the connector to finalise its construction. This method shall be called before calling any of the other
-     * {@link ITransportConnector} methods. Failing in doing so may result in undefined behaviour.
+     * {@link ITransportConnector} methods. As such, it is typically invoked by the driver responsible for the creation
+     * of the connector. Failing to invoke this method may result in undefined behaviour.
      *
      * Not supposed to be remoted.
+     * @throws RemoteException in case of issues on the remote side
      */
     void prepare() throws RemoteException;
 
     /**
      * Return the initialisation status.
      *
      * @return true if the connector is initialised, otherwise false
+     * @throws RemoteException in case of issues on the remote side
      */
     boolean isInitialised() throws RemoteException;
 
@@ -77,27 +83,31 @@ public interface ITransportConnector extends Remote {
      *
      * @param properties the properties that this connector shall use
      * @throws TransportException in case of problems
+     * @throws RemoteException in case of issues on the remote side
      */
     void initialise(Map<String, Object> properties) throws TransportException, RemoteException;
 
     /**
      * Ask the connector to perform the connection to the external system.
      *
      * @throws TransportException in case of problems
+     * @throws RemoteException in case of issues on the remote side
      */
     void connect() throws TransportException, RemoteException;
 
     /**
      * Ask the connector to disconnect from the external system.
      *
      * @throws TransportException in case of problems
+     * @throws RemoteException in case of issues on the remote side
      */
     void disconnect() throws TransportException, RemoteException;
 
     /**
      * Ask the connector to abort the connection to the external system.
      *
      * @throws TransportException in case of problems
+     * @throws RemoteException in case of issues on the remote side
      */
     void abort() throws TransportException, RemoteException;
 
@@ -106,6 +116,7 @@ public interface ITransportConnector extends Remote {
      * implementation after calling this method may result in undefined behaviour.
      *
      * Not supposed to be remoted.
+     * @throws RemoteException in case of issues on the remote side
      */
     void dispose() throws RemoteException;
 
@@ -115,27 +126,46 @@ public interface ITransportConnector extends Remote {
      * required value type.
      *
      * @return the initialisation map
+     * @throws RemoteException in case of issues on the remote side
      */
     Map<String, Pair<String, ValueTypeEnum>> getSupportedProperties() throws RemoteException;
 
     /**
      * Return the current initialisation property values.
      *
      * @return the current initialisation property values
+     * @throws RemoteException in case of issues on the remote side
      */
     Map<String, Object> getCurrentProperties() throws RemoteException;
 
     /**
      * Register a subscriber to this transport connector.
      *
      * @param listener the subscriber to register
+     * @throws RemoteException in case of issues on the remote side
      */
     void register(ITransportSubscriber listener) throws RemoteException;
 
     /**
      * Deregister a subscriber from this transport connector.
      *
      * @param listener the subscriber to deregister
+     * @throws RemoteException in case of issues on the remote side
      */
     void deregister(ITransportSubscriber listener) throws RemoteException;
+
+    /**
+     * Mark this transport connector to enable auto-reconnection when the connection is lost.
+     * @param reconnect true to enable reconnection, otherwise false
+     * @throws RemoteException in case of issues on the remote side
+     */
+    void setReconnect(boolean reconnect) throws RemoteException;
+
+    /**
+     * Give back the value of the reconnect flag.
+     *
+     * @return the reconnect flag value
+     * @throws RemoteException in case of issues on the remote side
+     */
+    boolean isReconnect() throws RemoteException;
 }

eu.dariolucia.reatmetric.api/src/main/java/eu/dariolucia/reatmetric/api/transport/TransportStatus.java

@@ -30,13 +30,15 @@ public final class TransportStatus implements Serializable {
     private final AlarmState alarmState;
     private final long txRate;
     private final long rxRate;
+    private final boolean autoReconnect;
 
-    public TransportStatus(String name, TransportConnectionStatus status, long txRate, long rxRate, AlarmState alarmState) {
+    public TransportStatus(String name, TransportConnectionStatus status, long txRate, long rxRate, AlarmState alarmState, boolean autoReconnect) {
         this.name = name;
         this.status = status;
         this.txRate = txRate;
         this.rxRate = rxRate;
         this.alarmState = alarmState;
+        this.autoReconnect = autoReconnect;
     }
 
     /**
@@ -85,6 +87,15 @@ public AlarmState getAlarmState() {
         return alarmState;
     }
 
+    /**
+     * The auto-reconnection of the transport status.
+     *
+     * @return the auto-reconnection flag linked to this connector
+     */
+    public boolean isAutoReconnect() {
+        return autoReconnect;
+    }
+
     @Override
     public String toString() {
         return "TransportStatus{" +
@@ -93,6 +104,7 @@ public String toString() {
                 ", txRate=" + txRate +
                 ", rxRate=" + rxRate +
                 ", alarmState=" + alarmState +
+                ", autoReconnect=" + autoReconnect +
                 '}';
     }
 }

eu.dariolucia.reatmetric.core/src/main/java/eu/dariolucia/reatmetric/core/ReatmetricSystemImpl.java

@@ -40,6 +40,10 @@
 import eu.dariolucia.reatmetric.api.scheduler.IScheduler;
 import eu.dariolucia.reatmetric.api.scheduler.ISchedulerFactory;
 import eu.dariolucia.reatmetric.api.transport.ITransportConnector;
+import eu.dariolucia.reatmetric.api.transport.ITransportSubscriber;
+import eu.dariolucia.reatmetric.api.transport.TransportConnectionStatus;
+import eu.dariolucia.reatmetric.api.transport.TransportStatus;
+import eu.dariolucia.reatmetric.api.transport.exceptions.TransportException;
 import eu.dariolucia.reatmetric.core.api.*;
 import eu.dariolucia.reatmetric.core.api.exceptions.DriverException;
 import eu.dariolucia.reatmetric.core.configuration.DriverConfiguration;
@@ -60,13 +64,14 @@
 /**
  * Reference ReatMetric system implementation.
  */
-public class ReatmetricSystemImpl implements IReatmetricSystem, IServiceCoreContext, IDriverListener {
+public class ReatmetricSystemImpl implements IReatmetricSystem, IServiceCoreContext, IDriverListener, ITransportSubscriber {
 
     private static final Logger LOG = Logger.getLogger(ReatmetricSystemImpl.class.getName());
 
     private final List<IDriver> drivers = new ArrayList<>();
     private final List<ITransportConnector> transportConnectors = new ArrayList<>();
     private final List<ITransportConnector> transportConnectorsImmutable = Collections.unmodifiableList(transportConnectors);
+
     private final Map<Pair<String, String>, IRawDataRenderer> renderers = new HashMap<>();
     private final ServiceCoreConfiguration configuration;
 
@@ -81,6 +86,8 @@ public class ReatmetricSystemImpl implements IReatmetricSystem, IServiceCoreCont
 
     private volatile boolean initialised = false;
 
+    private final Timer executor = new Timer("ReatMetric Core - Executor", true);
+
     public ReatmetricSystemImpl(String initString) throws ReatmetricException {
         try {
             configuration = ServiceCoreConfiguration.load(new FileInputStream(initString));
@@ -159,13 +166,55 @@ public void initialise(Consumer<SystemStatus> consumer) throws ReatmetricExcepti
                 // Ignore for this method
             }
         }
+        // Check the connection status of the connectors (autoconnect, reconnect)
+        initialiseConnectorsProperties();
         // Derive system status
         deriveSystemStatus();
         initialised = true;
         // Done and ready to go
         LOG.info("Reatmetric Core System loaded with status " + systemStatus);
     }
 
+    private void initialiseConnectorsProperties() {
+        List<ITransportConnector> toBeStartedNow = new LinkedList<>();
+        for(ITransportConnector connector : transportConnectors) {
+            // Reconnection flag
+            boolean isReconnectExcluded = configuration.getAutostartConnectors().getReconnectExclusions().stream().anyMatch(o -> {
+                try {
+                    return connector.getName().matches(o);
+                } catch (RemoteException e) {
+                    LOG.log(Level.WARNING, "Cannot retrieve name of transport connector: " + e.getMessage(), e);
+                    return true;
+                }
+            });
+            try {
+                connector.setReconnect(configuration.getAutostartConnectors().isReconnect() && !isReconnectExcluded);
+            } catch (RemoteException e) {
+                LOG.log(Level.WARNING, "Cannot set reconnection status of transport connector: " + e.getMessage(), e);
+            }
+            boolean isAutostartExcluded = configuration.getAutostartConnectors().getStartupExclusions().stream().anyMatch(o -> {
+                try {
+                    return connector.getName().matches(o);
+                } catch (RemoteException e) {
+                    LOG.log(Level.WARNING, "Cannot retrieve name of transport connector: " + e.getMessage(), e);
+                    return true;
+                }
+            });
+            if(configuration.getAutostartConnectors().isStartup() && !isAutostartExcluded) {
+                toBeStartedNow.add(connector);
+            }
+        }
+        // Start now!
+        for(ITransportConnector connector : toBeStartedNow) {
+            try {
+                connector.initialise(connector.getCurrentProperties());
+                connector.connect();
+            } catch (Exception e) { // Protect the startup sequence!
+                LOG.log(Level.WARNING, "Cannot start transport connector: " + e.getMessage(), e);
+            }
+        }
+    }
+
     @Override
     public SystemStatus getStatus() {
         return systemStatus;
@@ -198,6 +247,13 @@ private void deriveSystemStatus() {
     @Override
     public void dispose() throws ReatmetricException {
         initialised = false;
+        for(ITransportConnector tc : transportConnectors) {
+            try {
+                tc.deregister(this);
+            } catch (RemoteException e) {
+                LOG.log(Level.WARNING, "Cannot deregister from transport connector: " + e.getMessage(), e);
+            }
+        }
         for(IDriver d : drivers) {
             d.dispose();
         }
@@ -239,6 +295,13 @@ private void registerActivityHandlers(String driverName, List<IActivityHandler>
 
     private void registerConnectors(List<ITransportConnector> transportConnectors) {
         this.transportConnectors.addAll(transportConnectors);
+        transportConnectors.forEach(o -> {
+            try {
+                o.register(this);
+            } catch (RemoteException e) {
+                LOG.log(Level.WARNING, "Cannot register to transport connector: " + e.getMessage(), e);
+            }
+        });
     }
 
     private IDriver loadDriver(DriverConfiguration dc) throws DriverException {
@@ -285,7 +348,7 @@ public IOperationalMessageProvisionService getOperationalMessageMonitorService()
     }
 
     @Override
-    public IOperationalMessageCollectorService getOperationalMessageCollectorService() throws ReatmetricException, RemoteException {
+    public IOperationalMessageCollectorService getOperationalMessageCollectorService() {
         return messageBroker.getCollectorService();
     }
 
@@ -404,4 +467,31 @@ public List<DebugInformation> currentDebugInfo() {
         //
         return toReturn;
     }
+
+    @Override
+    public void status(TransportStatus status) throws RemoteException {
+        if(status.isAutoReconnect() &&
+                (status.getStatus() == TransportConnectionStatus.ERROR ||
+                 status.getStatus() == TransportConnectionStatus.ABORTED ||
+                 status.getStatus() == TransportConnectionStatus.IDLE)) {
+            // Look for the connector
+            for(ITransportConnector c : transportConnectors) {
+                if(c.getName().equals(status.getName())) {
+                    executor.schedule(new TimerTask() {
+                        @Override
+                        public void run() {
+                            try {
+                                if(c.getConnectionStatus() != TransportConnectionStatus.OPEN && c.getConnectionStatus() != TransportConnectionStatus.CONNECTING) {
+                                    c.connect();
+                                }
+                            } catch (RemoteException | TransportException e) {
+                                LOG.log(Level.WARNING, "Cannot reconnect transport connector: " + e.getMessage(), e);
+                            }
+                        }
+                    }, 500);
+                    return;
+                }
+            }
+        }
+    }
 }