package com.digitalpetri.opcua.stack.client.fsm.states;

import java.util.concurrent.CompletableFuture;

import com.digitalpetri.opcua.stack.client.UaTcpStackClient;
import com.digitalpetri.opcua.stack.client.fsm.ConnectionEvent;
import com.digitalpetri.opcua.stack.client.fsm.ConnectionState;
import com.digitalpetri.opcua.stack.client.fsm.ConnectionStateFsm;
import com.digitalpetri.opcua.stack.core.StatusCodes;
import com.digitalpetri.opcua.stack.core.UaException;
import com.digitalpetri.opcua.stack.core.channel.ClientSecureChannel;
import com.digitalpetri.opcua.stack.core.types.builtin.StatusCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Reconnecting implements ConnectionState {

    private static final int MAX_RECONNECT_DELAY_SECONDS = 16;

    private final Logger logger = LoggerFactory.getLogger(getClass());

    private final CompletableFuture<ClientSecureChannel> channelFuture = new CompletableFuture<>();

    private volatile ClientSecureChannel secureChannel;

    private final long delaySeconds;
    private volatile long secureChannelId;

    public Reconnecting(long delaySeconds, long secureChannelId) {
        this.delaySeconds = delaySeconds;
        this.secureChannelId = secureChannelId;
    }

    @Override
    public CompletableFuture<Void> activate(ConnectionEvent event, ConnectionStateFsm fsm) {
        CompletableFuture<Void> future = new CompletableFuture<>();

        connect(fsm, true, new CompletableFuture<>()).whenComplete((sc, ex) -> {
            if (sc != null) {
                secureChannel = sc;
                fsm.handleEvent(ConnectionEvent.RECONNECT_SUCCEEDED);
            } else {
                channelFuture.completeExceptionally(ex);
                fsm.handleEvent(ConnectionEvent.RECONNECT_FAILED);
            }

            future.complete(null);
        });

        return future;
    }

    private CompletableFuture<ClientSecureChannel> connect(ConnectionStateFsm fsm,
                                                           boolean initialAttempt,
                                                           CompletableFuture<ClientSecureChannel> future) {

        logger.debug("Reconnecting...");

        UaTcpStackClient.bootstrap(fsm.getClient(), secureChannelId).whenComplete((sc, ex) -> {
            if (sc != null) {
                logger.debug("Channel bootstrap succeeded: localAddress={}, remoteAddress={}",
                        sc.getChannel().localAddress(), sc.getChannel().remoteAddress());

                future.complete(sc);
            } else {
                logger.debug("Channel bootstrap failed: {}", ex.getMessage(), ex);

                StatusCode statusCode = UaException.extract(ex)
                        .map(UaException::getStatusCode)
                        .orElse(StatusCode.BAD);

                boolean secureChannelError =
                        statusCode.getValue() == StatusCodes.Bad_TcpSecureChannelUnknown ||
                                statusCode.getValue() == StatusCodes.Bad_SecureChannelClosed ||
                                statusCode.getValue() == StatusCodes.Bad_SecureChannelIdInvalid;

                if (initialAttempt && secureChannelError) {
                    // Try again if bootstrapping failed because we couldn't re-open the previous channel.
                    logger.debug("Previous channel unusable, retrying...");

                    secureChannelId = 0L;

                    connect(fsm, false, future);
                } else {
                    future.completeExceptionally(ex);
                }
            }
        });

        return future;
    }

    @Override
    public CompletableFuture<Void> deactivate(ConnectionEvent event, ConnectionStateFsm fsm) {
        return CF_VOID_COMPLETED;
    }

    @Override
    public ConnectionState transition(ConnectionEvent event, ConnectionStateFsm fsm) {
        switch (event) {
            case RECONNECT_FAILED:
                return new Reconnecting(nextDelay(), secureChannelId);

            case RECONNECT_SUCCEEDED:
                return new Connected(secureChannel, channelFuture);
        }

        return this;
    }

    @Override
    public CompletableFuture<ClientSecureChannel> getSecureChannel() {
        return channelFuture;
    }

    private long nextDelay() {
        if (delaySeconds == 0) {
            return 1;
        } else {
            return Math.min(delaySeconds << 1, MAX_RECONNECT_DELAY_SECONDS);
        }
    }

    @Override
    public String toString() {
        return "Reconnecting{" +
                "delaySeconds=" + delaySeconds +
                ", secureChannelId=" + secureChannelId +
                '}';
    }

}

@@ -96,16 +96,99 @@
 eFuture;
+%0Aimport java.util.concurrent.ScheduledFuture;%0Aimport java.util.concurrent.TimeUnit;
 %0A%0Aimport
@@ -437,16 +437,64 @@
 ateFsm;%0A
+import com.digitalpetri.opcua.stack.core.Stack;%0A
 import c
@@ -1077,24 +1077,82 @@
 uture%3C%3E();%0A%0A
+    private volatile ScheduledFuture%3C?%3E scheduledFuture;%0A%0A
     private 
@@ -1620,24 +1620,49 @@
 ();%0A%0A       
+ Runnable connect = () -%3E
  connect(fsm
@@ -2054,16 +2054,16 @@
 (null);%0A
-
         
@@ -2059,32 +2059,331 @@
 );%0A        %7D);%0A%0A
+        if (scheduledFuture == null %7C%7C (scheduledFuture != null && scheduledFuture.cancel(false))) %7B%0A            logger.debug(%22Reactivating in %7B%7D seconds...%22, delaySeconds);%0A%0A            scheduledFuture = Stack.sharedScheduledExecutor().schedule(connect, delaySeconds, TimeUnit.SECONDS);%0A        %7D%0A%0A
         return f

Use the reconnect delay