This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/cayenne.git
commit a8b87fd8317be2a04bf5cc42e430e703540c24d9 Author: Andrus Adamchik <[email protected]> AuthorDate: Tue Jul 21 17:33:51 2026 +0200 Internal DataSource impl cleanup --- .../datasource/BadValidationQueryException.java | 3 +- .../cayenne/datasource/CayenneDataSource.java | 4 +- .../datasource/ManagedPoolingDataSource.java | 5 +- .../cayenne/datasource/PoolingDataSource.java | 30 - .../datasource/PoolingDataSourceBuilder.java | 4 +- .../datasource/PoolingDataSourceManager.java | 7 +- .../cayenne/datasource/StoppedDataSource.java | 5 +- .../datasource/UnmanagedPoolingDataSource.java | 741 ++++++++++----------- .../cayenne/datasource/CayenneDataSourceTest.java | 4 +- 9 files changed, 377 insertions(+), 426 deletions(-) diff --git a/cayenne/src/main/java/org/apache/cayenne/datasource/BadValidationQueryException.java b/cayenne/src/main/java/org/apache/cayenne/datasource/BadValidationQueryException.java index f7f4600a4..58157adb8 100644 --- a/cayenne/src/main/java/org/apache/cayenne/datasource/BadValidationQueryException.java +++ b/cayenne/src/main/java/org/apache/cayenne/datasource/BadValidationQueryException.java @@ -21,8 +21,7 @@ package org.apache.cayenne.datasource; import java.sql.SQLException; /** - * Thrown when the pool fails to validate a fresh connection that is known to be - * in a good state. + * Thrown when the pool fails to validate a fresh connection that is known to be in a good state. * * @since 4.0 */ diff --git a/cayenne/src/main/java/org/apache/cayenne/datasource/CayenneDataSource.java b/cayenne/src/main/java/org/apache/cayenne/datasource/CayenneDataSource.java index 1911f098d..86b2a67c7 100644 --- a/cayenne/src/main/java/org/apache/cayenne/datasource/CayenneDataSource.java +++ b/cayenne/src/main/java/org/apache/cayenne/datasource/CayenneDataSource.java @@ -228,7 +228,7 @@ public class CayenneDataSource { /** * Builds a DataSource that is pooling if {@link #pool(int, int)} was called, and non-pooling otherwise. A - * pooling DataSource implements {@link PoolingDataSource} and must be explicitly closed by the caller when no + * pooling DataSource is {@link AutoCloseable} and must be explicitly closed by the caller when no * longer in use. */ public DataSource build() { @@ -247,7 +247,7 @@ public class CayenneDataSource { return minConnections != null ? pool(nonPooling) : nonPooling; } - private PoolingDataSource pool(DataSource nonPooling) { + private DataSource pool(DataSource nonPooling) { PoolingDataSourceParameters parameters = new PoolingDataSourceParameters(); parameters.setMinConnections(minConnections); diff --git a/cayenne/src/main/java/org/apache/cayenne/datasource/ManagedPoolingDataSource.java b/cayenne/src/main/java/org/apache/cayenne/datasource/ManagedPoolingDataSource.java index 6237fcbb7..16c62dbd9 100644 --- a/cayenne/src/main/java/org/apache/cayenne/datasource/ManagedPoolingDataSource.java +++ b/cayenne/src/main/java/org/apache/cayenne/datasource/ManagedPoolingDataSource.java @@ -33,7 +33,7 @@ import java.util.logging.Logger; * * @since 4.0 */ -public class ManagedPoolingDataSource implements PoolingDataSource, ScopeEventListener { +public class ManagedPoolingDataSource implements DataSource, AutoCloseable, ScopeEventListener { private final PoolingDataSourceManager dataSourceManager; private DataSource dataSource; @@ -74,8 +74,7 @@ public class ManagedPoolingDataSource implements PoolingDataSource, ScopeEventLi @Override public void close() { - // swap the underlying DataSource to prevent further interaction with - // the callers + // swap the underlying DataSource to prevent further interaction with the callers this.dataSource = new StoppedDataSource(dataSource); // shut down the thread.. diff --git a/cayenne/src/main/java/org/apache/cayenne/datasource/PoolingDataSource.java b/cayenne/src/main/java/org/apache/cayenne/datasource/PoolingDataSource.java deleted file mode 100644 index d4e770595..000000000 --- a/cayenne/src/main/java/org/apache/cayenne/datasource/PoolingDataSource.java +++ /dev/null @@ -1,30 +0,0 @@ -/***************************************************************** - * 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 - * - * https://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.cayenne.datasource; - -import javax.sql.DataSource; - -/** - * A {@link DataSource} that pools connections and requires to be explicitly closed. - * - * @since 4.0 - */ -public interface PoolingDataSource extends DataSource, AutoCloseable { - -} diff --git a/cayenne/src/main/java/org/apache/cayenne/datasource/PoolingDataSourceBuilder.java b/cayenne/src/main/java/org/apache/cayenne/datasource/PoolingDataSourceBuilder.java index fd8fdd9bf..b1059757c 100644 --- a/cayenne/src/main/java/org/apache/cayenne/datasource/PoolingDataSourceBuilder.java +++ b/cayenne/src/main/java/org/apache/cayenne/datasource/PoolingDataSourceBuilder.java @@ -69,7 +69,7 @@ public class PoolingDataSourceBuilder { * Builds a pooling DataSource that needs to be explicitly closed by the * caller when no longer in use. */ - public PoolingDataSource build() { + public ManagedPoolingDataSource build() { // sanity checks... if (poolParameters.getMaxConnections() < 0) { @@ -94,7 +94,7 @@ public class PoolingDataSourceBuilder { return new UnmanagedPoolingDataSource(nonPoolingDataSource, poolParameters); } - private PoolingDataSource buildManaged(UnmanagedPoolingDataSource dataSource) { + private ManagedPoolingDataSource buildManaged(UnmanagedPoolingDataSource dataSource) { return new ManagedPoolingDataSource(dataSource); } diff --git a/cayenne/src/main/java/org/apache/cayenne/datasource/PoolingDataSourceManager.java b/cayenne/src/main/java/org/apache/cayenne/datasource/PoolingDataSourceManager.java index 0df84b7dc..9e5bcba06 100644 --- a/cayenne/src/main/java/org/apache/cayenne/datasource/PoolingDataSourceManager.java +++ b/cayenne/src/main/java/org/apache/cayenne/datasource/PoolingDataSourceManager.java @@ -27,17 +27,18 @@ package org.apache.cayenne.datasource; */ class PoolingDataSourceManager extends Thread { + private final UnmanagedPoolingDataSource dataSource; + private final long managerWakeTime; + private volatile boolean shouldStop; - private UnmanagedPoolingDataSource dataSource; - private long managerWakeTime; PoolingDataSourceManager(UnmanagedPoolingDataSource dataSource, long managerWakeTime) { setName("PoolingDataSourceManager-" + dataSource.hashCode()); setDaemon(true); this.dataSource = dataSource; - this.shouldStop = false; this.managerWakeTime = managerWakeTime; + this.shouldStop = false; } void shutdown() { diff --git a/cayenne/src/main/java/org/apache/cayenne/datasource/StoppedDataSource.java b/cayenne/src/main/java/org/apache/cayenne/datasource/StoppedDataSource.java index 7c355c780..0f37f0385 100644 --- a/cayenne/src/main/java/org/apache/cayenne/datasource/StoppedDataSource.java +++ b/cayenne/src/main/java/org/apache/cayenne/datasource/StoppedDataSource.java @@ -18,14 +18,13 @@ ****************************************************************/ package org.apache.cayenne.datasource; +import javax.sql.DataSource; import java.io.PrintWriter; import java.sql.Connection; import java.sql.SQLException; import java.sql.SQLFeatureNotSupportedException; import java.util.logging.Logger; -import javax.sql.DataSource; - /** * A DataSource wrapper that prevents callers from obtaining connections from * another {@link DataSource}. @@ -34,7 +33,7 @@ import javax.sql.DataSource; */ class StoppedDataSource implements DataSource { - private DataSource dataSource; + private final DataSource dataSource; public StoppedDataSource(DataSource dataSource) { this.dataSource = dataSource; diff --git a/cayenne/src/main/java/org/apache/cayenne/datasource/UnmanagedPoolingDataSource.java b/cayenne/src/main/java/org/apache/cayenne/datasource/UnmanagedPoolingDataSource.java index 3fc1cb1d4..2f139dba0 100644 --- a/cayenne/src/main/java/org/apache/cayenne/datasource/UnmanagedPoolingDataSource.java +++ b/cayenne/src/main/java/org/apache/cayenne/datasource/UnmanagedPoolingDataSource.java @@ -18,11 +18,15 @@ ****************************************************************/ package org.apache.cayenne.datasource; +import org.apache.cayenne.CayenneRuntimeException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.sql.DataSource; import java.io.PrintWriter; import java.sql.Connection; import java.sql.SQLException; import java.sql.SQLFeatureNotSupportedException; -import java.util.Collections; import java.util.Map; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; @@ -30,386 +34,365 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; -import javax.sql.DataSource; - -import org.apache.cayenne.CayenneRuntimeException; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - /** * A non-blocking {@link DataSource} with a pool of connections. - * + * * @since 4.0 */ -public class UnmanagedPoolingDataSource implements PoolingDataSource { - - // An old hack that fixes Sybase problems with autocommit. Used idea from - // Jonas org.objectweb.jonas.jdbc_xa.ConnectionImpl - // (http://www.objectweb.org/jonas/). - // - // If problem is not the one that can be fixed by this patch, original - // exception is rethrown. If exception occurs when fixing the problem, new - // exception is thrown. - // - static void sybaseAutoCommitPatch(Connection c, SQLException e, boolean autoCommit) throws SQLException { - - String s = e.getMessage().toLowerCase(); - if (s.contains("set chained command not allowed")) { - - // TODO: doing 'commit' here is extremely dangerous... we need to - // get a hold of Sybase instance and verify whether this issue is - // still there, and fix it differently (and perhaps generically) by - // calling 'rollback' on connections (can we do it when getting - // connection from the pool? returning it to the pool?) - - c.commit(); - c.setAutoCommit(autoCommit); // Shouldn't fail now. - } else { - throw e; - } - } - - /** - * An exception indicating that a connection request waiting in the queue - * timed out and was unable to obtain a connection. - */ - public static class ConnectionUnavailableException extends SQLException { - private static final long serialVersionUID = 1063973806941023165L; - - public ConnectionUnavailableException(String message) { - super(message); - } - } - - /** - * Defines a maximum time in milliseconds that a connection request could - * wait in the connection queue. After this period expires, an exception - * will be thrown in the calling method. - */ - public static final int MAX_QUEUE_WAIT_DEFAULT = 20000; - - private static final Logger LOGGER = LoggerFactory.getLogger(UnmanagedPoolingDataSource.class); - - private DataSource nonPoolingDataSource; - private long maxQueueWaitTime; - - private Map<PoolAwareConnection, Object> pool; - private Semaphore poolCap; - private BlockingQueue<PoolAwareConnection> available; - - private int maxIdleConnections; - private int minConnections; - private int maxConnections; - private String validationQuery; - - static int maxIdleConnections(int min, int max) { - return min == max ? min : min + (int) Math.ceil((max - min) / 2d); - } - - public UnmanagedPoolingDataSource(DataSource nonPoolingDataSource, PoolingDataSourceParameters parameters) { - - int minConnections = parameters.getMinConnections(); - int maxConnections = parameters.getMaxConnections(); - - // sanity check - if (minConnections < 0) { - throw new IllegalArgumentException("Negative min connections: " + minConnections); - } - - if (maxConnections < 0) { - throw new IllegalArgumentException("Negative max connections: " + maxConnections); - } - - if (minConnections > maxConnections) { - throw new IllegalArgumentException("Min connections (" + minConnections - + ") is greater than max connections (" + maxConnections + ")"); - } - - this.nonPoolingDataSource = nonPoolingDataSource; - this.maxQueueWaitTime = parameters.getMaxQueueWaitTime(); - this.validationQuery = parameters.getValidationQuery(); - this.minConnections = minConnections; - this.maxConnections = maxConnections; - this.pool = new ConcurrentHashMap<PoolAwareConnection, Object>((int) (maxConnections / 0.75)); - this.available = new ArrayBlockingQueue<PoolAwareConnection>(maxConnections); - this.poolCap = new Semaphore(maxConnections); - this.maxIdleConnections = maxIdleConnections(minConnections, maxConnections); - - // grow pool to min connections - try { - for (int i = 0; i < minConnections; i++) { - PoolAwareConnection c = createUnchecked(); - reclaim(c); - } - } catch (BadValidationQueryException e) { - throw new CayenneRuntimeException("Bad validation query: " + validationQuery, e); - } catch (SQLException e) { - LOGGER.info("Error creating new connection when starting connection pool, ignoring", e); - } - } - - int poolSize() { - return pool.size(); - } - - int availableSize() { - return available.size(); - } - - int canExpandSize() { - return poolCap.availablePermits(); - } - - @Override - public void close() { - - // expecting surrounding environment to block new requests for - // connections before calling this method. Still previously unchecked - // connections may be returned. I.e. "pool" will not grow during - // shutdown, which is the only thing that we need - - for (PoolAwareConnection c : pool.keySet()) { - retire(c); - } - - available.clear(); - pool = Collections.emptyMap(); - } - - void managePool() { - - // do not grow or shrink abruptly ... open or close 1 connection on - // each call - - if (available.size() < minConnections) { - - try { - PoolAwareConnection c = createUnchecked(); - if (c != null) { - reclaim(c); - } - } catch (SQLException e) { - LOGGER.info("Error creating new connection when managing connection pool, ignoring", e); - } - - } else if (available.size() > maxIdleConnections) { - - PoolAwareConnection c = uncheckNonBlocking(false); - if (c != null) { - retire(c); - } - } - - } - - /** - * Closes the connection and removes it from the pool. The connection must - * be an unchecked connection. - */ - void retire(PoolAwareConnection connection) { - pool.remove(connection); - - poolCap.release(); - - try { - connection.getConnection().close(); - } catch (SQLException e) { - // ignore? - } - } - - /** - * Returns connection back to the pool if possible. The connection must be - * an unchecked connection. - */ - void reclaim(PoolAwareConnection connection) { - - // TODO: rollback any in-process tx? - - // the queue may overflow potentially and we won't be able to add the - // object - if (!available.offer(connection)) { - retire(connection); - } - } - - PoolAwareConnection uncheckNonBlocking(boolean validate) { - PoolAwareConnection c = available.poll(); - return validate ? validateUnchecked(c) : c; - } - - PoolAwareConnection uncheckBlocking(boolean validate) { - PoolAwareConnection c; - try { - c = available.poll(maxQueueWaitTime, TimeUnit.MILLISECONDS); - } catch (InterruptedException e) { - return null; - } - - return validate ? validateUnchecked(c) : c; - } - - PoolAwareConnection validateUnchecked(PoolAwareConnection c) { - - if (c == null || c.validate()) { - return c; - } - - // this will recursively validate all connections that exist in the pool - // until a valid one is found or a pool is exhausted - retire(c); - return validateUnchecked(available.poll()); - } - - PoolAwareConnection createUnchecked() throws SQLException { - - if (!poolCap.tryAcquire()) { - return null; - } - - PoolAwareConnection c; - try { - c = createWrapped(); - } catch (SQLException e) { - poolCap.release(); - throw e; - } - - pool.put(c, 1); - - // even though we got a fresh connection, let's still validate it... - // This will provide consistent behavior between cached and uncached - // connections in respect to invalid validation queries - if (!c.validate()) { - throw new BadValidationQueryException( - "Can't validate a fresh connection. Likely validation query is wrong: " + validationQuery); - } - - return c; - } - - PoolAwareConnection createWrapped() throws SQLException { - return new PoolAwareConnection(this, createUnwrapped(), validationQuery); - } - - /** - * Creates a new connection. - */ - Connection createUnwrapped() throws SQLException { - return nonPoolingDataSource.getConnection(); - } - - /** - * Updates connection state to a default state. - */ - Connection resetState(Connection c) throws SQLException { - - // TODO: tx isolation level? - - if (!c.getAutoCommit()) { - - try { - c.setAutoCommit(true); - } catch (SQLException e) { - UnmanagedPoolingDataSource.sybaseAutoCommitPatch(c, e, true); - } - } - - c.clearWarnings(); - return c; - } - - @Override - public Connection getConnection() throws SQLException { - - // strategy for getting a connection - - // 1. quick peek for available connections - // 2. create new one - // 3. wait for a user to return connection - - PoolAwareConnection c; - - c = uncheckNonBlocking(true); - if (c != null) { - return resetState(c); - } - - c = createUnchecked(); - if (c != null) { - return resetState(c); - } - - c = uncheckBlocking(true); - if (c != null) { - return resetState(c); - } - - int poolSize = poolSize(); - int canGrow = poolCap.availablePermits(); - - throw new ConnectionUnavailableException("Can't obtain connection. Request to pool timed out. Total pool size: " - + poolSize + ", can expand by: " + canGrow); - } - - @Override - public Connection getConnection(String userName, String password) throws SQLException { - throw new UnsupportedOperationException( - "Connections for a specific user are not supported by the pooled DataSource"); - } - - @Override - public int getLoginTimeout() throws java.sql.SQLException { - return nonPoolingDataSource.getLoginTimeout(); - } - - @Override - public void setLoginTimeout(int seconds) throws java.sql.SQLException { - nonPoolingDataSource.setLoginTimeout(seconds); - } - - @Override - public PrintWriter getLogWriter() throws java.sql.SQLException { - return nonPoolingDataSource.getLogWriter(); - } - - @Override - public void setLogWriter(PrintWriter out) throws java.sql.SQLException { - nonPoolingDataSource.setLogWriter(out); - } - - @Override - public boolean isWrapperFor(Class<?> iface) throws SQLException { - return (UnmanagedPoolingDataSource.class.equals(iface)) ? true : nonPoolingDataSource.isWrapperFor(iface); - } - - @SuppressWarnings("unchecked") - @Override - public <T> T unwrap(Class<T> iface) throws SQLException { - return UnmanagedPoolingDataSource.class.equals(iface) ? (T) this : nonPoolingDataSource.unwrap(iface); - } - - @Override - public java.util.logging.Logger getParentLogger() throws SQLFeatureNotSupportedException { - return nonPoolingDataSource.getParentLogger(); - } - - String getValidationQuery() { - return validationQuery; - } - - long getMaxQueueWaitTime() { - return maxQueueWaitTime; - } - - int getMaxIdleConnections() { - return maxIdleConnections; - } - - int getMinConnections() { - return minConnections; - } - - int getMaxConnections() { - return maxConnections; - } +public class UnmanagedPoolingDataSource implements DataSource, AutoCloseable { + + // An old hack that fixes Sybase problems with autocommit. Used idea from + // Jonas org.objectweb.jonas.jdbc_xa.ConnectionImpl + // (http://www.objectweb.org/jonas/). + // + // If problem is not the one that can be fixed by this patch, original + // exception is rethrown. If exception occurs when fixing the problem, new + // exception is thrown. + // + static void sybaseAutoCommitPatch(Connection c, SQLException e, boolean autoCommit) throws SQLException { + + String s = e.getMessage().toLowerCase(); + if (s.contains("set chained command not allowed")) { + + // TODO: doing 'commit' here is extremely dangerous... we need to + // get a hold of Sybase instance and verify whether this issue is + // still there, and fix it differently (and perhaps generically) by + // calling 'rollback' on connections (can we do it when getting + // connection from the pool? returning it to the pool?) + + c.commit(); + c.setAutoCommit(autoCommit); // Shouldn't fail now. + } else { + throw e; + } + } + + /** + * An exception indicating that a connection request waiting in the queue + * timed out and was unable to obtain a connection. + */ + public static class ConnectionUnavailableException extends SQLException { + private static final long serialVersionUID = 1063973806941023165L; + + public ConnectionUnavailableException(String message) { + super(message); + } + } + + /** + * Defines a maximum time in milliseconds that a connection request could + * wait in the connection queue. After this period expires, an exception + * will be thrown in the calling method. + */ + public static final int MAX_QUEUE_WAIT_DEFAULT = 20000; + + private static final Logger LOGGER = LoggerFactory.getLogger(UnmanagedPoolingDataSource.class); + + private final DataSource nonPoolingDataSource; + private final long maxQueueWaitTime; + + private final Semaphore poolCap; + private final BlockingQueue<PoolAwareConnection> available; + + private final int maxIdleConnections; + private final int minConnections; + private final int maxConnections; + private final String validationQuery; + + private final Map<PoolAwareConnection, Object> pool; + + static int maxIdleConnections(int min, int max) { + return min == max ? min : min + (int) Math.ceil((max - min) / 2d); + } + + public UnmanagedPoolingDataSource(DataSource nonPoolingDataSource, PoolingDataSourceParameters parameters) { + + int minConnections = parameters.getMinConnections(); + int maxConnections = parameters.getMaxConnections(); + + // sanity check + if (minConnections < 0) { + throw new IllegalArgumentException("Negative min connections: " + minConnections); + } + + if (maxConnections < 0) { + throw new IllegalArgumentException("Negative max connections: " + maxConnections); + } + + if (minConnections > maxConnections) { + throw new IllegalArgumentException("Min connections (" + minConnections + + ") is greater than max connections (" + maxConnections + ")"); + } + + this.nonPoolingDataSource = nonPoolingDataSource; + this.maxQueueWaitTime = parameters.getMaxQueueWaitTime(); + this.validationQuery = parameters.getValidationQuery(); + this.minConnections = minConnections; + this.maxConnections = maxConnections; + this.pool = new ConcurrentHashMap<>((int) (maxConnections / 0.75)); + this.available = new ArrayBlockingQueue<>(maxConnections); + this.poolCap = new Semaphore(maxConnections); + this.maxIdleConnections = maxIdleConnections(minConnections, maxConnections); + + // grow pool to min connections + try { + for (int i = 0; i < minConnections; i++) { + PoolAwareConnection c = createUnchecked(); + reclaim(c); + } + } catch (BadValidationQueryException e) { + throw new CayenneRuntimeException("Bad validation query: " + validationQuery, e); + } catch (SQLException e) { + LOGGER.info("Error creating new connection when starting connection pool, ignoring", e); + } + } + + int poolSize() { + return pool.size(); + } + + int availableSize() { + return available.size(); + } + + int canExpandSize() { + return poolCap.availablePermits(); + } + + @Override + public void close() { + + // expecting surrounding environment to block new requests for + // connections before calling this method. Still previously unchecked + // connections may be returned. I.e. "pool" will not grow during + // shutdown, which is the only thing that we need + + for (PoolAwareConnection c : pool.keySet()) { + retire(c); + } + + available.clear(); + pool.clear(); + } + + void managePool() { + + // do not grow or shrink abruptly ... open or close 1 connection on + // each call + + if (available.size() < minConnections) { + + try { + PoolAwareConnection c = createUnchecked(); + if (c != null) { + reclaim(c); + } + } catch (SQLException e) { + LOGGER.info("Error creating new connection when managing connection pool, ignoring", e); + } + + } else if (available.size() > maxIdleConnections) { + + PoolAwareConnection c = uncheckNonBlocking(false); + if (c != null) { + retire(c); + } + } + + } + + /** + * Closes the connection and removes it from the pool. The connection must + * be an unchecked connection. + */ + void retire(PoolAwareConnection connection) { + pool.remove(connection); + + poolCap.release(); + + try { + connection.getConnection().close(); + } catch (SQLException e) { + // ignore? + } + } + + /** + * Returns connection back to the pool if possible. The connection must be + * an unchecked connection. + */ + void reclaim(PoolAwareConnection connection) { + + // TODO: rollback any in-process tx? + + // the queue may overflow potentially and we won't be able to add the + // object + if (!available.offer(connection)) { + retire(connection); + } + } + + PoolAwareConnection uncheckNonBlocking(boolean validate) { + PoolAwareConnection c = available.poll(); + return validate ? validateUnchecked(c) : c; + } + + private PoolAwareConnection uncheckBlocking() { + PoolAwareConnection c; + try { + c = available.poll(maxQueueWaitTime, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + return null; + } + + return validateUnchecked(c); + } + + PoolAwareConnection validateUnchecked(PoolAwareConnection c) { + + if (c == null || c.validate()) { + return c; + } + + // this will recursively validate all connections that exist in the pool + // until a valid one is found or a pool is exhausted + retire(c); + return validateUnchecked(available.poll()); + } + + PoolAwareConnection createUnchecked() throws SQLException { + + if (!poolCap.tryAcquire()) { + return null; + } + + PoolAwareConnection c; + try { + c = createWrapped(); + } catch (SQLException e) { + poolCap.release(); + throw e; + } + + pool.put(c, 1); + + // even though we got a fresh connection, let's still validate it... + // This will provide consistent behavior between cached and uncached + // connections in respect to invalid validation queries + if (!c.validate()) { + throw new BadValidationQueryException( + "Can't validate a fresh connection. Likely validation query is wrong: " + validationQuery); + } + + return c; + } + + PoolAwareConnection createWrapped() throws SQLException { + return new PoolAwareConnection(this, createUnwrapped(), validationQuery); + } + + /** + * Creates a new connection. + */ + Connection createUnwrapped() throws SQLException { + return nonPoolingDataSource.getConnection(); + } + + /** + * Updates connection state to a default state. + */ + Connection resetState(Connection c) throws SQLException { + + // TODO: tx isolation level? + + if (!c.getAutoCommit()) { + + try { + c.setAutoCommit(true); + } catch (SQLException e) { + UnmanagedPoolingDataSource.sybaseAutoCommitPatch(c, e, true); + } + } + + c.clearWarnings(); + return c; + } + + @Override + public Connection getConnection() throws SQLException { + + // strategy for getting a connection - + // 1. quick peek for available connections + // 2. create new one + // 3. wait for a user to return connection + + PoolAwareConnection c; + + c = uncheckNonBlocking(true); + if (c != null) { + return resetState(c); + } + + c = createUnchecked(); + if (c != null) { + return resetState(c); + } + + c = uncheckBlocking(); + if (c != null) { + return resetState(c); + } + + int poolSize = poolSize(); + int canGrow = poolCap.availablePermits(); + + throw new ConnectionUnavailableException("Can't obtain connection. Request to pool timed out. Total pool size: " + + poolSize + ", can expand by: " + canGrow); + } + + @Override + public Connection getConnection(String userName, String password) throws SQLException { + throw new UnsupportedOperationException( + "Connections for a specific user are not supported by the pooled DataSource"); + } + + @Override + public int getLoginTimeout() throws java.sql.SQLException { + return nonPoolingDataSource.getLoginTimeout(); + } + + @Override + public void setLoginTimeout(int seconds) throws java.sql.SQLException { + nonPoolingDataSource.setLoginTimeout(seconds); + } + + @Override + public PrintWriter getLogWriter() throws java.sql.SQLException { + return nonPoolingDataSource.getLogWriter(); + } + + @Override + public void setLogWriter(PrintWriter out) throws java.sql.SQLException { + nonPoolingDataSource.setLogWriter(out); + } + + @Override + public boolean isWrapperFor(Class<?> iface) throws SQLException { + return UnmanagedPoolingDataSource.class.equals(iface) || nonPoolingDataSource.isWrapperFor(iface); + } + + @SuppressWarnings("unchecked") + @Override + public <T> T unwrap(Class<T> iface) throws SQLException { + return UnmanagedPoolingDataSource.class.equals(iface) ? (T) this : nonPoolingDataSource.unwrap(iface); + } + + @Override + public java.util.logging.Logger getParentLogger() throws SQLFeatureNotSupportedException { + return nonPoolingDataSource.getParentLogger(); + } + + int getMaxConnections() { + return maxConnections; + } } diff --git a/cayenne/src/test/java/org/apache/cayenne/datasource/CayenneDataSourceTest.java b/cayenne/src/test/java/org/apache/cayenne/datasource/CayenneDataSourceTest.java index 9223c17c4..b3e7b2346 100644 --- a/cayenne/src/test/java/org/apache/cayenne/datasource/CayenneDataSourceTest.java +++ b/cayenne/src/test/java/org/apache/cayenne/datasource/CayenneDataSourceTest.java @@ -55,7 +55,7 @@ public class CayenneDataSourceTest { .pool(1, 2) .build(); - try (PoolingDataSource pooling = assertInstanceOf(ManagedPoolingDataSource.class, dataSource); + try (ManagedPoolingDataSource pooling = assertInstanceOf(ManagedPoolingDataSource.class, dataSource); Connection c = pooling.getConnection()) { assertFalse(c.isClosed()); } @@ -135,7 +135,7 @@ public class CayenneDataSourceTest { DataSource dataSource = CayenneDataSource.fromProperties(properties).build(); - try (PoolingDataSource pooling = assertInstanceOf(ManagedPoolingDataSource.class, dataSource); + try (ManagedPoolingDataSource pooling = assertInstanceOf(ManagedPoolingDataSource.class, dataSource); Connection c = pooling.getConnection()) { assertFalse(c.isClosed()); }
