/*
 * ====================================================================
 * 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.
 * ====================================================================
 *
 * This software consists of voluntary contributions made by many
 * individuals on behalf of the Apache Software Foundation.  For more
 * information on the Apache Software Foundation, please see
 * <http://www.apache.org/>.
 *
 */
package org.apache.hc.core5.pool;

import java.io.IOException;
import java.time.Clock;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.hc.core5.annotation.Contract;
import org.apache.hc.core5.annotation.Experimental;
import org.apache.hc.core5.annotation.ThreadingBehavior;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.io.CloseMode;
import org.apache.hc.core5.io.ModalCloseable;
import org.apache.hc.core5.util.Args;
import org.apache.hc.core5.util.TimeValue;
import org.apache.hc.core5.util.Timeout;

/**
 * Route-segmented connection pool with reduced contention and a small, conditional
 * round-robin helper for busy multi-route scenarios.
 *
 * <p>Per-route state is kept in independent segments. Disposal of connections is offloaded
 * to a bounded executor so slow graceful closes do not block threads leasing on other routes.
 * A minimal round-robin helper is engaged only when there are many pending routes and
 * there is global headroom; it never scans all routes.</p>
 *
 * @param <R> route key type
 * @param <C> connection type (must be {@link ModalCloseable})
 * @see ManagedConnPool
 * @see PoolReusePolicy
 * @see DisposalCallback
 * @since 5.4
 */
@Contract(threading = ThreadingBehavior.SAFE_CONDITIONAL)
@Experimental
public final class RouteSegmentedConnPool<R, C extends ModalCloseable> implements ManagedConnPool<R, C> {

    // Engage RR helper only when there are enough pending routes to benefit.
    private static final int RR_MIN_PENDING_ROUTES = 8;
    private static final int RR_BUDGET = 64;
    private static final int RR_INLINE_FALLBACK_BUDGET = 8;

    private final PoolReusePolicy reusePolicy;
    private final TimeValue timeToLive;
    private final long ttlMillis;
    private final boolean hasTtl;
    private final DisposalCallback<C> disposal;

    private final ConnPoolListener<R> connPoolListener;

    private final AtomicInteger defaultMaxPerRoute = new AtomicInteger(5);

    private final ConcurrentHashMap<R, Segment> segments = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<R, Integer> maxPerRoute = new ConcurrentHashMap<>();
    private final AtomicInteger totalAllocated = new AtomicInteger(0);
    private final AtomicInteger maxTotal = new AtomicInteger(25);

    private final AtomicBoolean closed = new AtomicBoolean(false);

    private final ScheduledThreadPoolExecutor timeouts;

    private final Clock clock;

    /**
     * Dedicated executor for asynchronous, best-effort disposal.
     * Bounded queue; on saturation we fall back to IMMEDIATE close on the caller thread.
     */
    private final ThreadPoolExecutor disposer;

    // Minimal fair round-robin over routes with waiters (no global scans).
    private final ConcurrentLinkedQueue<R> pendingQueue = new ConcurrentLinkedQueue<>();
    private final AtomicBoolean draining = new AtomicBoolean(false);
    private final AtomicInteger pendingRouteCount = new AtomicInteger(0);

    public RouteSegmentedConnPool(
            final int defaultMaxPerRoute,
            final int maxTotal,
            final TimeValue timeToLive,
            final PoolReusePolicy reusePolicy,
            final DisposalCallback<C> disposal) {
        this(defaultMaxPerRoute, maxTotal, timeToLive, reusePolicy, disposal, null);
    }

    public RouteSegmentedConnPool(
            final int defaultMaxPerRoute,
            final int maxTotal,
            final TimeValue timeToLive,
            final PoolReusePolicy reusePolicy,
            final DisposalCallback<C> disposal,
            final ConnPoolListener<R> connPoolListener) {
        this(defaultMaxPerRoute, maxTotal, timeToLive, reusePolicy, disposal, connPoolListener, Clock.systemUTC());
    }

