@Override
public void connect(final ClientCallback<ClientConnection> listener, InetSocketAddress bindAddress,final URI uri, final XnioIoThread ioThread, final XnioSsl ssl, final ByteBufferPool bufferPool, final OptionMap options) {
    ChannelListener<StreamConnection> openListener = new ChannelListener<StreamConnection>() {
        @Override
        public void handleEvent(StreamConnection connection) {
            handleConnected(connection, listener, uri, ssl, bufferPool, options);
        }
    };
    IoFuture.Notifier<StreamConnection, Object> notifier = new IoFuture.Notifier<StreamConnection, Object>() {
        @Override
        public void notify(IoFuture<? extends StreamConnection> ioFuture, Object o) {
            if (ioFuture.getStatus() == IoFuture.Status.FAILED) {
                listener.failed(ioFuture.getException());
            }
        }
    };
    if(bindAddress == null) {
        ioThread.openStreamConnection(new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 8009 : uri.getPort()), openListener, options).addNotifier(notifier, null);
    } else {
        ioThread.openStreamConnection(bindAddress, new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 8009 : uri.getPort()), openListener, null, options).addNotifier(notifier, null);
    }
}
 

public IoFuture<ClientConnection> connect(InetSocketAddress bindAddress, final URI uri, final XnioIoThread ioThread, XnioSsl ssl, ByteBufferPool bufferPool, OptionMap options) {
    ClientProvider provider = getClientProvider(uri);
    final FutureResult<ClientConnection> result = new FutureResult<>();
    provider.connect(new ClientCallback<ClientConnection>() {
        @Override
        public void completed(ClientConnection r) {
            result.setResult(r);
        }

        @Override
        public void failed(IOException e) {
            result.setException(e);
        }
    }, bindAddress, uri, ioThread, ssl, bufferPool, options);
    return result.getIoFuture();
}
 

@Override
public void connect(ClientCallback<ClientConnection> listener, InetSocketAddress bindAddress, URI uri, XnioIoThread ioThread, XnioSsl ssl, ByteBufferPool bufferPool, OptionMap options) {
    if (uri.getScheme().equals("https")) {
        if (ssl == null) {
            listener.failed(UndertowMessages.MESSAGES.sslWasNull());
            return;
        }
        OptionMap tlsOptions = OptionMap.builder().addAll(options).set(Options.SSL_STARTTLS, true).getMap();
        if (bindAddress == null) {
            ssl.openSslConnection(ioThread, new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 443 : uri.getPort()), createOpenListener(listener, bufferPool, tlsOptions, uri), tlsOptions).addNotifier(createNotifier(listener), null);
        } else {
            ssl.openSslConnection(ioThread, bindAddress, new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 443 : uri.getPort()), createOpenListener(listener, bufferPool, tlsOptions, uri), tlsOptions).addNotifier(createNotifier(listener), null);
        }
    } else {
        if (bindAddress == null) {
            ioThread.openStreamConnection(new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 80 : uri.getPort()), createOpenListener(listener, bufferPool, options, uri), options).addNotifier(createNotifier(listener), null);
        } else {
            ioThread.openStreamConnection(bindAddress, new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 80 : uri.getPort()), createOpenListener(listener, bufferPool, options, uri), null, options).addNotifier(createNotifier(listener), null);
        }
    }
}
 

/**
 * Gets the host data for this thread
 *
 * @return The data for this thread
 */
private HostThreadData getData() {
    Thread thread = Thread.currentThread();
    if (!(thread instanceof XnioIoThread)) {
        throw UndertowMessages.MESSAGES.canOnlyBeCalledByIoThread();
    }
    XnioIoThread ioThread = (XnioIoThread) thread;
    HostThreadData data = hostThreadData.get(ioThread);
    if (data != null) {
        return data;
    }
    data = new HostThreadData();
    HostThreadData existing = hostThreadData.putIfAbsent(ioThread, data);
    if (existing != null) {
        return existing;
    }
    return data;
}
 

/**
 * Should only be used for tests.
 *
 */
