/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.tinkerpop.gremlin.driver;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;

/**
 * Identifies a server within the {@link Cluster} at a specific address.
 *
 * @author Stephen Mallette (http://stephen.genoprime.com)
 */
public final class Host {
    private static final Logger logger = LoggerFactory.getLogger(Host.class);
    private final InetSocketAddress address;
    private final URI hostUri;
    private volatile boolean isAvailable;
    private final Cluster cluster;
    private final String hostLabel;

    final AtomicReference<ScheduledFuture<?>> reconnectionAttempt = new AtomicReference<>(null);

    Host(final InetSocketAddress address, final Cluster cluster) {
        this.cluster = cluster;
        this.address = address;
        this.hostUri = makeUriFromAddress(address, cluster.connectionPoolSettings().enableSsl);
        hostLabel = String.format("Host{address=%s, hostUri=%s}", address, hostUri);
    }

    public InetSocketAddress getAddress() {
        return address;
    }

    public URI getHostUri() {
        return hostUri;
    }

    public boolean isAvailable() {
        return isAvailable;
    }

    void makeAvailable() {
        isAvailable = true;
    }

    void makeUnavailable(final Function<Host, Boolean> reconnect) {
        isAvailable = false;

        // only do a connection re-attempt if one is not already in progress
        synchronized (reconnectionAttempt) {
            if (reconnectionAttempt.get() == null) {
                reconnectionAttempt.set(this.cluster.executor().scheduleAtFixedRate(() -> {
                    logger.debug("Trying to reconnect to dead host at {}", this);
                    if (reconnect.apply(this))
                        reconnected();
                }, cluster.connectionPoolSettings().reconnectInitialDelay, cluster.connectionPoolSettings().reconnectInterval, TimeUnit.MILLISECONDS));
            }
        }
    }

    private void reconnected() {
        reconnectionAttempt.get().cancel(false);
        reconnectionAttempt.set(null);
        makeAvailable();
    }

    private static URI makeUriFromAddress(final InetSocketAddress addy, final boolean ssl) {
        try {
            final String scheme = ssl ? "wss" : "ws";
            return new URI(scheme, null, addy.getHostName(), addy.getPort(), "/gremlin", null, null);
        } catch (URISyntaxException use) {
            throw new RuntimeException(String.format("URI for host could not be constructed from: %s", addy), use);
        }
    }

    @Override
    public String toString() {
        return hostLabel;
    }

    public static interface Listener {
        public void onAvailable(final Host host);

        public void onUnavailable(final Host host);

        public void onNew(final Host host);

        public void onRemove(final Host host);
    }
}

@@ -1634,77 +1634,114 @@
 nce%3C
-ScheduledFuture%3C?%3E%3E reconnectionAttempt = new AtomicReference%3C%3E(
+Boolean%3E retryInProgress = new AtomicReference%3C%3E(Boolean.FALSE);%0A    ScheduledFuture%3C?%3E retryThread = 
 null
-)
 ;%0A%0A 
@@ -2513,138 +2513,100 @@
     
-synchronized (reconnectionAttempt) %7B%0A            if (reconnectionAttempt.get() == null) %7B%0A                reconnectionAttempt.set(
+if (retryInProgress.compareAndSet(Boolean.FALSE, Boolean.TRUE)) %7B%0A            retryThread = 
 this
@@ -2653,16 +2653,70 @@
 () -%3E %7B%0A
+                                                      
         
@@ -2785,24 +2785,78 @@
 %7B%7D%22, this);%0A
+                                                      
             
@@ -2889,16 +2889,31 @@
 y(this))
+ reconnected();
 %0A       
@@ -2929,31 +2929,46 @@
         
-reconnected();%0A
+                              
         
@@ -3033,16 +3033,86 @@
 alDelay,
+%0A                                                                     
  cluster
@@ -3182,25 +3182,10 @@
 NDS)
-)
 ;%0A
-            %7D%0A
     
@@ -3242,85 +3242,306 @@
     
-reconnectionAttempt.get().cancel(false);%0A        reconnectionAttempt.set(null
+// race condition!  retry boolean could be set to false, a new retryThread created above%0A        // and then cancelled here.   But we're only executing this at all because we *have* reconnected%0A        retryThread.cancel(false);%0A        retryThread = null;%0A        retryInProgress.set(Boolean.FALSE
 );%0A 

Use an AtomicReference<Boolean> to protect the reconnect task, rather than synchronizing on it.