    RouteSegmentedConnPool(
            final int defaultMaxPerRoute,
            final int maxTotal,
            final TimeValue timeToLive,
            final PoolReusePolicy reusePolicy,
            final DisposalCallback<C> disposal,
            final ConnPoolListener<R> connPoolListener,
            final Clock clock) {

        this.defaultMaxPerRoute.set(defaultMaxPerRoute > 0 ? defaultMaxPerRoute : 5);
        this.maxTotal.set(maxTotal > 0 ? maxTotal : 25);
        this.timeToLive = timeToLive != null ? timeToLive : TimeValue.NEG_ONE_MILLISECOND;

        if (this.timeToLive.getDuration() >= 0) {
            this.ttlMillis = this.timeToLive.toMilliseconds();
            this.hasTtl = true;
        } else {
            this.ttlMillis = -1L;
            this.hasTtl = false;
        }

        this.reusePolicy = reusePolicy != null ? reusePolicy : PoolReusePolicy.LIFO;
        this.disposal = Args.notNull(disposal, "disposal");
        this.connPoolListener = connPoolListener;

        this.clock = Args.notNull(clock, "clock");

        final ThreadFactory tf = r -> {
            final Thread t = new Thread(r, "seg-pool-timeouts");
            t.setDaemon(true);
            return t;
        };
        this.timeouts = new ScheduledThreadPoolExecutor(1, tf);
        this.timeouts.setRemoveOnCancelPolicy(true);

        // Asynchronous disposer for slow GRACEFUL closes. Capped at 32 so very
        // many-core boxes do not over-provision; threads time out when idle.
        final int cores = Math.max(2, Runtime.getRuntime().availableProcessors());
        final int nThreads = Math.min(32, Math.max(8, cores * 2));
        final int qsize = 4096;

        final ThreadFactory df = r -> {
            final Thread t = new Thread(r, "seg-pool-disposer");
            t.setDaemon(true);
            return t;
        };
        this.disposer = new ThreadPoolExecutor(
                nThreads, nThreads,
                30_000L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(qsize),
                df,
                new ThreadPoolExecutor.AbortPolicy());
        this.disposer.allowCoreThreadTimeOut(true);
    }

    final class Segment {
        final ConcurrentLinkedDeque<PoolEntry<R, C>> available = new ConcurrentLinkedDeque<>();
        final ConcurrentLinkedDeque<Waiter> waiters = new ConcurrentLinkedDeque<>();
        final AtomicInteger allocated = new AtomicInteger(0);
        final AtomicBoolean enqueued = new AtomicBoolean(false);

        int limitPerRoute(final R route) {
            final Integer v = maxPerRoute.get(route);
            return v != null ? v : defaultMaxPerRoute.get();
        }
    }

    final class Waiter extends CompletableFuture<PoolEntry<R, C>> {
        final R route;
        final Segment seg;
        final Timeout requestTimeout;
        final Object state;
        volatile boolean cancelled;
        volatile ScheduledFuture<?> timeoutTask;

        Waiter(final R route, final Segment seg, final Timeout t, final Object s) {
            this.route = route;
            this.seg = seg;
            this.requestTimeout = t != null ? t : Timeout.DISABLED;
            this.state = s;
            this.cancelled = false;
            this.timeoutTask = null;
        }

        @Override
        public boolean cancel(final boolean mayInterruptIfRunning) {
            this.cancelled = true;
            cancelTimeout(this);
            this.seg.waiters.remove(this);
            dequeueIfDrained(this.seg);
            maybeCleanupSegment(this.route, this.seg);
            return super.cancel(mayInterruptIfRunning);
        }

        @Override
        public PoolEntry<R, C> get(final long timeout, final TimeUnit unit)
                throws InterruptedException, java.util.concurrent.ExecutionException, TimeoutException {
            try {
                return super.get(timeout, unit);
            } catch (final TimeoutException ex) {
                if (cancel(true)) {
                    throw ex;
                }
                return super.get();
            }
        }
    }

    @Override
    public Future<PoolEntry<R, C>> lease(
            final R route,
            final Object state,
            final Timeout requestTimeout,
            final FutureCallback<PoolEntry<R, C>> callback) {

        ensureOpen();
        final Segment seg = segments.computeIfAbsent(route, r -> new Segment());

        // 1) Try available
        PoolEntry<R, C> hit;
        for (; ; ) {
            hit = pollAvailable(seg, state);
            if (hit == null) {
                break;
            }
            final long now = clock.millis();
            if (hit.getExpiryDeadline().isBefore(now) || isPastTtl(hit, now)) {
                discardAndDecr(hit, CloseMode.GRACEFUL);
                continue;
            }
            break;
        }
        if (hit != null) {
            fireOnLease(route);
            if (callback != null) {
                callback.completed(hit);
            }
            return CompletableFuture.completedFuture(hit);
        }

        // 2) Try to allocate new within caps
        if (tryAllocateOne(route, seg)) {
            final PoolEntry<R, C> entry = new PoolEntry<>(route, timeToLive, disposal, clock);
            fireOnLease(route);
            if (callback != null) {
                callback.completed(entry);
            }
            return CompletableFuture.completedFuture(entry);
        }

        // 3) Enqueue waiter with timeout
        final Waiter w = new Waiter(route, seg, requestTimeout, state);
        seg.waiters.addLast(w);
        enqueueIfNeeded(route, seg);

        // Late hit after enqueuing
        final PoolEntry<R, C> late = pollAvailable(seg, state);
        if (late != null) {
            if (w.complete(late)) {
                cancelTimeout(w);
                fireOnLease(route);
                if (callback != null) {
                    callback.completed(late);
                }
                dequeueIfDrained(seg);
                return w;
            } else {
                boolean handedOff = false;
                for (Waiter other; (other = seg.waiters.pollFirst()) != null; ) {
                    if (other.cancelled || other.isDone()) {
                        continue;
                    }
                    if (compatible(other.state, late.getState())) {
                        cancelTimeout(other);
                        handedOff = other.complete(late);
                        if (handedOff) {
                            fireOnLease(other.route);
                            break;
                        }
                    }
                }
                if (!handedOff) {
                    offerAvailable(seg, late);
                }
            }
        }

        scheduleTimeout(w, seg);

        if (callback != null) {
            w.whenComplete((pe, ex) -> {
                if (ex != null) {
                    final Exception failure;
                    if (ex instanceof Exception) {
                        failure = (Exception) ex;
                    } else {
                        failure = new Exception(ex);
                    }
                    callback.failed(failure);
                } else {
                    callback.completed(pe);
                }
            });
        }

        triggerDrainIfMany();
        return w;
    }

    @Override
    public void release(final PoolEntry<R, C> entry, final boolean reusable) {
        if (entry == null) {
            return;
        }
        final R route = entry.getRoute();
        final Segment seg = segments.get(route);
        if (seg == null) {
            discardEntry(entry, CloseMode.GRACEFUL);
            fireOnRelease(route);
            return;
        }

        final long now = clock.millis();
        final boolean stillValid = reusable && !isPastTtl(entry, now) && !entry.getExpiryDeadline().isBefore(now);

        if (stillValid) {
            if (!handOffToCompatibleWaiter(entry, seg)) {
                offerAvailable(seg, entry);
                if (!seg.waiters.isEmpty()) {
                    enqueueIfNeeded(route, seg);
                    triggerDrainIfMany();
                }
            }
        } else {
            discardAndDecr(entry, CloseMode.GRACEFUL);
        }

        maybeCleanupSegment(route, seg);
        fireOnRelease(route);
    }

    @Override
    public void close() throws IOException {
        close(CloseMode.GRACEFUL);
    }

    @Override
    public void close(final CloseMode closeMode) {
        if (!closed.compareAndSet(false, true)) {
            return;
        }

        timeouts.shutdownNow();

        for (final Map.Entry<R, Segment> e : segments.entrySet()) {
            final Segment seg = e.getValue();

            for (final Waiter w : seg.waiters) {
                w.cancelled = true;
                cancelTimeout(w);
                w.completeExceptionally(new TimeoutException("Pool closed"));
            }
            seg.waiters.clear();
            if (seg.enqueued.getAndSet(false)) {
                pendingRouteCount.decrementAndGet();
            }

            for (final PoolEntry<R, C> p : seg.available) {
                if (seg.available.remove(p)) {
                    discardEntry(p, closeMode);
                }
            }
            seg.available.clear();

            final int alloc = seg.allocated.getAndSet(0);
            if (alloc != 0) {
                totalAllocated.addAndGet(-alloc);
            }
        }
        segments.clear();
        pendingQueue.clear();
        pendingRouteCount.set(0);

        // Let in-flight graceful closes progress; no blocking here.
        disposer.shutdown();
    }