void closeCurrentConnections() {
    final CountDownLatch latch = new CountDownLatch(hostThreadData.size());
    for(final Map.Entry<XnioIoThread, HostThreadData> data : hostThreadData.entrySet()) {
        data.getKey().execute(new Runnable() {
            @Override
            public void run() {
                ConnectionHolder d = data.getValue().availableConnections.poll();
                while (d != null) {
                    IoUtils.safeClose(d.clientConnection);
                    d = data.getValue().availableConnections.poll();
                }
                data.getValue().connections = 0;
                latch.countDown();
            }
        });
    }
    try {
        latch.await(10, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}
 

/**
 * Schedules a task for future execution. If the execution is rejected because the worker is shutting
 * down then it is logged at debug level and the exception is not re-thrown
 *  @param thread   The IO thread
 * @param task     The task to execute
 * @param timeout  The timeout
 * @param timeUnit The time unit
 */
public static XnioExecutor.Key executeAfter(XnioIoThread thread, Runnable task, long timeout, TimeUnit timeUnit) {
    try {
        return thread.executeAfter(task, timeout, timeUnit);
    } catch (RejectedExecutionException e) {
        if(thread.getWorker().isShutdown()) {
            UndertowLogger.ROOT_LOGGER.debugf(e, "Failed to schedule task %s as worker is shutting down", task);
            //we just return a bogus key in this case
            return new XnioExecutor.Key() {
                @Override
                public boolean remove() {
                    return false;
                }
            };
        } else {
            throw e;
        }
    }
}
 

@Override
public void connect(ClientCallback<ClientConnection> listener, InetSocketAddress bindAddress, URI uri, XnioIoThread ioThread, XnioSsl ssl, ByteBufferPool bufferPool, OptionMap options) {
    if (uri.getScheme().equals(HTTPS)) {
        if (ssl == null) {
            listener.failed(UndertowMessages.MESSAGES.sslWasNull());
            return;
        }
        OptionMap tlsOptions = OptionMap.builder().addAll(options).set(Options.SSL_STARTTLS, true).getMap();
        if (bindAddress == null) {
            ssl.openSslConnection(ioThread, new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 443 : uri.getPort()), createOpenListener(listener, bufferPool, tlsOptions, uri), tlsOptions).addNotifier(createNotifier(listener), null);
        } else {
            ssl.openSslConnection(ioThread, bindAddress, new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 443 : uri.getPort()), createOpenListener(listener, bufferPool, tlsOptions, uri), tlsOptions).addNotifier(createNotifier(listener), null);
        }
    } else {
        if (bindAddress == null) {
            ioThread.openStreamConnection(new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 80 : uri.getPort()), createOpenListener(listener, bufferPool, options, uri), options).addNotifier(createNotifier(listener), null);
        } else {
            ioThread.openStreamConnection(bindAddress, new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 80 : uri.getPort()), createOpenListener(listener, bufferPool, options, uri), null, options).addNotifier(createNotifier(listener), null);
        }
    }
}
 

@Override
public void handleRequest(HttpServerExchange serverExchange) throws Exception {
    AuthContext context = AuthContext.initialize(serverExchange);
    serverExchange.putAttachment(AUTH_CONTEXT_KEY, context);
    // Make sure the exchange attachment is removed in the end
    serverExchange.addExchangeCompleteListener((exchange, nextListener) -> {
        exchange.removeAttachment(AUTH_CONTEXT_KEY);
        nextListener.proceed();
    });
    if (context.isMissingTenantHeader()) {
        endExchange(serverExchange, BAD_REQUEST, MISSING_HEADERS_MSG);
        return;
    }

    // Marks the request as dispatched. If we don't do this, the exchange will be terminated by the container when
    // this method returns, but we need to wait for Kubernetes' master response.
    serverExchange.dispatch();
    XnioIoThread ioThread = serverExchange.getIoThread();
    ConnectionPool connectionPool = connectionPools.computeIfAbsent(ioThread, t -> new ConnectionPool(connectionFactory, componentName));
    PooledConnectionWaiter waiter = createWaiter(serverExchange);
    if (!connectionPool.offer(waiter)) {
        endExchange(serverExchange, INTERNAL_SERVER_ERROR, TOO_MANY_PENDING_REQUESTS);
    }
}
 

