Package reactor.pool

Class SimpleDequePool<POOLABLE>

java.lang.Object
reactor.pool.SimpleDequePool<POOLABLE>
All Implemented Interfaces:
Disposable, InstrumentedPool<POOLABLE>, InstrumentedPool.PoolMetrics, Pool<POOLABLE>
public class SimpleDequePool<POOLABLE> extends Object
The SimpleDequePool is based on Deque for idle resources and pending Pool.acquire() Monos, allowing both to be ordered either LIFO or FIFO. It uses non-blocking drain loops to deliver resources to borrowers, which means that a resource could be handed off on any of the following threads:
  • any thread on which a resource was last allocated
  • any thread on which a resource was recently released
  • any thread on which an Pool.acquire() Mono was subscribed
For a more deterministic approach, the PoolBuilder.acquisitionScheduler(Scheduler) property of the builder can be used.
Author:
Simon Baslé

  • Field Details

  • Method Details

    • acquire

      public Mono<PooledRef<POOLABLE>> acquire()
      Description copied from interface: Pool
      Manually acquire a POOLABLE from the pool upon subscription and become responsible for its release. The object is wrapped in a PooledRef which can be used for manually releasing the object back to the pool or invalidating it. As a result, you MUST maintain a reference to it throughout the code that makes use of the underlying resource.

      This is typically the case when one needs to wrap the actual resource into a decorator version, where the reference to the PooledRef can be stored. On the other hand, if the resource and its usage directly expose reactive APIs, you might want to prefer to use Pool.withPoolable(Function).

      The resulting Mono emits the PooledRef as the POOLABLE becomes available. Cancelling the Subscription before the POOLABLE has been emitted will either avoid object acquisition entirely or will translate to a release of the POOLABLE. Once the resource is emitted though, it is the responsibility of the caller to release the poolable object via the PooledRef release methods when the resource is not used anymore (directly OR indirectly, eg. the results from multiple statements derived from a DB connection type of resource have all been processed).

      Returns:
      a Mono, each subscription to which represents an individual act of acquiring a pooled object and manually managing its lifecycle from there on
      See Also:

    • acquire

      public Mono<PooledRef<POOLABLE>> acquire(Duration timeout)
      Description copied from interface: Pool
      Manually acquire a POOLABLE from the pool upon subscription and become responsible for its release. The provided Duration acts as a timeout that only applies if the acquisition is added to the pending queue, i.e. there is no idle resource and no new resource can be created currently, so one needs to wait for a release before a resource can be delivered. For a timeout that covers both this pending case and the time it would take to allocate a new resource, simply apply the Mono.timeout(Duration) operator to the returned Mono. For a timeout that only applies to resource allocation, build the pool with the standard Mono.timeout(Duration) operator chained to the allocator.

      The object is wrapped in a PooledRef which can be used for manually releasing the object back to the pool or invalidating it. As a result, you MUST maintain a reference to it throughout the code that makes use of the underlying resource.

      This is typically the case when one needs to wrap the actual resource into a decorator version, where the reference to the PooledRef can be stored. On the other hand, if the resource and its usage directly expose reactive APIs, you might want to prefer to use Pool.withPoolable(Function).

      The resulting Mono emits the PooledRef as the POOLABLE becomes available. Cancelling the Subscription before the POOLABLE has been emitted will either avoid object acquisition entirely or will translate to a release of the POOLABLE. Once the resource is emitted though, it is the responsibility of the caller to release the poolable object via the PooledRef release methods when the resource is not used anymore (directly OR indirectly, eg. the results from multiple statements derived from a DB connection type of resource have all been processed).

      Returns:
      a Mono, each subscription to which represents an individual act of acquiring a pooled object and manually managing its lifecycle from there on
      See Also:

    • acquiredSize

      public int acquiredSize()
      Description copied from interface: InstrumentedPool.PoolMetrics
      Measure the current number of resources that have been successfully acquired and are in active use, outside of the control of the pool until they're released back to it. This number is only incremented after the resource has been successfully allocated and is about to be handed off to the subscriber of Pool.acquire().
      Specified by:
      acquiredSize in interface InstrumentedPool.PoolMetrics
      Returns:
      the number of acquired resources

    • disposeLater

      public Mono<Void> disposeLater()
      Description copied from interface: Pool
      Returns a Mono that represents a lazy asynchronous shutdown of this Pool. Shutdown doesn't happen until the Mono is subscribed. Otherwise, it performs the same steps as in the imperative counterpart, Pool.dispose().

      If the pool has been already shut down, returns Mono.empty(). Completion of the Mono indicates completion of the shutdown process.

      Returns:
      a Mono triggering the shutdown of the pool once subscribed.

    • idleSize

      public int idleSize()
      Description copied from interface: InstrumentedPool.PoolMetrics
      Measure the current number of idle resources in the Pool.

      Note that some resources might be lazily evicted when they're next considered for an incoming Pool.acquire() call. Such resources would still count towards this method.

      Specified by:
      idleSize in interface InstrumentedPool.PoolMetrics
      Returns:
      the number of idle resources

    • warmup

      public Mono<Integer> warmup()
      Description copied from interface: Pool
      Warms up the Pool, if needed. This typically instructs the pool to check for a minimum size and allocate necessary objects when the minimum is not reached. The resulting Mono emits the number of extra resources it created as a result of the allocation minimum.

      Note that no work nor allocation is performed until the Mono is subscribed to.

      Implementations MAY include more behavior, but there is no restriction on the way this method is called by users (it should be possible to call it at any time, as many times as needed or not at all).

      Returns:
      a cold Mono that triggers resource warmup and emits the number of warmed up resources

    • pendingAcquireSize

      public int pendingAcquireSize()
      Description copied from interface: InstrumentedPool.PoolMetrics
      Measure the current number of "pending" acquire Monos in the Pool.

      An acquire is in the pending state when it is attempted at a point when no idle resource is available in the pool, and no new resource can be created.

      Returns:
      the number of pending acquire

    • isDisposed

      public boolean isDisposed()

    • config

      public PoolConfig<POOLABLE> config()
      Description copied from interface: Pool
      Return the pool's configuration.
      Specified by:
      config in interface Pool<POOLABLE>
      Returns:
      the PoolConfig

    • metrics

      public InstrumentedPool.PoolMetrics metrics()
      Specified by:
      metrics in interface InstrumentedPool<POOLABLE>
      Returns:
      a InstrumentedPool.PoolMetrics object to be used to get live gauges about the Pool

    • allocatedSize

      public int allocatedSize()
      Description copied from interface: InstrumentedPool.PoolMetrics
      Measure the current number of allocated resources in the Pool, acquired or idle.
      Specified by:
      allocatedSize in interface InstrumentedPool.PoolMetrics
      Returns:
      the total number of allocated resources managed by the Pool

    • getMaxAllocatedSize

      public int getMaxAllocatedSize()
      Description copied from interface: InstrumentedPool.PoolMetrics
      Get the maximum number of live resources this Pool will allow.

      A Pool might be unbounded, in which case this method returns Integer.MAX_VALUE.

      Specified by:
      getMaxAllocatedSize in interface InstrumentedPool.PoolMetrics
      Returns:
      the maximum number of live resources that can be allocated by this Pool

    • getMaxPendingAcquireSize

      public int getMaxPendingAcquireSize()
      Description copied from interface: InstrumentedPool.PoolMetrics
      Get the maximum number of Pool.acquire() this Pool can queue in a pending state when no available resource is immediately handy (and the Pool cannot allocate more resources).

      A Pool pending queue might be unbounded, in which case this method returns Integer.MAX_VALUE.

      Specified by:
      getMaxPendingAcquireSize in interface InstrumentedPool.PoolMetrics
      Returns:
      the maximum number of pending acquire that can be enqueued by this Pool

    • secondsSinceLastInteraction

      public long secondsSinceLastInteraction()
      Description copied from interface: InstrumentedPool.PoolMetrics
      Measure the duration in seconds since the pool was last interacted with in a meaningful way. This is a best effort indicator of pool inactivity, provided the pool counters (InstrumentedPool.PoolMetrics.acquiredSize(), InstrumentedPool.PoolMetrics.idleSize(), InstrumentedPool.PoolMetrics.pendingAcquireSize() and InstrumentedPool.PoolMetrics.allocatedSize()) are also at zero.

      The lower the duration, the greater the chances that an interaction could be occurring in parallel to this call. This is why the duration is truncated to the second. A pool implementation that cannot yet support this measurement MAY choose to return -1 seconds instead.

      Interactions include background eviction, disposal of the pool, explicit pool warmup, resource acquisition and release (in the default implementation, any interaction triggering the drain loop)...

      Specified by:
      secondsSinceLastInteraction in interface InstrumentedPool.PoolMetrics
      Returns:
      a number of seconds indicative of the time elapsed since last pool interaction
      See Also:

    • isInactiveForMoreThan

      public boolean isInactiveForMoreThan(Duration duration)
      Description copied from interface: InstrumentedPool.PoolMetrics
      A convenience way to check the pool is inactive, in the sense that InstrumentedPool.PoolMetrics.acquiredSize(), InstrumentedPool.PoolMetrics.idleSize(), InstrumentedPool.PoolMetrics.pendingAcquireSize() and InstrumentedPool.PoolMetrics.allocatedSize() are all at zero and that the last recorded interaction with the pool (InstrumentedPool.PoolMetrics.secondsSinceLastInteraction()) was more than or exactly duration ago.
      Specified by:
      isInactiveForMoreThan in interface InstrumentedPool.PoolMetrics
      Returns:
      true if the pool can be considered inactive (see above), false otherwise
      See Also: