1 /***********************************************************************
2  * Copyright (c) 2013-2024 Commonwealth Computer Research, Inc.
3  * All rights reserved. This program and the accompanying materials
4  * are made available under the terms of the Apache License, Version 2.0
5  * which accompanies this distribution and is available at
6  * http://www.opensource.org/licenses/apache2.0.php.
7  ***********************************************************************/
8 
9 package org.locationtech.geomesa.utils.io
10 
11 import org.apache.commons.pool2.impl.{DefaultPooledObject, GenericObjectPool, GenericObjectPoolConfig}
12 import org.apache.commons.pool2.{BasePooledObjectFactory, PooledObject, SwallowedExceptionListener}
13 
14 import java.io.Closeable
15 
16 /**
17   * Pool for sharing a finite set of closeable resources
18   *
19   * @tparam T object type
20   */
21 trait CloseablePool[T <: Closeable] extends Closeable {
22 
23   /**
24     * Borrow an item from the pool. The item will be returned to the pool after execution.
25     *
26     * @param fn function to execute on the borrowed item
27     * @tparam U return type
28     * @return
29     */
30   def borrow[U](fn: T => U): U
31 }
32 
33 object CloseablePool {
34 
35   /**
36     * Create a pool
37     *
38     * @param factory method for instantiating pool objects
39     * @param size max size of the pool
40     * @tparam T object type
41     * @return
42     */
43   def apply[T <: Closeable](factory: => T, size: Int = 8): CloseablePool[T] = {
44     val config = new GenericObjectPoolConfig[T]()
45     config.setMaxTotal(size)
46     new CommonsPoolPool(factory, config)
47   }
48 
49   /**
50     * Apache commons-pool-backed pool
51     *
52     * @param create factory method for creating new pool objects
53     * @param config pool configuration options
54     * @tparam T object type
55     */
56   class CommonsPoolPool[T <: Closeable](create: => T, config: GenericObjectPoolConfig[T]) extends CloseablePool[T] {
57 
58     private val factory: BasePooledObjectFactory[T] = new BasePooledObjectFactory[T] {
59       override def wrap(obj: T) = new DefaultPooledObject[T](obj)
60       override def create(): T = CommonsPoolPool.this.create
61       override def destroyObject(p: PooledObject[T]): Unit = p.getObject.close()
62     }
63 
64     private val pool = new GenericObjectPool[T](factory, config)
65 
66     override def borrow[U](fn: T => U): U = {
67       val t = pool.borrowObject()
68       try { fn(t) } finally {
69         pool.returnObject(t)
70       }
71     }
72 
73     override def close(): Unit = {
74       val errors = new java.util.concurrent.LinkedBlockingQueue[Exception]()
75       pool.setSwallowedExceptionListener(new SwallowedExceptionListener() {
76         override def onSwallowException(e: Exception): Unit = errors.offer(e)
77       })
78       pool.close()
79       if (!errors.isEmpty) {
80         val e = errors.poll()
81         while (!errors.isEmpty) {
82           e.addSuppressed(errors.poll())
83         }
84         throw e
85       }
86     }
87   }
88 }