    @Override
    public void closeIdle(final TimeValue idleTime) {
        final long cutoff = clock.millis()
                - Math.max(0L, idleTime != null ? idleTime.toMilliseconds() : 0L);

        for (final Map.Entry<R, Segment> e : segments.entrySet()) {
            final R route = e.getKey();
            final Segment seg = e.getValue();

            int processed = 0;
            final int cap = 64;

            for (final PoolEntry<R, C> p : seg.available) {
                if (p.getUpdated() <= cutoff) {
                    if (seg.available.remove(p)) {
                        discardAndDecr(p, CloseMode.GRACEFUL);
                        if (++processed == cap) {
                            break;
                        }
                    }
                }
            }
            maybeCleanupSegment(route, seg);
        }
    }

    @Override
    public void closeExpired() {
        final long now = clock.millis();

        for (final Map.Entry<R, Segment> e : segments.entrySet()) {
            final R route = e.getKey();
            final Segment seg = e.getValue();

            int processed = 0;
            final int cap = 64;

            for (final PoolEntry<R, C> p : seg.available) {
                if (p.getExpiryDeadline().isBefore(now) || isPastTtl(p, now)) {
                    if (seg.available.remove(p)) {
                        discardAndDecr(p, CloseMode.GRACEFUL);
                        if (++processed == cap) {
                            break;
                        }
                    }
                }
            }
            maybeCleanupSegment(route, seg);
        }
    }

    @Override
    public Set<R> getRoutes() {
        final Set<R> out = new HashSet<>();
        for (final Map.Entry<R, Segment> e : segments.entrySet()) {
            final Segment s = e.getValue();
            if (!s.available.isEmpty() || s.allocated.get() > 0 || !s.waiters.isEmpty()) {
                out.add(e.getKey());
            }
        }
        return out;
    }

    @Override
    public int getMaxTotal() {
        return maxTotal.get();
    }

    @Override
    public void setMaxTotal(final int max) {
        maxTotal.set(Math.max(1, max));
    }

    @Override
    public int getDefaultMaxPerRoute() {
        return defaultMaxPerRoute.get();
    }

    @Override
    public void setDefaultMaxPerRoute(final int max) {
        defaultMaxPerRoute.set(Math.max(1, max));
    }

    @Override
    public int getMaxPerRoute(final R route) {
        final Integer v = maxPerRoute.get(route);
        return v != null ? v : defaultMaxPerRoute.get();
    }

    @Override
    public void setMaxPerRoute(final R route, final int max) {
        if (max <= 0) {
            maxPerRoute.remove(route);
        } else {
            maxPerRoute.put(route, max);
        }
    }

    @Override
    public PoolStats getTotalStats() {
        int leased = 0, availableCount = 0, pending = 0;
        for (final Segment seg : segments.values()) {
            final int alloc = seg.allocated.get();
            final int avail = seg.available.size();
            leased += Math.max(0, alloc - avail);
            availableCount += avail;
            pending += seg.waiters.size();
        }
        return new PoolStats(leased, pending, availableCount, getMaxTotal());
    }

    @Override
    public PoolStats getStats(final R route) {
        final Segment seg = segments.get(route);
        if (seg == null) {
            return new PoolStats(0, 0, 0, getMaxPerRoute(route));
        }
        final int alloc = seg.allocated.get();
        final int avail = seg.available.size();
        final int leased = Math.max(0, alloc - avail);
        final int pending = seg.waiters.size();
        return new PoolStats(leased, pending, avail, getMaxPerRoute(route));
    }

    private void ensureOpen() {
        if (closed.get()) {
            throw new IllegalStateException("Pool is closed");
        }
    }

    private boolean isPastTtl(final PoolEntry<R, C> entry, final long now) {
        if (!hasTtl) {
            return false;
        }
        return now - entry.getCreated() >= ttlMillis;
    }

    private void scheduleTimeout(final Waiter w, final Segment seg) {
        if (!TimeValue.isPositive(w.requestTimeout)) {
            return;
        }
        w.timeoutTask = timeouts.schedule(() -> {
            if (w.isDone()) {
                return;
            }
            w.cancelled = true;
            seg.waiters.remove(w);
            w.completeExceptionally(new TimeoutException("Lease timed out"));
            dequeueIfDrained(seg);
            maybeCleanupSegment(w.route, seg);

            final PoolEntry<R, C> p = pollAvailable(seg, w.state);
            if (p != null) {
                if (!handOffToCompatibleWaiter(p, seg)) {
                    offerAvailable(seg, p);
                }
            }
        }, w.requestTimeout.toMilliseconds(), TimeUnit.MILLISECONDS);
    }

    private void cancelTimeout(final Waiter w) {
        final ScheduledFuture<?> t = w.timeoutTask;
        if (t != null) {
            t.cancel(false);
        }
    }

    private void offerAvailable(final Segment seg, final PoolEntry<R, C> p) {
        if (reusePolicy == PoolReusePolicy.LIFO) {
            seg.available.addFirst(p);
        } else {
            seg.available.addLast(p);
        }
    }

    private PoolEntry<R, C> pollAvailable(final Segment seg, final Object neededState) {
        if (neededState == null) {
            return seg.available.pollFirst();
        }
        for (final PoolEntry<R, C> p : seg.available) {
            if (compatible(neededState, p.getState())) {
                if (seg.available.remove(p)) {
                    return p;
                }
            }
        }
        return null;
    }

    private boolean compatible(final Object needed, final Object have) {
        return needed == null || Objects.equals(needed, have);
    }

    private boolean handOffToCompatibleWaiter(final PoolEntry<R, C> entry, final Segment seg) {
        Deque<Waiter> skipped = null;
        boolean handedOff = false;

        for (; ; ) {
            final Waiter w = seg.waiters.pollFirst();
            if (w == null) {
                break;
            }
            if (w.cancelled || w.isDone()) {
                continue;
            }
            if (compatible(w.state, entry.getState())) {
                cancelTimeout(w);
                handedOff = w.complete(entry);
                if (handedOff) {
                    fireOnLease(w.route);
                    dequeueIfDrained(seg);
                    break;
                }
            } else {
                if (skipped == null) {
                    skipped = new ArrayDeque<>();
                }
                skipped.addLast(w);
            }
        }

        if (skipped != null) {
            while (!skipped.isEmpty()) {
                seg.waiters.addFirst(skipped.pollLast());
            }
        }
        return handedOff;
    }

    private void discardAndDecr(final PoolEntry<R, C> p, final CloseMode mode) {
        totalAllocated.decrementAndGet();
        final Segment seg = segments.get(p.getRoute());
        if (seg != null) {
            seg.allocated.decrementAndGet();
        }
        discardEntry(p, mode);
    }

    private void maybeCleanupSegment(final R route, final Segment seg) {
        if (seg.allocated.get() == 0 && seg.available.isEmpty() && seg.waiters.isEmpty()) {
            segments.remove(route, seg);
            if (seg.enqueued.getAndSet(false)) {
                pendingRouteCount.decrementAndGet();
            }
        }
    }

    private boolean tryAllocateOne(final R route, final Segment seg) {
        for (; ; ) {
            final int tot = totalAllocated.get();
            if (tot >= maxTotal.get()) {
                return false;
            }
            if (!totalAllocated.compareAndSet(tot, tot + 1)) {
                continue;
            }
            for (; ; ) {
                final int per = seg.allocated.get();
                if (per >= seg.limitPerRoute(route)) {
                    totalAllocated.decrementAndGet();
                    return false;
                }
                if (seg.allocated.compareAndSet(per, per + 1)) {
                    return true;
                }
            }
        }
    }

    private void enqueueIfNeeded(final R route, final Segment seg) {
        if (seg.enqueued.compareAndSet(false, true)) {
            pendingQueue.offer(route);
            pendingRouteCount.incrementAndGet();
        }
    }