private void onConnectionCreationFailure(Exception e) {
    log.debug("Failed to create client connection", e);
    if (stop) {
        return;
    }
    // Wait a bit before trying to create a connection again
    XnioIoThread ioThread = (XnioIoThread) Thread.currentThread();
    ioThread.executeAfter(() -> {
        removeTimedOutWaiters();
        if (!stop && !waiters.isEmpty() && !isFull()) {
            // It's still necessary to create a connection only if the pool is not stopped, there still is
            // a client waiting for a connection and the pool is not full
            createConnection();
        }
    }, 1, SECONDS);
}
 

private void closeAllConnections(Runnable onAllClosed) {
    for (Iterator<PooledConnection> iterator = connections.iterator(); iterator.hasNext(); ) {
        PooledConnection connection = iterator.next();
        // Don't close a connection if it's still used
        if (connection.idle) {
            iterator.remove();
            IoUtils.safeClose(connection);
        }
    }
    // Is there any connection still used?
    if (connections.isEmpty()) {
        // No, invoked the stop callback
        onAllClosed.run();
    } else {
        // Yes, wait a bit and try close idle conections again
        XnioIoThread ioThread = (XnioIoThread) Thread.currentThread();
        ioThread.executeAfter(() -> closeAllConnections(onAllClosed), 500, MILLISECONDS);
    }
}
 

@Override
public void connect(final ClientCallback<ClientConnection> listener, final InetSocketAddress bindAddress, final URI uri, final XnioIoThread ioThread, final XnioSsl ssl, final ByteBufferPool bufferPool, final OptionMap options) {

    if (bindAddress == null) {
        ioThread.openStreamConnection(new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 80 : uri.getPort()), createOpenListener(listener, bufferPool, options, uri.getHost()), options).addNotifier(createNotifier(listener), null);
    } else {
        ioThread.openStreamConnection(bindAddress, new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 80 : uri.getPort()), createOpenListener(listener, bufferPool, options, uri.getHost()), null, options).addNotifier(createNotifier(listener), null);
    }
}
 

@Override
public void connect(final ClientCallback<ClientConnection> listener, InetSocketAddress bindAddress, final URI uri, final XnioIoThread ioThread, final XnioSsl ssl, final ByteBufferPool bufferPool, final OptionMap options) {
    if (ssl == null) {
        listener.failed(UndertowMessages.MESSAGES.sslWasNull());
        return;
    }
    if(bindAddress == null) {
        OptionMap tlsOptions = OptionMap.builder().addAll(options).set(Options.SSL_STARTTLS, true).getMap();
        ssl.openSslConnection(ioThread, new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 443 : uri.getPort()), createOpenListener(listener, uri, ssl, bufferPool, tlsOptions), options).addNotifier(createNotifier(listener), null);
    } else {
        ssl.openSslConnection(ioThread, bindAddress, new InetSocketAddress(uri.getHost(), uri.getPort() == -1 ? 443 : uri.getPort()), createOpenListener(listener, uri, ssl, bufferPool, options), options).addNotifier(createNotifier(listener), null);
    }

}
 

@Override
public XnioIoThread getIoThread() {
    if(channel == null) {
        return null;
    }
    return channel.getIoThread();
}
 

private SessionImpl(final InMemorySessionManager sessionManager, final String sessionId, final SessionConfig sessionCookieConfig, final XnioIoThread executor, final XnioWorker worker, final Object evictionToken, final int maxInactiveInterval) {
    this.sessionManager = sessionManager;
    this.sessionId = sessionId;
    this.sessionCookieConfig = sessionCookieConfig;
    this.executor = executor;
    this.worker = worker;
    this.evictionToken = evictionToken;
    creationTime = lastAccessed = System.currentTimeMillis();
    this.maxInactiveInterval = maxInactiveInterval;
}
 

/**
 * Register a new node.
 *
 * @param config         the node configuration
 * @param balancerConfig the balancer configuration
 * @param ioThread       the associated I/O thread
 * @param bufferPool     the buffer pool
 * @return whether the node could be created or not
 */
public synchronized boolean addNode(final NodeConfig config, final Balancer.BalancerBuilder balancerConfig, final XnioIoThread ioThread, final ByteBufferPool bufferPool) {

    final String jvmRoute = config.getJvmRoute();
    final Node existing = nodes.get(jvmRoute);
    if (existing != null) {
        if (config.getConnectionURI().equals(existing.getNodeConfig().getConnectionURI())) {
            // TODO better check if they are the same
            existing.resetState();
            return true;
        } else {
            existing.markRemoved();
            removeNode(existing);
            if (!existing.isInErrorState()) {
                return false; // replies with MNODERM error
            }
        }
    }

    final String balancerRef = config.getBalancer();
    Balancer balancer = balancers.get(balancerRef);
    if (balancer != null) {
        UndertowLogger.ROOT_LOGGER.debugf("Balancer %s already exists, replacing", balancerRef);
    }
    balancer = balancerConfig.build();
    balancers.put(balancerRef, balancer);

    final Node node = new Node(config, balancer, ioThread, bufferPool, this);
    nodes.put(jvmRoute, node);
    // Schedule the health check
    scheduleHealthCheck(node, ioThread);
    // Reset the load factor periodically
    if (updateLoadTask.cancelKey == null) {
        updateLoadTask.cancelKey = ioThread.executeAtInterval(updateLoadTask, modCluster.getHealthCheckInterval(), TimeUnit.MILLISECONDS);
    }
    // Remove from the failover groups
    failoverDomains.remove(node.getJvmRoute());
    UndertowLogger.ROOT_LOGGER.registeringNode(jvmRoute, config.getConnectionURI());
    return true;
}
 

void scheduleHealthCheck(final Node node, XnioIoThread ioThread) {
    assert Thread.holdsLock(this);
    HealthCheckTask task = healthChecks.get(ioThread);
    if (task == null) {
        task = new HealthCheckTask(removeBrokenNodesThreshold, healthChecker);
        healthChecks.put(ioThread, task);
        task.cancelKey = ioThread.executeAtInterval(task, modCluster.getHealthCheckInterval(), TimeUnit.MILLISECONDS);
    }
    task.nodes.add(node);
}
 

void removeHealthCheck(final Node node, XnioIoThread ioThread) {
    assert Thread.holdsLock(this);
    final HealthCheckTask task = healthChecks.get(ioThread);
    if (task == null) {
        return;
    }
    task.nodes.remove(node);
    if (task.nodes.size() == 0) {
        healthChecks.remove(ioThread);
        task.cancelKey.remove();
    }
}
 

/**
 * Try to open a socket connection to given address.
 *
 * @param address     the socket address
 * @param exchange    the http servers exchange
 * @param callback    the ping callback
 * @param options     the options
 */
static void pingHost(InetSocketAddress address, HttpServerExchange exchange, PingCallback callback, OptionMap options) {

    final XnioIoThread thread = exchange.getIoThread();
    final XnioWorker worker = thread.getWorker();
    final HostPingTask r = new HostPingTask(address, worker, callback, options);
    // Schedule timeout task
    scheduleCancelTask(exchange.getIoThread(), r, 5, TimeUnit.SECONDS);
    exchange.dispatch(exchange.isInIoThread() ? SameThreadExecutor.INSTANCE : thread, r);
}
 

HttpClientPingTask(URI connection, RequestExchangeListener exchangeListener, XnioIoThread thread, UndertowClient client, XnioSsl xnioSsl, ByteBufferPool bufferPool, OptionMap options) {
    this.connection = connection;
    this.thread = thread;
    this.client = client;
    this.xnioSsl = xnioSsl;
    this.bufferPool = bufferPool;
    this.options = options;
    this.exchangeListener = exchangeListener;
}
 

static void scheduleCancelTask(final XnioIoThread ioThread, final CancellableTask cancellable, final long timeout, final TimeUnit timeUnit ) {
    final XnioExecutor.Key key = WorkerUtils.executeAfter(ioThread, new Runnable() {
        @Override
        public void run() {
            cancellable.cancel();
        }
    }, timeout, timeUnit);
    cancellable.setCancelKey(key);
}
 

protected Node(NodeConfig nodeConfig, Balancer balancerConfig, XnioIoThread ioThread, ByteBufferPool bufferPool, ModClusterContainer container) {
    this.id = idGen.incrementAndGet();
    this.jvmRoute = nodeConfig.getJvmRoute();
    this.nodeConfig = nodeConfig;
    this.ioThread = ioThread;
    this.bufferPool = bufferPool;
    this.balancerConfig = balancerConfig;
    this.container = container;
    this.connectionPoolManager = new NodeConnectionPoolManager();
    this.connectionPool = new ProxyConnectionPool(connectionPoolManager, nodeConfig.getConnectionURI(), container.getXnioSsl(), container.getClient(), container.getClientOptions());
}
 