    private void dequeueIfDrained(final Segment seg) {
        if (seg.waiters.isEmpty() && seg.enqueued.getAndSet(false)) {
            pendingRouteCount.decrementAndGet();
        }
    }

    private void triggerDrainIfMany() {
        if (pendingRouteCount.get() < RR_MIN_PENDING_ROUTES) {
            return;
        }
        if (totalAllocated.get() >= maxTotal.get()) {
            return;
        }
        if (!draining.compareAndSet(false, true)) {
            return;
        }

        final Runnable task = () -> {
            try {
                serveRoundRobin(RR_BUDGET);
            } finally {
                draining.set(false);
                if (pendingRouteCount.get() >= RR_MIN_PENDING_ROUTES
                        && totalAllocated.get() < maxTotal.get()
                        && !pendingQueue.isEmpty()) {
                    triggerDrainIfMany();
                }
            }
        };

        try {
            disposer.execute(task);
        } catch (final RejectedExecutionException rejected) {
            // Disposer saturated: keep pool correct; do a tiny bounded inline drain.
            try {
                serveRoundRobin(RR_INLINE_FALLBACK_BUDGET);
            } finally {
                draining.set(false);
            }
        }
    }

    private void serveRoundRobin(final int budget) {
        int created = 0;

        for (; created < budget; ) {
            final R route = pendingQueue.poll();
            if (route == null) {
                break;
            }
            final Segment seg = segments.get(route);
            if (seg == null) {
                continue;
            }
            if (seg.waiters.isEmpty()) {
                if (seg.enqueued.getAndSet(false)) {
                    pendingRouteCount.decrementAndGet();
                }
                continue;
            }

            if (!tryAllocateOne(route, seg)) {
                pendingQueue.offer(route);
                continue;
            }

            final Waiter w = seg.waiters.pollFirst();
            if (w == null || w.cancelled || w.isDone()) {
                seg.allocated.decrementAndGet();
                totalAllocated.decrementAndGet();
            } else {
                final PoolEntry<R, C> entry = new PoolEntry<>(route, timeToLive, disposal, clock);
                cancelTimeout(w);
                w.complete(entry);
                fireOnLease(w.route);
                created++;
            }

            if (!seg.waiters.isEmpty()) {
                pendingQueue.offer(route);
            } else {
                if (seg.enqueued.getAndSet(false)) {
                    pendingRouteCount.decrementAndGet();
                }
            }
        }
    }

    /**
     * Dispose a pool entry's connection asynchronously if possible; under pressure fall back
     * to IMMEDIATE close on the caller.
     */
    private void discardEntry(final PoolEntry<R, C> p, final CloseMode preferred) {
        final CloseMode mode = preferred != null ? preferred : CloseMode.IMMEDIATE;

        // Keep immediate closes predictable: do it now.
        if (mode == CloseMode.IMMEDIATE) {
            p.discardConnection(CloseMode.IMMEDIATE);
            return;
        }

        if (disposer.isShutdown()) {
            p.discardConnection(CloseMode.IMMEDIATE);
            return;
        }

        final BlockingQueue<Runnable> q = disposer.getQueue();
        if (q.remainingCapacity() == 0) {
            p.discardConnection(CloseMode.IMMEDIATE);
            return;
        }
        try {
            disposer.execute(() -> {
                try {
                    p.discardConnection(mode);
                } catch (final RuntimeException ignore) {
                    // best-effort
                }
            });
        } catch (final RejectedExecutionException saturated) {
            // Saturated or shutting down: never block caller
            p.discardConnection(CloseMode.IMMEDIATE);
        }
    }

    private void fireOnLease(final R route) {
        final ConnPoolListener<R> listener = this.connPoolListener;
        if (listener != null) {
            try {
                listener.onLease(route, this);
            } catch (final RuntimeException ignore) {
                // ignore
            }
        }
    }

    private void fireOnRelease(final R route) {
        final ConnPoolListener<R> listener = this.connPoolListener;
        if (listener != null) {
            try {
                listener.onRelease(route, this);
            } catch (final RuntimeException ignore) {
                // ignore
            }
        }
    }
}