@Override
public void stop() {
    Set<Entry<XnioIoThread, ConnectionPool>> entries = connectionPools.entrySet();
    CountDownLatch latch = new CountDownLatch(entries.size());
    entries.forEach(entry -> {
        // Connection pool is not thread safe and #stop must be called on the corresponding io thread
        entry.getKey().execute(() -> entry.getValue().stop(latch::countDown));
    });
    Uninterruptibles.awaitUninterruptibly(latch, 5, SECONDS);
    connectionFactory.close();
}
 

private ConnectionPool(ConnectionFactory connectionFactory, String componentName) {
    this.connectionFactory = connectionFactory;
    connections = new ArrayList<>(MAX_CONNECTIONS_PER_THREAD);
    waiters = new ArrayDeque<>();
    XnioIoThread ioThread = (XnioIoThread) Thread.currentThread();
    periodicTaskKey = ioThread.executeAtInterval(this::periodicTask, 1, SECONDS);
    ongoingCreations = 0;
    stop = false;
}
 

private boolean validateNonceWithCount(Nonce nonce, int nonceCount, final XnioIoThread executor) {
    // This point could have been reached either because the knownNonces map contained the key or because
    // it didn't and a count was supplied - either way need to double check the contents of knownNonces once
    // the lock is in place.
    synchronized (knownNonces) {
        Nonce value = knownNonces.get(nonce.nonce);
        long now = System.currentTimeMillis();
        // For the purpose of this validation we also add the cacheTimePostExpiry - when nextNonce is subsequently
        // called it will decide if we are in the interval to replace the nonce.
        long earliestAccepted = now - (overallTimeOut + cacheTimePostExpiry);
        if (value == null) {
            if (nonce.timeStamp < 0) {
                // Means it was in there, now it isn't - most likely a timestamp expiration mid check - abandon validation.
                return false;
            }

            if (nonce.timeStamp > earliestAccepted && nonce.timeStamp <= now) {
                knownNonces.put(nonce.nonce, nonce);
                long timeTillExpiry = nonce.timeStamp - earliestAccepted;
                nonce.executorKey = WorkerUtils.executeAfter(executor, new KnownNonceCleaner(nonce.nonce), timeTillExpiry,
                        TimeUnit.MILLISECONDS);
                return true;
            }

            return false;
        } else {
            // We have it, just need to verify that it has not expired and that the nonce key is valid.
            if (value.timeStamp < earliestAccepted || value.timeStamp > now) {
                // The embedded timestamp is either expired or somehow is after now!!
                return false;
            }

            if (value.getMaxNonceCount() < nonceCount) {
                value.setMaxNonceCount(nonceCount);
                return true;
            }

            return false;
        }

    }

}
 

@Override
public XnioIoThread getIoThread() {
    return null;
}
 

public XnioIoThread getIoThread() {
    return tcpServer.getIoThread();
}
 

@Override
public IoFuture<SslConnection> openSslConnection(final XnioIoThread ioThread, final InetSocketAddress bindAddress, final InetSocketAddress destination, final ChannelListener<? super SslConnection> openListener, final ChannelListener<? super BoundChannel> bindListener, final OptionMap optionMap) {
    final FutureResult<SslConnection> futureResult = new FutureResult<>(ioThread);
    final IoFuture<StreamConnection> connection = ioThread.openStreamConnection(bindAddress, destination, new StreamConnectionChannelListener(optionMap, destination, futureResult, openListener), bindListener, optionMap);
    return setupSslConnection(futureResult, connection);
}
 

@Override
public XnioIoThread getReadThread() {
    return delegate.getIoThread();
}
 

@Override
public XnioIoThread getWriteThread() {
    return delegate.getIoThread();
}
 

@Override
public void connect(final ClientCallback<ClientConnection> listener, final URI uri, final XnioIoThread ioThread, final XnioSsl ssl, final ByteBufferPool bufferPool, final OptionMap options) {
    connect(listener, null, uri, ioThread, ssl, bufferPool, options);
}