[jboss-cvs] JBossAS SVN: r114834 - in branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector: src/main/org/jboss/resource/adapter/jdbc and 2 other directories.
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Fri Feb 13 00:36:48 EST 2015
Author: dstephan
Date: 2015-02-13 00:36:48 -0500 (Fri, 13 Feb 2015)
New Revision: 114834
Added:
branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/ReentrantLock.java
branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jms/ReentrantLock.java
Modified:
branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/build.xml
branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/BaseWrapperManagedConnection.java
branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/WrappedConnection.java
branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/remote/WrapperDataSourceService.java
branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jms/JmsManagedConnection.java
branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jms/JmsSession.java
Log:
JBPAPP-11214 - Backport multiple multi-threaded JCA issues ( JBPAPP-5596, JBPAPP-6684, JBPAPP-7102 )
Modified: branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/build.xml
===================================================================
--- branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/build.xml 2015-02-12 18:08:11 UTC (rev 114833)
+++ branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/build.xml 2015-02-13 05:36:48 UTC (rev 114834)
@@ -354,6 +354,7 @@
<include name="org/jboss/resource/adapter/jdbc/CachedPreparedStatement.class"/>
<include name="org/jboss/resource/adapter/jdbc/PreparedStatementCache*.class"/>
<include name="org/jboss/resource/adapter/jdbc/StatementAccess.class"/>
+ <include name="org/jboss/resource/adapter/jdbc/ReentrantLock.class"/>
<include name="org/jboss/resource/adapter/jdbc/jdk5/*.class"/>
<include name="org/jboss/resource/adapter/jdbc/jdk6/*.class"/>
<include name="org/jboss/resource/adapter/jdbc/URLSelectorStrategy.class"/>
Modified: branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/BaseWrapperManagedConnection.java
===================================================================
--- branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/BaseWrapperManagedConnection.java 2015-02-12 18:08:11 UTC (rev 114833)
+++ branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/BaseWrapperManagedConnection.java 2015-02-13 05:36:48 UTC (rev 114834)
@@ -36,7 +36,6 @@
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.ReentrantLock;
import javax.resource.ResourceException;
import javax.resource.spi.ConnectionEvent;
@@ -54,709 +53,761 @@
/**
* BaseWrapperManagedConnection
- *
+ *
* @author <a href="mailto:d_jencks at users.sourceforge.net">David Jencks</a>
* @author <a href="mailto:adrian at jboss.com">Adrian Brock</a>
* @author <a href="mailto:weston.price at jboss.com>Weston Price</a>
* @version $Revision$
*/
-public abstract class BaseWrapperManagedConnection implements ManagedConnection , JBossConnectionStatistics
+public abstract class BaseWrapperManagedConnection implements ManagedConnection, JBossConnectionStatistics
{
- private static final WrappedConnectionFactory wrappedConnectionFactory;
-
- protected final BaseWrapperManagedConnectionFactory mcf;
+ private static final WrappedConnectionFactory wrappedConnectionFactory;
- protected final Connection con;
+ protected final BaseWrapperManagedConnectionFactory mcf;
- protected final Properties props;
+ protected final Connection con;
- private final int transactionIsolation;
+ protected final Properties props;
- private final boolean readOnly;
+ private final int transactionIsolation;
- private ReentrantLock lock = new ReentrantLock();
+ private final boolean readOnly;
- private final Collection cels = new ArrayList();
+ private ReentrantLock lock = new ReentrantLock(true);
- private final Set handles = new HashSet();
+ private final Collection cels = new ArrayList();
- private PreparedStatementCache psCache = null;
+ private final Set handles = new HashSet();
- protected final Object stateLock = new Object();
+ private PreparedStatementCache psCache = null;
- protected boolean inManagedTransaction = false;
+ protected final Object stateLock = new Object();
- protected AtomicBoolean inLocalTransaction = new AtomicBoolean(false);
+ protected boolean inManagedTransaction = false;
- protected boolean jdbcAutoCommit = true;
+ protected AtomicBoolean inLocalTransaction = new AtomicBoolean(false);
- protected boolean underlyingAutoCommit = true;
+ protected boolean jdbcAutoCommit = true;
- protected boolean jdbcReadOnly;
+ protected boolean underlyingAutoCommit = true;
- protected boolean underlyingReadOnly;
+ protected boolean jdbcReadOnly;
- protected int jdbcTransactionIsolation;
+ protected boolean underlyingReadOnly;
- protected boolean destroyed = false;
-
- static
- {
- Class connectionFactory = null;
- try
- {
- connectionFactory = BaseWrapperManagedConnection.class.forName("org.jboss.resource.adapter.jdbc.jdk6.WrappedConnectionFactoryJDK6");
- }
- catch (ClassNotFoundException e)
- {
- try
- {
- connectionFactory = BaseWrapperManagedConnection.class.forName("org.jboss.resource.adapter.jdbc.jdk5.WrappedConnectionFactoryJDK5");
- }
- catch (ClassNotFoundException ex)
- {
- throw new RuntimeException("Unabled to load wrapped connection factory", ex);
- }
- }
- try
- {
- wrappedConnectionFactory = (WrappedConnectionFactory) connectionFactory.newInstance();
- }
- catch (Exception e)
- {
- throw new RuntimeException("Error initializign connection factory", e);
- }
- }
+ protected int jdbcTransactionIsolation;
- public BaseWrapperManagedConnection(final BaseWrapperManagedConnectionFactory mcf, final Connection con,
- final Properties props, final int transactionIsolation, final int psCacheSize) throws SQLException
- {
- this.mcf = mcf;
- this.con = con;
- this.props = props;
+ protected boolean destroyed = false;
- if (psCacheSize > 0)
- psCache = new PreparedStatementCache(psCacheSize);
+ static
+ {
+ Class connectionFactory = null;
+ try
+ {
+ connectionFactory = BaseWrapperManagedConnection.class.forName("org.jboss.resource.adapter.jdbc.jdk6.WrappedConnectionFactoryJDK6");
+ }
+ catch (ClassNotFoundException e)
+ {
+ try
+ {
+ connectionFactory = BaseWrapperManagedConnection.class.forName("org.jboss.resource.adapter.jdbc.jdk5.WrappedConnectionFactoryJDK5");
+ }
+ catch (ClassNotFoundException ex)
+ {
+ throw new RuntimeException("Unabled to load wrapped connection factory", ex);
+ }
+ }
+ try
+ {
+ wrappedConnectionFactory = (WrappedConnectionFactory) connectionFactory.newInstance();
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException("Error initializign connection factory", e);
+ }
+ }
- if (transactionIsolation == -1)
- this.transactionIsolation = con.getTransactionIsolation();
+ public BaseWrapperManagedConnection(final BaseWrapperManagedConnectionFactory mcf, final Connection con, final Properties props,
+ final int transactionIsolation, final int psCacheSize) throws SQLException
+ {
+ this.mcf = mcf;
+ this.con = con;
+ this.props = props;
- else
- {
- this.transactionIsolation = transactionIsolation;
- con.setTransactionIsolation(transactionIsolation);
- }
+ if (psCacheSize > 0)
+ psCache = new PreparedStatementCache(psCacheSize);
- readOnly = con.isReadOnly();
+ if (transactionIsolation == -1)
+ this.transactionIsolation = con.getTransactionIsolation();
- if (mcf.getNewConnectionSQL() != null)
- {
- Statement s = con.createStatement();
- try
- {
- s.execute(mcf.getNewConnectionSQL());
- }
- finally
- {
- s.close();
- }
- }
+ else
+ {
+ this.transactionIsolation = transactionIsolation;
+ con.setTransactionIsolation(transactionIsolation);
+ }
- underlyingReadOnly = readOnly;
- jdbcReadOnly = readOnly;
- jdbcTransactionIsolation = this.transactionIsolation;
- }
+ readOnly = con.isReadOnly();
- public void addConnectionEventListener(ConnectionEventListener cel)
- {
- synchronized (cels)
- {
- cels.add(cel);
- }
- }
+ if (mcf.getNewConnectionSQL() != null)
+ {
+ Statement s = con.createStatement();
+ try
+ {
+ s.execute(mcf.getNewConnectionSQL());
+ }
+ finally
+ {
+ s.close();
+ }
+ }
- public void removeConnectionEventListener(ConnectionEventListener cel)
- {
- synchronized (cels)
- {
- cels.remove(cel);
- }
- }
+ underlyingReadOnly = readOnly;
+ jdbcReadOnly = readOnly;
+ jdbcTransactionIsolation = this.transactionIsolation;
+ }
- public void associateConnection(Object handle) throws ResourceException
- {
- if (!(handle instanceof WrappedConnection))
- throw new JBossResourceException("Wrong kind of connection handle to associate" + handle);
- ((WrappedConnection) handle).setManagedConnection(this);
- synchronized (handles)
- {
- handles.add(handle);
- }
- }
+ public void addConnectionEventListener(ConnectionEventListener cel)
+ {
+ synchronized (cels)
+ {
+ cels.add(cel);
+ }
+ }
- public PrintWriter getLogWriter() throws ResourceException
- {
- // TODO: implement this javax.resource.spi.ManagedConnection method
- return null;
- }
+ public void removeConnectionEventListener(ConnectionEventListener cel)
+ {
+ synchronized (cels)
+ {
+ cels.remove(cel);
+ }
+ }
- public ManagedConnectionMetaData getMetaData() throws ResourceException
- {
- // TODO: implement this javax.resource.spi.ManagedConnection method
- return null;
- }
+ public void associateConnection(Object handle) throws ResourceException
+ {
+ if (!(handle instanceof WrappedConnection))
+ throw new JBossResourceException("Wrong kind of connection handle to associate" + handle);
+ ((WrappedConnection) handle).setManagedConnection(this);
+ synchronized (handles)
+ {
+ handles.add(handle);
+ }
+ }
- public void setLogWriter(PrintWriter param1) throws ResourceException
- {
- // TODO: implement this javax.resource.spi.ManagedConnection method
- }
+ public PrintWriter getLogWriter() throws ResourceException
+ {
+ // TODO: implement this javax.resource.spi.ManagedConnection method
+ return null;
+ }
- public void cleanup() throws ResourceException
- {
- synchronized (handles)
- {
- for (Iterator i = handles.iterator(); i.hasNext();)
- {
- WrappedConnection lc = (WrappedConnection) i.next();
- lc.setManagedConnection(null);
- }
- handles.clear();
- }
- //reset all the properties we know about to defaults.
- synchronized (stateLock)
- {
- jdbcAutoCommit = true;
- jdbcReadOnly = readOnly;
- if (jdbcTransactionIsolation != transactionIsolation)
- {
- try
- {
- con.setTransactionIsolation(transactionIsolation);
- jdbcTransactionIsolation = transactionIsolation;
- }
- catch (SQLException e)
- {
- mcf.log.warn("Error resetting transaction isolation ", e);
- }
- }
- }
- // I'm recreating the lock object when we return to the pool
- // because it looks too nasty to expect the connection handle
- // to unlock properly in certain race conditions
- // where the dissociation of the managed connection is "random".
- lock = new ReentrantLock();
- }
+ public ManagedConnectionMetaData getMetaData() throws ResourceException
+ {
+ // TODO: implement this javax.resource.spi.ManagedConnection method
+ return null;
+ }
- protected void lock()
- {
- lock.lock();
- }
+ public void setLogWriter(PrintWriter param1) throws ResourceException
+ {
+ // TODO: implement this javax.resource.spi.ManagedConnection method
+ }
- protected void tryLock() throws SQLException
- {
- int tryLock = mcf.getUseTryLock();
- if (tryLock <= 0)
- {
- lock();
- return;
- }
- try
- {
- if (lock.tryLock(tryLock, TimeUnit.SECONDS) == false)
- throw new SQLException("Unable to obtain lock in " + tryLock + " seconds: " + this);
- }
- catch (InterruptedException e)
- {
- throw new SQLException("Interrupted attempting lock: " + this);
- }
- }
-
- protected void unlock()
- {
- lock.unlock();
- }
+ public void cleanup() throws ResourceException
+ {
+ boolean isActive = false;
- public Object getConnection(Subject subject, ConnectionRequestInfo cri) throws ResourceException
- {
- checkIdentity(subject, cri);
- WrappedConnection lc = wrappedConnectionFactory.createWrappedConnection(this);
- synchronized (handles)
- {
- handles.add(lc);
- }
- return lc;
- }
+ if (lock.hasQueuedThreads())
+ {
+ Collection<Thread> threads = lock.getQueuedThreads();
+ for (Thread thread : threads)
+ {
+ Throwable t = new Throwable("Thread waiting for lock during cleanup");
+ t.setStackTrace(thread.getStackTrace());
- public void destroy() throws ResourceException
- {
- synchronized (stateLock)
- {
- destroyed = true;
- }
+ mcf.log.warn(t.getMessage(), t);
+ }
- cleanup();
- try
- {
- // See JBAS-5678
- if (underlyingAutoCommit == false)
- con.rollback();
- }
- catch (SQLException ignored)
- {
- getLog().trace("Ignored error during rollback: ", ignored);
- }
- try
- {
- con.close();
- }
- catch (SQLException ignored)
- {
- getLog().trace("Ignored error during close: ", ignored);
- }
- }
+ isActive = true;
+ }
- public boolean checkValid()
- {
- SQLException e = mcf.isValidConnection(con);
+ if (lock.isLocked())
+ {
+ Throwable t = new Throwable("Lock owned during cleanup");
+ t.setStackTrace(lock.getOwner().getStackTrace());
- if (e == null)
- // It's ok
- return true;
- else
- {
- getLog().warn("Destroying connection that is not valid, due to the following exception: " + con, e);
- broadcastConnectionError(e);
- return false;
- }
- }
+ mcf.log.warn(t.getMessage(), t);
- public Properties getProperties()
- {
- return this.props;
+ isActive = true;
+ }
+ synchronized (handles)
+ {
+ for (Iterator i = handles.iterator(); i.hasNext();)
+ {
+ WrappedConnection lc = (WrappedConnection) i.next();
+ lc.setManagedConnection(null);
+ }
- }
+ handles.clear();
- void closeHandle(WrappedConnection handle)
- {
- synchronized (stateLock)
- {
- if (destroyed)
- return;
- }
+ }
+ // reset all the properties we know about to defaults.
+ synchronized (stateLock)
+ {
+ jdbcAutoCommit = true;
+ jdbcReadOnly = readOnly;
+ if (jdbcTransactionIsolation != transactionIsolation)
+ {
+ try
+ {
+ con.setTransactionIsolation(transactionIsolation);
+ jdbcTransactionIsolation = transactionIsolation;
+ }
+ catch (SQLException e)
+ {
+ mcf.log.warn("Error resetting transaction isolation ", e);
+ }
+ }
+ }
- synchronized (handles)
- {
- handles.remove(handle);
- }
- ConnectionEvent ce = new ConnectionEvent(this, ConnectionEvent.CONNECTION_CLOSED);
- ce.setConnectionHandle(handle);
- Collection copy = null;
- synchronized (cels)
- {
- copy = new ArrayList(cels);
- }
- for (Iterator i = copy.iterator(); i.hasNext();)
- {
- ConnectionEventListener cel = (ConnectionEventListener) i.next();
- cel.connectionClosed(ce);
- }
- }
+ if (isActive)
+ {
+ // There are active lock - make sure that the JCA container kills
+ // this handle by throwing an exception
- Throwable connectionError(Throwable t)
- {
- if(t instanceof SQLException)
- {
- if(mcf.isStaleConnection((SQLException)t))
- {
- t = new StaleConnectionException((SQLException)t);
-
- }else
- {
- if(mcf.isExceptionFatal((SQLException)t))
- {
- broadcastConnectionError(t);
-
- }
- }
- }
- else
- {
- broadcastConnectionError(t);
- }
+ throw new ResourceException("Still active locks for " + this);
+ }
+ }
- return t;
- }
+ protected void lock()
+ {
+ lock.lock();
+ }
+ protected void tryLock() throws SQLException
+ {
+ int tryLock = mcf.getUseTryLock();
+ if (tryLock <= 0)
+ {
+ lock();
+ return;
+ }
+ try
+ {
+ if (lock.tryLock(tryLock, TimeUnit.SECONDS) == false)
+ throw new SQLException("Unable to obtain lock in " + tryLock + " seconds: " + this);
+ }
+ catch (InterruptedException e)
+ {
+ throw new SQLException("Interrupted attempting lock: " + this);
+ }
+ }
- protected void broadcastConnectionError(Throwable e)
- {
- synchronized (stateLock)
- {
- if (destroyed)
- {
- Logger log = getLog();
- if (log.isTraceEnabled())
- log.trace("Not broadcasting error, already destroyed " + this, e);
- return;
- }
- }
+ protected void unlock()
+ {
+ if (lock.isLocked())
+ {
+ lock.unlock();
+ }
+ else
+ {
+ mcf.log.warn("Owner is null");
+ Throwable t = new Throwable("Thread trying to unlock");
+ t.setStackTrace(Thread.currentThread().getStackTrace());
- Exception ex = null;
- if (e instanceof Exception)
- ex = (Exception) e;
- else
- ex = new ResourceAdapterInternalException("Unexpected error", e);
- ConnectionEvent ce = new ConnectionEvent(this, ConnectionEvent.CONNECTION_ERROR_OCCURRED, ex);
- Collection copy = null;
- synchronized (cels)
- {
- copy = new ArrayList(cels);
- }
- for (Iterator i = copy.iterator(); i.hasNext();)
- {
- ConnectionEventListener cel = (ConnectionEventListener) i.next();
- try
- {
- cel.connectionErrorOccurred(ce);
- }
- catch (Throwable t)
- {
- getLog().warn("Error notifying of connection error for listener: " + cel, t);
- }
- }
- }
+ mcf.log.warn(t.getMessage(), t);
+ }
- Connection getConnection() throws SQLException
- {
- if (con == null)
- throw new SQLException("Connection has been destroyed!!!");
- return con;
- }
+ }
- PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException
- {
- if (psCache != null)
- {
- PreparedStatementCache.Key key = new PreparedStatementCache.Key(sql,
- PreparedStatementCache.Key.PREPARED_STATEMENT, resultSetType, resultSetConcurrency);
- CachedPreparedStatement cachedps = (CachedPreparedStatement) psCache.get(key);
- if (cachedps != null)
- {
- if (canUse(cachedps))
- cachedps.inUse();
- else
- return doPrepareStatement(sql, resultSetType, resultSetConcurrency);
- }
- else
- {
- PreparedStatement ps = doPrepareStatement(sql, resultSetType, resultSetConcurrency);
- cachedps = wrappedConnectionFactory.createCachedPreparedStatement(ps);
- psCache.insert(key, cachedps);
- }
- return cachedps;
- }
- else
- return doPrepareStatement(sql, resultSetType, resultSetConcurrency);
- }
+ public Object getConnection(Subject subject, ConnectionRequestInfo cri) throws ResourceException
+ {
+ checkIdentity(subject, cri);
+ WrappedConnection lc = wrappedConnectionFactory.createWrappedConnection(this);
+ synchronized (handles)
+ {
+ handles.add(lc);
+ }
+ return lc;
+ }
- PreparedStatement doPrepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException
- {
- return con.prepareStatement(sql, resultSetType, resultSetConcurrency);
- }
+ public void destroy() throws ResourceException
+ {
+ synchronized (stateLock)
+ {
+ destroyed = true;
+ }
- CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException
- {
- if (psCache != null)
- {
- PreparedStatementCache.Key key = new PreparedStatementCache.Key(sql,
- PreparedStatementCache.Key.CALLABLE_STATEMENT, resultSetType, resultSetConcurrency);
- CachedCallableStatement cachedps = (CachedCallableStatement) psCache.get(key);
- if (cachedps != null)
- {
- if (canUse(cachedps))
- cachedps.inUse();
- else
- return doPrepareCall(sql, resultSetType, resultSetConcurrency);
- }
- else
- {
- CallableStatement cs = doPrepareCall(sql, resultSetType, resultSetConcurrency);
- cachedps = wrappedConnectionFactory.createCachedCallableStatement(cs);
- psCache.insert(key, cachedps);
- }
- return cachedps;
- }
- else
- return doPrepareCall(sql, resultSetType, resultSetConcurrency);
- }
+ cleanup();
+ try
+ {
+ // See JBAS-5678
+ if (underlyingAutoCommit == false)
+ con.rollback();
+ }
+ catch (SQLException ignored)
+ {
+ getLog().trace("Ignored error during rollback: ", ignored);
+ }
+ try
+ {
+ con.close();
+ }
+ catch (SQLException ignored)
+ {
+ getLog().trace("Ignored error during close: ", ignored);
+ }
+ }
- CallableStatement doPrepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException
- {
- return con.prepareCall(sql, resultSetType, resultSetConcurrency);
- }
+ public boolean checkValid()
+ {
+ SQLException e = mcf.isValidConnection(con);
- boolean canUse(CachedPreparedStatement cachedps)
- {
- // Nobody is using it so we are ok
- if (cachedps.isInUse() == false)
- return true;
+ if (e == null)
+ // It's ok
+ return true;
+ else
+ {
+ getLog().warn("Destroying connection that is not valid, due to the following exception: " + con, e);
+ broadcastConnectionError(e);
+ return false;
+ }
+ }
- // Cannot reuse prepared statements in auto commit mode
- // if will close the previous usage of the PS
- if (underlyingAutoCommit == true)
- return false;
+ public Properties getProperties()
+ {
+ return this.props;
- // We have been told not to share
- return mcf.sharePS;
- }
+ }
- protected Logger getLog()
- {
- return mcf.log;
- }
+ void closeHandle(WrappedConnection handle)
+ {
+ synchronized (stateLock)
+ {
+ if (destroyed)
+ return;
+ }
- private void checkIdentity(Subject subject, ConnectionRequestInfo cri) throws ResourceException
- {
- Properties newProps = mcf.getConnectionProperties(subject, cri);
- if (!props.equals(newProps))
- {
- throw new JBossResourceException("Wrong credentials passed to getConnection!");
- } // end of if ()
- }
+ synchronized (handles)
+ {
+ handles.remove(handle);
+ }
+ ConnectionEvent ce = new ConnectionEvent(this, ConnectionEvent.CONNECTION_CLOSED);
+ ce.setConnectionHandle(handle);
+ Collection copy = null;
+ synchronized (cels)
+ {
+ copy = new ArrayList(cels);
+ }
+ for (Iterator i = copy.iterator(); i.hasNext();)
+ {
+ ConnectionEventListener cel = (ConnectionEventListener) i.next();
+ cel.connectionClosed(ce);
+ }
+ }
- /**
- * The <code>checkTransaction</code> method makes sure the adapter follows the JCA
- * autocommit contract, namely all statements executed outside a container managed transaction
- * or a component managed transaction should be autocommitted. To avoid continually calling
- * setAutocommit(enable) before and after container managed transactions, we keep track of the state
- * and check it before each transactional method call.
- */
- void checkTransaction() throws SQLException
- {
- synchronized (stateLock)
- {
- if (inManagedTransaction)
- return;
+ Throwable connectionError(Throwable t)
+ {
+ if (t instanceof SQLException)
+ {
+ boolean stale = mcf.isStaleConnection((SQLException) t);
- // Check autocommit
- if (jdbcAutoCommit != underlyingAutoCommit)
- {
- con.setAutoCommit(jdbcAutoCommit);
- underlyingAutoCommit = jdbcAutoCommit;
- }
- }
+ if (stale)
+ {
+ t = new StaleConnectionException((SQLException) t);
- if (jdbcAutoCommit == false && inLocalTransaction.getAndSet(true) == false)
- {
- ArrayList copy;
- synchronized (cels)
- {
- copy = new ArrayList(cels);
- }
- ConnectionEvent ce = new ConnectionEvent(this, ConnectionEvent.LOCAL_TRANSACTION_STARTED);
- for (int i = 0; i < copy.size(); ++i)
- {
- ConnectionEventListener cel = (ConnectionEventListener) copy.get(i);
- try
- {
- cel.localTransactionStarted(ce);
- }
- catch (Throwable t)
- {
- getLog().trace("Error notifying of connection committed for listener: " + cel, t);
- }
- }
- }
+ }
+ else
+ {
+ boolean fatalException = mcf.isExceptionFatal((SQLException) t);
- checkState();
- }
+ if (fatalException)
+ {
+ broadcastConnectionError(t);
- protected void checkState() throws SQLException
- {
- synchronized (stateLock)
- {
- // Check readonly
- if (jdbcReadOnly != underlyingReadOnly)
- {
- con.setReadOnly(jdbcReadOnly);
- underlyingReadOnly = jdbcReadOnly;
- }
- }
- }
+ }
+ }
+ }
+ else
+ {
+ broadcastConnectionError(t);
+ }
- boolean isJdbcAutoCommit()
- {
- return inManagedTransaction ? false : jdbcAutoCommit;
- }
+ return t;
+ }
- void setJdbcAutoCommit(final boolean jdbcAutoCommit) throws SQLException
- {
- synchronized (stateLock)
- {
- if (inManagedTransaction)
- throw new SQLException("You cannot set autocommit during a managed transaction!");
- this.jdbcAutoCommit = jdbcAutoCommit;
- }
+ protected void broadcastConnectionError(Throwable e)
+ {
+ synchronized (stateLock)
+ {
+ if (destroyed)
+ {
+ Logger log = getLog();
+ if (log.isTraceEnabled())
+ log.trace("Not broadcasting error, already destroyed " + this, e);
+ return;
+ }
+ }
- if (jdbcAutoCommit && inLocalTransaction.getAndSet(false))
- {
- ArrayList copy;
- synchronized (cels)
- {
- copy = new ArrayList(cels);
- }
- ConnectionEvent ce = new ConnectionEvent(this, ConnectionEvent.LOCAL_TRANSACTION_COMMITTED);
- for (int i = 0; i < copy.size(); ++i)
- {
- ConnectionEventListener cel = (ConnectionEventListener) copy.get(i);
- try
- {
- cel.localTransactionCommitted(ce);
- }
- catch (Throwable t)
- {
- getLog().trace("Error notifying of connection committed for listener: " + cel, t);
- }
- }
- }
- }
+ // We need to unlock() before sending the connection error to the
+ // event listeners. Otherwise the lock won't be in sync once
+ // cleanup() is called
+ unlock();
- boolean isJdbcReadOnly()
- {
- return jdbcReadOnly;
- }
+ Exception ex = null;
+ if (e instanceof Exception)
+ ex = (Exception) e;
+ else
+ ex = new ResourceAdapterInternalException("Unexpected error", e);
+ ConnectionEvent ce = new ConnectionEvent(this, ConnectionEvent.CONNECTION_ERROR_OCCURRED, ex);
+ Collection copy = null;
+ synchronized (cels)
+ {
+ copy = new ArrayList(cels);
+ }
+ for (Iterator i = copy.iterator(); i.hasNext();)
+ {
+ ConnectionEventListener cel = (ConnectionEventListener) i.next();
+ try
+ {
+ cel.connectionErrorOccurred(ce);
+ }
+ catch (Throwable t)
+ {
+ getLog().warn("Error notifying of connection error for listener: " + cel, t);
+ }
+ }
+ }
- void setJdbcReadOnly(final boolean readOnly) throws SQLException
- {
- synchronized (stateLock)
- {
- if (inManagedTransaction)
- throw new SQLException("You cannot set read only during a managed transaction!");
- this.jdbcReadOnly = readOnly;
- }
- }
+ Connection getConnection() throws SQLException
+ {
+ if (con == null)
+ throw new SQLException("Connection has been destroyed!!!");
+ return con;
+ }
- int getJdbcTransactionIsolation()
- {
- return jdbcTransactionIsolation;
- }
+ PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException
+ {
+ if (psCache != null)
+ {
+ PreparedStatementCache.Key key = new PreparedStatementCache.Key(sql, PreparedStatementCache.Key.PREPARED_STATEMENT, resultSetType,
+ resultSetConcurrency);
+ CachedPreparedStatement cachedps = (CachedPreparedStatement) psCache.get(key);
+ if (cachedps != null)
+ {
+ if (canUse(cachedps))
+ cachedps.inUse();
+ else
+ return doPrepareStatement(sql, resultSetType, resultSetConcurrency);
+ }
+ else
+ {
+ PreparedStatement ps = doPrepareStatement(sql, resultSetType, resultSetConcurrency);
+ cachedps = wrappedConnectionFactory.createCachedPreparedStatement(ps);
+ psCache.insert(key, cachedps);
+ }
+ return cachedps;
+ }
+ else
+ return doPrepareStatement(sql, resultSetType, resultSetConcurrency);
+ }
- void setJdbcTransactionIsolation(final int isolationLevel) throws SQLException
- {
- synchronized (stateLock)
- {
- this.jdbcTransactionIsolation = isolationLevel;
- con.setTransactionIsolation(jdbcTransactionIsolation);
- }
- }
+ PreparedStatement doPrepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException
+ {
+ return con.prepareStatement(sql, resultSetType, resultSetConcurrency);
+ }
- void jdbcCommit() throws SQLException
- {
- synchronized (stateLock)
- {
- if (inManagedTransaction)
- throw new SQLException("You cannot commit during a managed transaction!");
- if (jdbcAutoCommit)
- throw new SQLException("You cannot commit with autocommit set!");
- }
- con.commit();
+ CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException
+ {
+ if (psCache != null)
+ {
+ PreparedStatementCache.Key key = new PreparedStatementCache.Key(sql, PreparedStatementCache.Key.CALLABLE_STATEMENT, resultSetType,
+ resultSetConcurrency);
+ CachedCallableStatement cachedps = (CachedCallableStatement) psCache.get(key);
+ if (cachedps != null)
+ {
+ if (canUse(cachedps))
+ cachedps.inUse();
+ else
+ return doPrepareCall(sql, resultSetType, resultSetConcurrency);
+ }
+ else
+ {
+ CallableStatement cs = doPrepareCall(sql, resultSetType, resultSetConcurrency);
+ cachedps = wrappedConnectionFactory.createCachedCallableStatement(cs);
+ psCache.insert(key, cachedps);
+ }
+ return cachedps;
+ }
+ else
+ return doPrepareCall(sql, resultSetType, resultSetConcurrency);
+ }
- if (inLocalTransaction.getAndSet(false))
- {
- ArrayList copy;
- synchronized (cels)
- {
- copy = new ArrayList(cels);
- }
- ConnectionEvent ce = new ConnectionEvent(this, ConnectionEvent.LOCAL_TRANSACTION_COMMITTED);
- for (int i = 0; i < copy.size(); ++i)
- {
- ConnectionEventListener cel = (ConnectionEventListener) copy.get(i);
- try
- {
- cel.localTransactionCommitted(ce);
- }
- catch (Throwable t)
- {
- getLog().trace("Error notifying of connection committed for listener: " + cel, t);
- }
- }
- }
- }
+ CallableStatement doPrepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException
+ {
+ return con.prepareCall(sql, resultSetType, resultSetConcurrency);
+ }
- void jdbcRollback() throws SQLException
- {
- synchronized (stateLock)
- {
- if (inManagedTransaction)
- throw new SQLException("You cannot rollback during a managed transaction!");
- if (jdbcAutoCommit)
- throw new SQLException("You cannot rollback with autocommit set!");
- }
- con.rollback();
+ boolean canUse(CachedPreparedStatement cachedps)
+ {
+ // Nobody is using it so we are ok
+ if (cachedps.isInUse() == false)
+ return true;
- if (inLocalTransaction.getAndSet(false))
- {
- ArrayList copy;
- synchronized (cels)
- {
- copy = new ArrayList(cels);
- }
- ConnectionEvent ce = new ConnectionEvent(this, ConnectionEvent.LOCAL_TRANSACTION_ROLLEDBACK);
- for (int i = 0; i < copy.size(); ++i)
- {
- ConnectionEventListener cel = (ConnectionEventListener) copy.get(i);
- try
- {
- cel.localTransactionRolledback(ce);
- }
- catch (Throwable t)
- {
- getLog().trace("Error notifying of connection rollback for listener: " + cel, t);
- }
- }
- }
- }
+ // Cannot reuse prepared statements in auto commit mode
+ // if will close the previous usage of the PS
+ if (underlyingAutoCommit == true)
+ return false;
- void jdbcRollback(Savepoint savepoint) throws SQLException
- {
- synchronized (stateLock)
- {
- if (inManagedTransaction)
- throw new SQLException("You cannot rollback during a managed transaction!");
- if (jdbcAutoCommit)
- throw new SQLException("You cannot rollback with autocommit set!");
- }
- con.rollback(savepoint);
- }
+ // We have been told not to share
+ return mcf.sharePS;
+ }
- int getTrackStatements()
- {
- return mcf.trackStatements;
- }
+ protected Logger getLog()
+ {
+ return mcf.log;
+ }
- boolean isTransactionQueryTimeout()
- {
- return mcf.isTransactionQueryTimeout;
- }
+ private void checkIdentity(Subject subject, ConnectionRequestInfo cri) throws ResourceException
+ {
+ Properties newProps = mcf.getConnectionProperties(subject, cri);
+ if (!props.equals(newProps))
+ {
+ throw new JBossResourceException("Wrong credentials passed to getConnection!");
+ } // end of if ()
+ }
- int getQueryTimeout()
- {
- return mcf.getQueryTimeout();
- }
+ /**
+ * The <code>checkTransaction</code> method makes sure the adapter follows
+ * the JCA autocommit contract, namely all statements executed outside a
+ * container managed transaction or a component managed transaction should
+ * be autocommitted. To avoid continually calling setAutocommit(enable)
+ * before and after container managed transactions, we keep track of the
+ * state and check it before each transactional method call.
+ */
+ void checkTransaction() throws SQLException
+ {
+ synchronized (stateLock)
+ {
+ if (inManagedTransaction)
+ return;
- protected void checkException(SQLException e) throws ResourceException
- {
- connectionError(e);
- throw new JBossResourceException("SQLException", e);
- }
-
- public Object listConnectionStats()
- {
- if(psCache != null)
- return psCache.toString();
- else
- return "-1"; //-1 indicates NoCache
- }
+ // Check autocommit
+ if (jdbcAutoCommit != underlyingAutoCommit)
+ {
+ con.setAutoCommit(jdbcAutoCommit);
+ underlyingAutoCommit = jdbcAutoCommit;
+ }
+ }
+
+ if (jdbcAutoCommit == false && inLocalTransaction.getAndSet(true) == false)
+ {
+ ArrayList copy;
+ synchronized (cels)
+ {
+ copy = new ArrayList(cels);
+ }
+ ConnectionEvent ce = new ConnectionEvent(this, ConnectionEvent.LOCAL_TRANSACTION_STARTED);
+ for (int i = 0; i < copy.size(); ++i)
+ {
+ ConnectionEventListener cel = (ConnectionEventListener) copy.get(i);
+ try
+ {
+ cel.localTransactionStarted(ce);
+ }
+ catch (Throwable t)
+ {
+ getLog().trace("Error notifying of connection committed for listener: " + cel, t);
+ }
+ }
+ }
+
+ checkState();
+ }
+
+ protected void checkState() throws SQLException
+ {
+ synchronized (stateLock)
+ {
+ // Check readonly
+ if (jdbcReadOnly != underlyingReadOnly)
+ {
+ con.setReadOnly(jdbcReadOnly);
+ underlyingReadOnly = jdbcReadOnly;
+ }
+ }
+ }
+
+ boolean isJdbcAutoCommit()
+ {
+ return inManagedTransaction ? false : jdbcAutoCommit;
+ }
+
+ void setJdbcAutoCommit(final boolean jdbcAutoCommit) throws SQLException
+ {
+ synchronized (stateLock)
+ {
+ if (inManagedTransaction)
+ throw new SQLException("You cannot set autocommit during a managed transaction!");
+ this.jdbcAutoCommit = jdbcAutoCommit;
+ }
+
+ if (jdbcAutoCommit && inLocalTransaction.getAndSet(false))
+ {
+ ArrayList copy;
+ synchronized (cels)
+ {
+ copy = new ArrayList(cels);
+ }
+ ConnectionEvent ce = new ConnectionEvent(this, ConnectionEvent.LOCAL_TRANSACTION_COMMITTED);
+ for (int i = 0; i < copy.size(); ++i)
+ {
+ ConnectionEventListener cel = (ConnectionEventListener) copy.get(i);
+ try
+ {
+ cel.localTransactionCommitted(ce);
+ }
+ catch (Throwable t)
+ {
+ getLog().trace("Error notifying of connection committed for listener: " + cel, t);
+ }
+ }
+ }
+ }
+
+ boolean isJdbcReadOnly()
+ {
+ return jdbcReadOnly;
+ }
+
+ void setJdbcReadOnly(final boolean readOnly) throws SQLException
+ {
+ synchronized (stateLock)
+ {
+ if (inManagedTransaction)
+ throw new SQLException("You cannot set read only during a managed transaction!");
+ this.jdbcReadOnly = readOnly;
+ }
+ }
+
+ int getJdbcTransactionIsolation()
+ {
+ return jdbcTransactionIsolation;
+ }
+
+ void setJdbcTransactionIsolation(final int isolationLevel) throws SQLException
+ {
+ synchronized (stateLock)
+ {
+ this.jdbcTransactionIsolation = isolationLevel;
+ con.setTransactionIsolation(jdbcTransactionIsolation);
+ }
+ }
+
+ void jdbcCommit() throws SQLException
+ {
+ synchronized (stateLock)
+ {
+ if (inManagedTransaction)
+ throw new SQLException("You cannot commit during a managed transaction!");
+ if (jdbcAutoCommit)
+ throw new SQLException("You cannot commit with autocommit set!");
+ }
+ con.commit();
+
+ if (inLocalTransaction.getAndSet(false))
+ {
+ ArrayList copy;
+ synchronized (cels)
+ {
+ copy = new ArrayList(cels);
+ }
+ ConnectionEvent ce = new ConnectionEvent(this, ConnectionEvent.LOCAL_TRANSACTION_COMMITTED);
+ for (int i = 0; i < copy.size(); ++i)
+ {
+ ConnectionEventListener cel = (ConnectionEventListener) copy.get(i);
+ try
+ {
+ cel.localTransactionCommitted(ce);
+ }
+ catch (Throwable t)
+ {
+ getLog().trace("Error notifying of connection committed for listener: " + cel, t);
+ }
+ }
+ }
+ }
+
+ void jdbcRollback() throws SQLException
+ {
+ synchronized (stateLock)
+ {
+ if (inManagedTransaction)
+ throw new SQLException("You cannot rollback during a managed transaction!");
+ if (jdbcAutoCommit)
+ throw new SQLException("You cannot rollback with autocommit set!");
+ }
+ con.rollback();
+
+ if (inLocalTransaction.getAndSet(false))
+ {
+ ArrayList copy;
+ synchronized (cels)
+ {
+ copy = new ArrayList(cels);
+ }
+ ConnectionEvent ce = new ConnectionEvent(this, ConnectionEvent.LOCAL_TRANSACTION_ROLLEDBACK);
+ for (int i = 0; i < copy.size(); ++i)
+ {
+ ConnectionEventListener cel = (ConnectionEventListener) copy.get(i);
+ try
+ {
+ cel.localTransactionRolledback(ce);
+ }
+ catch (Throwable t)
+ {
+ getLog().trace("Error notifying of connection rollback for listener: " + cel, t);
+ }
+ }
+ }
+ }
+
+ void jdbcRollback(Savepoint savepoint) throws SQLException
+ {
+ synchronized (stateLock)
+ {
+ if (inManagedTransaction)
+ throw new SQLException("You cannot rollback during a managed transaction!");
+ if (jdbcAutoCommit)
+ throw new SQLException("You cannot rollback with autocommit set!");
+ }
+ con.rollback(savepoint);
+ }
+
+ int getTrackStatements()
+ {
+ return mcf.trackStatements;
+ }
+
+ boolean isTransactionQueryTimeout()
+ {
+ return mcf.isTransactionQueryTimeout;
+ }
+
+ int getQueryTimeout()
+ {
+ return mcf.getQueryTimeout();
+ }
+
+ protected void checkException(SQLException e) throws ResourceException
+ {
+ connectionError(e);
+ throw new JBossResourceException("SQLException", e);
+ }
+
+ public Object listConnectionStats()
+ {
+ if (psCache != null)
+ return psCache.toString();
+ else
+ return "-1"; // -1 indicates NoCache
+ }
}
Added: branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/ReentrantLock.java
===================================================================
--- branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/ReentrantLock.java (rev 0)
+++ branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/ReentrantLock.java 2015-02-13 05:36:48 UTC (rev 114834)
@@ -0,0 +1,59 @@
+/*
+ * JBoss, Home of Professional Open Source.
+ * Copyright 2006, Red Hat Middleware LLC, and individual contributors
+ * as indicated by the @author tags. See the copyright.txt file in the
+ * distribution for a full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.jboss.resource.adapter.jdbc;
+
+import java.util.Collection;
+
+/**
+ * ReentrantLock override
+ * @author <a href="mailto:jesper.pedersen at jboss.org">Jesper Pedersen</a>
+ */
+public class ReentrantLock extends java.util.concurrent.locks.ReentrantLock
+{
+ /** Serial version uid */
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * Constructor
+ * @param fair Fair locking
+ */
+ public ReentrantLock(boolean fair)
+ {
+ super(fair);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public Thread getOwner()
+ {
+ return super.getOwner();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public Collection<Thread> getQueuedThreads()
+ {
+ return super.getQueuedThreads();
+ }
+}
\ No newline at end of file
Modified: branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/WrappedConnection.java
===================================================================
--- branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/WrappedConnection.java 2015-02-12 18:08:11 UTC (rev 114833)
+++ branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/WrappedConnection.java 2015-02-13 05:36:48 UTC (rev 114834)
@@ -47,893 +47,906 @@
*/
public abstract class WrappedConnection extends JBossWrapper implements Connection
{
- private static final Logger log = Logger.getLogger(WrappedConnection.class);
+ private static final Logger log = Logger.getLogger(WrappedConnection.class);
- private BaseWrapperManagedConnection mc;
+ private volatile BaseWrapperManagedConnection mc;
- private WrapperDataSource dataSource;
-
- private HashMap<WrappedStatement, Throwable> statements;
+ private BaseWrapperManagedConnection lockedMC;
- private boolean closed = false;
+ private int lockCount;
- private int trackStatements;
-
- public WrappedConnection(final BaseWrapperManagedConnection mc)
- {
- this.mc = mc;
- if (mc != null)
- trackStatements = mc.getTrackStatements();
- }
+ private WrapperDataSource dataSource;
- void setManagedConnection(final BaseWrapperManagedConnection mc)
- {
- this.mc = mc;
- if (mc != null)
- trackStatements = mc.getTrackStatements();
- }
+ private HashMap<WrappedStatement, Throwable> statements;
- protected void lock() throws SQLException
- {
- BaseWrapperManagedConnection mc = this.mc;
- if (mc != null)
- mc.tryLock();
- else
- throw new SQLException("Connection is not associated with a managed connection." + this);
- }
+ private boolean closed = false;
- protected void unlock()
- {
- BaseWrapperManagedConnection mc = this.mc;
- if (mc != null)
- mc.unlock();
- // We recreate the lock when returned to the pool
- // so missing the unlock after disassociation is not important
- }
+ private int trackStatements;
- public WrapperDataSource getDataSource()
- {
- return dataSource;
- }
-
- protected void setDataSource(WrapperDataSource dataSource)
- {
- this.dataSource = dataSource;
- }
-
- public void setReadOnly(boolean readOnly) throws SQLException
- {
- lock();
- try
- {
- checkStatus();
- mc.setJdbcReadOnly(readOnly);
- }
- finally
- {
- unlock();
- }
- }
+ public WrappedConnection(final BaseWrapperManagedConnection mc)
+ {
+ setManagedConnection(mc);
+ }
- public boolean isReadOnly() throws SQLException
- {
- checkStatus();
- return mc.isJdbcReadOnly();
- }
+ void setManagedConnection(final BaseWrapperManagedConnection mc)
+ {
+ this.mc = mc;
+ this.lockCount = 0;
- public void close() throws SQLException
- {
- closed = true;
- if (mc != null)
- {
- if (trackStatements != BaseWrapperManagedConnectionFactory.TRACK_STATEMENTS_FALSE_INT)
- {
- synchronized (this)
- {
- if (statements != null)
- {
- for (Iterator<Map.Entry<WrappedStatement, Throwable>> i = statements.entrySet().iterator(); i.hasNext(); )
- {
- Map.Entry<WrappedStatement, Throwable> entry = i.next();
- WrappedStatement ws = entry.getKey();
- if (trackStatements == BaseWrapperManagedConnectionFactory.TRACK_STATEMENTS_TRUE_INT)
- {
- Throwable stackTrace = entry.getValue();
- log.warn("Closing a statement you left open, please do your own housekeeping", stackTrace);
- }
- try
- {
- ws.internalClose();
- }
- catch (Throwable t)
- {
- log.warn("Exception trying to close statement:", t);
- }
- }
- }
- }
- }
- mc.closeHandle(this);
- }
- mc = null;
- dataSource = null;
- }
+ if (mc != null)
+ {
+ trackStatements = mc.getTrackStatements();
+ }
+ else
+ {
+ // Reset the lockedMC reference when returned to the pool
+ lockedMC = null;
+ }
+ }
- public boolean isClosed() throws SQLException
- {
- return closed;
- }
+ protected void lock() throws SQLException
+ {
+ BaseWrapperManagedConnection mc = this.mc;
+ if (mc != null)
+ {
+ mc.tryLock();
+ if (lockedMC == null)
+ lockedMC = mc;
- protected abstract WrappedStatement wrapStatement(Statement statement);
-
- public Statement createStatement() throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return wrapStatement(mc.getConnection().createStatement());
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ lockCount++;
+ }
+ else
+ throw new SQLException("Connection is not associated with a managed connection." + this);
+ }
- public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return wrapStatement(mc.getConnection().createStatement(resultSetType, resultSetConcurrency));
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ protected void unlock()
+ {
+ BaseWrapperManagedConnection mc = this.lockedMC;
+ if (--lockCount == 0)
+ lockedMC = null;
+ if (mc != null)
+ mc.unlock();
+ }
- public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability)
- throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return wrapStatement(mc.getConnection()
- .createStatement(resultSetType, resultSetConcurrency, resultSetHoldability));
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public WrapperDataSource getDataSource()
+ {
+ return dataSource;
+ }
- protected abstract WrappedPreparedStatement wrapPreparedStatement(PreparedStatement statement);
+ protected void setDataSource(WrapperDataSource dataSource)
+ {
+ this.dataSource = dataSource;
+ }
- public PreparedStatement prepareStatement(String sql) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return wrapPreparedStatement(mc.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY));
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public void setReadOnly(boolean readOnly) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkStatus();
+ mc.setJdbcReadOnly(readOnly);
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency)
- throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return wrapPreparedStatement(mc.prepareStatement(sql, resultSetType, resultSetConcurrency));
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public boolean isReadOnly() throws SQLException
+ {
+ checkStatus();
+ return mc.isJdbcReadOnly();
+ }
- public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency,
- int resultSetHoldability) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return wrapPreparedStatement(mc.getConnection()
- .prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability));
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public void close() throws SQLException
+ {
+ closed = true;
+ if (mc != null)
+ {
+ if (trackStatements != BaseWrapperManagedConnectionFactory.TRACK_STATEMENTS_FALSE_INT)
+ {
+ synchronized (this)
+ {
+ if (statements != null)
+ {
+ for (Iterator<Map.Entry<WrappedStatement, Throwable>> i = statements.entrySet().iterator(); i.hasNext();)
+ {
+ Map.Entry<WrappedStatement, Throwable> entry = i.next();
+ WrappedStatement ws = entry.getKey();
+ if (trackStatements == BaseWrapperManagedConnectionFactory.TRACK_STATEMENTS_TRUE_INT)
+ {
+ Throwable stackTrace = entry.getValue();
+ log.warn("Closing a statement you left open, please do your own housekeeping", stackTrace);
+ }
+ try
+ {
+ ws.internalClose();
+ }
+ catch (Throwable t)
+ {
+ log.warn("Exception trying to close statement:", t);
+ }
+ }
+ }
+ }
+ }
+ mc.closeHandle(this);
+ }
+ mc = null;
+ dataSource = null;
+ }
- public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return wrapPreparedStatement(mc.getConnection().prepareStatement(sql, autoGeneratedKeys));
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public boolean isClosed() throws SQLException
+ {
+ return closed;
+ }
- public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return wrapPreparedStatement(mc.getConnection().prepareStatement(sql, columnIndexes));
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ protected abstract WrappedStatement wrapStatement(Statement statement);
- public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return wrapPreparedStatement(mc.getConnection().prepareStatement(sql, columnNames));
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public Statement createStatement() throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return wrapStatement(mc.getConnection().createStatement());
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- protected abstract WrappedCallableStatement wrapCallableStatement(CallableStatement statement);
+ public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return wrapStatement(mc.getConnection().createStatement(resultSetType, resultSetConcurrency));
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public CallableStatement prepareCall(String sql) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return wrapCallableStatement(mc.prepareCall(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY));
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return wrapStatement(mc.getConnection().createStatement(resultSetType, resultSetConcurrency, resultSetHoldability));
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return wrapCallableStatement(mc.prepareCall(sql, resultSetType, resultSetConcurrency));
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ protected abstract WrappedPreparedStatement wrapPreparedStatement(PreparedStatement statement);
- public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency,
- int resultSetHoldability) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return wrapCallableStatement(mc.getConnection()
- .prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability));
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public PreparedStatement prepareStatement(String sql) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return wrapPreparedStatement(mc.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY));
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public String nativeSQL(String sql) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return mc.getConnection().nativeSQL(sql);
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return wrapPreparedStatement(mc.prepareStatement(sql, resultSetType, resultSetConcurrency));
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public void setAutoCommit(boolean autocommit) throws SQLException
- {
- lock();
- try
- {
- checkStatus();
- mc.setJdbcAutoCommit(autocommit);
- }
- finally
- {
- unlock();
- }
- }
+ public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return wrapPreparedStatement(mc.getConnection().prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability));
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public boolean getAutoCommit() throws SQLException
- {
- lock();
- try
- {
- checkStatus();
- return mc.isJdbcAutoCommit();
- }
- finally
- {
- unlock();
- }
- }
+ public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return wrapPreparedStatement(mc.getConnection().prepareStatement(sql, autoGeneratedKeys));
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public void commit() throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- mc.jdbcCommit();
- }
- finally
- {
- unlock();
- }
- }
+ public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return wrapPreparedStatement(mc.getConnection().prepareStatement(sql, columnIndexes));
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public void rollback() throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- mc.jdbcRollback();
- }
- finally
- {
- unlock();
- }
- }
+ public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return wrapPreparedStatement(mc.getConnection().prepareStatement(sql, columnNames));
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public void rollback(Savepoint savepoint) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- mc.jdbcRollback(savepoint);
- }
- finally
- {
- unlock();
- }
- }
+ protected abstract WrappedCallableStatement wrapCallableStatement(CallableStatement statement);
- public DatabaseMetaData getMetaData() throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return mc.getConnection().getMetaData();
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public CallableStatement prepareCall(String sql) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return wrapCallableStatement(mc.prepareCall(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY));
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public void setCatalog(String catalog) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- mc.getConnection().setCatalog(catalog);
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return wrapCallableStatement(mc.prepareCall(sql, resultSetType, resultSetConcurrency));
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public String getCatalog() throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return mc.getConnection().getCatalog();
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return wrapCallableStatement(mc.getConnection().prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability));
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public void setTransactionIsolation(int isolationLevel) throws SQLException
- {
- lock();
- try
- {
- checkStatus();
- mc.setJdbcTransactionIsolation(isolationLevel);
- }
- finally
- {
- unlock();
- }
- }
+ public String nativeSQL(String sql) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return mc.getConnection().nativeSQL(sql);
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public int getTransactionIsolation() throws SQLException
- {
- lock();
- try
- {
- checkStatus();
- return mc.getJdbcTransactionIsolation();
- }
- finally
- {
- unlock();
- }
- }
+ public void setAutoCommit(boolean autocommit) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkStatus();
+ mc.setJdbcAutoCommit(autocommit);
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public SQLWarning getWarnings() throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return mc.getConnection().getWarnings();
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public boolean getAutoCommit() throws SQLException
+ {
+ lock();
+ try
+ {
+ checkStatus();
+ return mc.isJdbcAutoCommit();
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public void clearWarnings() throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- mc.getConnection().clearWarnings();
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public void commit() throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ mc.jdbcCommit();
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- @SuppressWarnings("unchecked")
- public Map getTypeMap() throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return mc.getConnection().getTypeMap();
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public void rollback() throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ mc.jdbcRollback();
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- @SuppressWarnings("unchecked")
- public void setTypeMap(Map typeMap) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- mc.getConnection().setTypeMap(typeMap);
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public void rollback(Savepoint savepoint) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ mc.jdbcRollback(savepoint);
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public void setHoldability(int holdability) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- mc.getConnection().setHoldability(holdability);
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public DatabaseMetaData getMetaData() throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return mc.getConnection().getMetaData();
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public int getHoldability() throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return mc.getConnection().getHoldability();
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public void setCatalog(String catalog) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ mc.getConnection().setCatalog(catalog);
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public Savepoint setSavepoint() throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return mc.getConnection().setSavepoint();
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public String getCatalog() throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return mc.getConnection().getCatalog();
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public Savepoint setSavepoint(String name) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- return mc.getConnection().setSavepoint(name);
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public void setTransactionIsolation(int isolationLevel) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkStatus();
+ mc.setJdbcTransactionIsolation(isolationLevel);
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public void releaseSavepoint(Savepoint savepoint) throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- try
- {
- mc.getConnection().releaseSavepoint(savepoint);
- }
- catch (Throwable t)
- {
- throw checkException(t);
- }
- }
- finally
- {
- unlock();
- }
- }
+ public int getTransactionIsolation() throws SQLException
+ {
+ lock();
+ try
+ {
+ checkStatus();
+ return mc.getJdbcTransactionIsolation();
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public Connection getUnderlyingConnection() throws SQLException
- {
- lock();
- try
- {
- checkTransaction();
- return mc.getConnection();
- }
- finally
- {
- unlock();
- }
- }
+ public SQLWarning getWarnings() throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return mc.getConnection().getWarnings();
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- protected Connection getWrappedObject() throws SQLException
- {
- return getUnderlyingConnection();
- }
+ public void clearWarnings() throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ mc.getConnection().clearWarnings();
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- protected void checkTransaction() throws SQLException
- {
- checkStatus();
- mc.checkTransaction();
- }
+ @SuppressWarnings("unchecked")
+ public Map getTypeMap() throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return mc.getConnection().getTypeMap();
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- void checkTransactionActive() throws SQLException
- {
- if (dataSource == null)
- return;
- dataSource.checkTransactionActive();
- }
+ @SuppressWarnings("unchecked")
+ public void setTypeMap(Map typeMap) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ mc.getConnection().setTypeMap(typeMap);
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- /**
- * The checkStatus method checks that the handle has not been closed and
- * that it is associated with a managed connection.
- *
- * @exception SQLException if an error occurs
- */
- protected void checkStatus() throws SQLException
- {
- if (closed)
- throw new SQLException("Connection handle has been closed and is unusable");
- if (mc == null)
- throw new SQLException("Connection handle is not currently associated with a ManagedConnection");
- checkTransactionActive();
- }
+ public void setHoldability(int holdability) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ mc.getConnection().setHoldability(holdability);
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- /**
- * The base checkException method rethrows the supplied exception, informing
- * the ManagedConnection of the error. Subclasses may override this to
- * filter exceptions based on their severity.
- *
- * @param t a throwable
- * @return the sql exception
- * @exception SQLException if an error occurs
- */
- protected SQLException checkException(Throwable t) throws SQLException
- {
- Throwable result = null;
- if (t instanceof AbstractMethodError)
- {
- t = new SQLFeatureNotSupportedException("Method is not implemented by JDBC driver", t);
- }
+ public int getHoldability() throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return mc.getConnection().getHoldability();
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- if (mc != null)
- result = mc.connectionError(t);
+ public Savepoint setSavepoint() throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return mc.getConnection().setSavepoint();
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- if (result instanceof SQLException)
- {
- throw (SQLException) result;
- }
- else
- {
- throw new NestedSQLException("Error", result);
- }
-
- }
+ public Savepoint setSavepoint(String name) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ return mc.getConnection().setSavepoint(name);
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- int getTrackStatements()
- {
- return trackStatements;
- }
-
- void registerStatement(WrappedStatement ws)
- {
- if (trackStatements == BaseWrapperManagedConnectionFactory.TRACK_STATEMENTS_FALSE_INT)
- return;
-
- synchronized (this)
- {
- if (statements == null)
- statements = new HashMap<WrappedStatement, Throwable>();
-
- if (trackStatements == BaseWrapperManagedConnectionFactory.TRACK_STATEMENTS_TRUE_INT)
- statements.put(ws, new Throwable("STACKTRACE"));
- else
- statements.put(ws, null);
- }
- }
+ public void releaseSavepoint(Savepoint savepoint) throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ try
+ {
+ mc.getConnection().releaseSavepoint(savepoint);
+ }
+ catch (Throwable t)
+ {
+ throw checkException(t);
+ }
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- void unregisterStatement(WrappedStatement ws)
- {
- if (trackStatements == BaseWrapperManagedConnectionFactory.TRACK_STATEMENTS_FALSE_INT)
- return;
- synchronized (this)
- {
- if (statements != null)
- statements.remove(ws);
- }
- }
+ public Connection getUnderlyingConnection() throws SQLException
+ {
+ lock();
+ try
+ {
+ checkTransaction();
+ return mc.getConnection();
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- void checkConfiguredQueryTimeout(WrappedStatement ws) throws SQLException
- {
- if (mc == null || dataSource == null)
- return;
+ protected Connection getWrappedObject() throws SQLException
+ {
+ return getUnderlyingConnection();
+ }
- int timeout = 0;
-
- // Use the transaction timeout
- if (mc.isTransactionQueryTimeout())
- timeout = dataSource.getTimeLeftBeforeTransactionTimeout();
-
- // Look for a configured value
- if (timeout <= 0)
- timeout = mc.getQueryTimeout();
-
- if (timeout > 0)
- ws.setQueryTimeout(timeout);
- }
-
- Logger getLogger()
- {
- return log;
- }
+ protected void checkTransaction() throws SQLException
+ {
+ checkStatus();
+ mc.checkTransaction();
+ }
+
+ void checkTransactionActive() throws SQLException
+ {
+ if (dataSource == null)
+ return;
+ dataSource.checkTransactionActive();
+ }
+
+ /**
+ * The checkStatus method checks that the handle has not been closed and
+ * that it is associated with a managed connection.
+ *
+ * @exception SQLException
+ * if an error occurs
+ */
+ protected void checkStatus() throws SQLException
+ {
+ if (closed)
+ throw new SQLException("Connection handle has been closed and is unusable");
+ if (mc == null)
+ throw new SQLException("Connection handle is not currently associated with a ManagedConnection");
+ checkTransactionActive();
+ }
+
+ /**
+ * The base checkException method rethrows the supplied exception, informing
+ * the ManagedConnection of the error. Subclasses may override this to
+ * filter exceptions based on their severity.
+ *
+ * @param t
+ * a throwable
+ * @return the sql exception
+ * @exception SQLException
+ * if an error occurs
+ */
+ protected SQLException checkException(Throwable t) throws SQLException
+ {
+ Throwable result = null;
+ if (t instanceof AbstractMethodError)
+ {
+ t = new SQLFeatureNotSupportedException("Method is not implemented by JDBC driver", t);
+ }
+
+ if (mc != null)
+ result = mc.connectionError(t);
+
+ if (result instanceof SQLException)
+ {
+ throw (SQLException) result;
+ }
+ else
+ {
+ throw new NestedSQLException("Error", result);
+ }
+
+ }
+
+ int getTrackStatements()
+ {
+ return trackStatements;
+ }
+
+ void registerStatement(WrappedStatement ws)
+ {
+ if (trackStatements == BaseWrapperManagedConnectionFactory.TRACK_STATEMENTS_FALSE_INT)
+ return;
+
+ synchronized (this)
+ {
+ if (statements == null)
+ statements = new HashMap<WrappedStatement, Throwable>();
+
+ if (trackStatements == BaseWrapperManagedConnectionFactory.TRACK_STATEMENTS_TRUE_INT)
+ statements.put(ws, new Throwable("STACKTRACE"));
+ else
+ statements.put(ws, null);
+ }
+ }
+
+ void unregisterStatement(WrappedStatement ws)
+ {
+ if (trackStatements == BaseWrapperManagedConnectionFactory.TRACK_STATEMENTS_FALSE_INT)
+ return;
+ synchronized (this)
+ {
+ if (statements != null)
+ statements.remove(ws);
+ }
+ }
+
+ void checkConfiguredQueryTimeout(WrappedStatement ws) throws SQLException
+ {
+ if (mc == null || dataSource == null)
+ return;
+
+ int timeout = 0;
+
+ // Use the transaction timeout
+ if (mc.isTransactionQueryTimeout())
+ timeout = dataSource.getTimeLeftBeforeTransactionTimeout();
+
+ // Look for a configured value
+ if (timeout <= 0)
+ timeout = mc.getQueryTimeout();
+
+ if (timeout > 0)
+ ws.setQueryTimeout(timeout);
+ }
+
+ Logger getLogger()
+ {
+ return log;
+ }
}
Modified: branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/remote/WrapperDataSourceService.java
===================================================================
--- branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/remote/WrapperDataSourceService.java 2015-02-12 18:08:11 UTC (rev 114833)
+++ branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jdbc/remote/WrapperDataSourceService.java 2015-02-13 05:36:48 UTC (rev 114834)
@@ -39,6 +39,7 @@
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
@@ -84,12 +85,12 @@
private ObjectName jmxInvokerName;
private Invoker delegateInvoker;
private Object theProxy;
- private HashMap marshalledInvocationMapping = new HashMap();
- private HashMap connectionMap = new HashMap();
- private HashMap statementMap = new HashMap();
- private HashMap resultSetMap = new HashMap();
- private HashMap lobMap = new HashMap();
- private HashMap databaseMetaDataMap = new HashMap();
+ private Map marshalledInvocationMapping = new HashMap();
+ private Map connectionMap = Collections.synchronizedMap(new HashMap());
+ private Map statementMap = Collections.synchronizedMap(new HashMap());
+ private Map resultSetMap = Collections.synchronizedMap(new HashMap());
+ private Map lobMap = Collections.synchronizedMap(new HashMap());
+ private Map databaseMetaDataMap = Collections.synchronizedMap(new HashMap());
private boolean trace = log.isTraceEnabled();
protected void startService() throws Exception
Modified: branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jms/JmsManagedConnection.java
===================================================================
--- branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jms/JmsManagedConnection.java 2015-02-12 18:08:11 UTC (rev 114833)
+++ branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jms/JmsManagedConnection.java 2015-02-13 05:36:48 UTC (rev 114834)
@@ -22,13 +22,13 @@
package org.jboss.resource.adapter.jms;
import java.io.PrintWriter;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.Vector;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.ReentrantLock;
import javax.jms.Connection;
import javax.jms.ExceptionListener;
@@ -68,772 +68,821 @@
/**
* Managed Connection, manages one or more JMS sessions.
- *
- * <p>Every ManagedConnection will have a physical JMSConnection under the
- * hood. This may leave out several session, as specifyed in 5.5.4 Multiple
- * Connection Handles. Thread safe semantics is provided
- *
- * <p>Hm. If we are to follow the example in 6.11 this will not work. We would
- * have to use the SAME session. This means we will have to guard against
- * concurrent access. We use a stack, and only allowes the handle at the
- * top of the stack to do things.
- *
- * <p>As to transactions we some fairly hairy alternatives to handle:
- * XA - we get an XA. We may now only do transaction through the
- * XAResource, since a XASession MUST throw exceptions in commit etc. But
- * since XA support implies LocatTransaction support, we will have to use
- * the XAResource in the LocalTransaction class.
- * LocalTx - we get a normal session. The LocalTransaction will then work
- * against the normal session api.
- *
- * <p>An invokation of JMS MAY BE DONE in none transacted context. What do we
- * do then? How much should we leave to the user???
- *
- * <p>One possible solution is to use transactions any way, but under the hood.
- * If not LocalTransaction or XA has been aquired by the container, we have
- * to do the commit in send and publish. (CHECK is the container required
- * to get a XA every time it uses a managed connection? No its is not, only
- * at creation!)
- *
- * <p>Does this mean that a session one time may be used in a transacted env,
- * and another time in a not transacted.
- *
- * <p>Maybe we could have this simple rule:
- *
- * <p>If a user is going to use non trans:
+ *
+ * <p>
+ * Every ManagedConnection will have a physical JMSConnection under the hood.
+ * This may leave out several session, as specifyed in 5.5.4 Multiple Connection
+ * Handles. Thread safe semantics is provided
+ *
+ * <p>
+ * Hm. If we are to follow the example in 6.11 this will not work. We would have
+ * to use the SAME session. This means we will have to guard against concurrent
+ * access. We use a stack, and only allowes the handle at the top of the stack
+ * to do things.
+ *
+ * <p>
+ * As to transactions we some fairly hairy alternatives to handle: XA - we get
+ * an XA. We may now only do transaction through the XAResource, since a
+ * XASession MUST throw exceptions in commit etc. But since XA support implies
+ * LocatTransaction support, we will have to use the XAResource in the
+ * LocalTransaction class. LocalTx - we get a normal session. The
+ * LocalTransaction will then work against the normal session api.
+ *
+ * <p>
+ * An invokation of JMS MAY BE DONE in none transacted context. What do we do
+ * then? How much should we leave to the user???
+ *
+ * <p>
+ * One possible solution is to use transactions any way, but under the hood. If
+ * not LocalTransaction or XA has been aquired by the container, we have to do
+ * the commit in send and publish. (CHECK is the container required to get a XA
+ * every time it uses a managed connection? No its is not, only at creation!)
+ *
+ * <p>
+ * Does this mean that a session one time may be used in a transacted env, and
+ * another time in a not transacted.
+ *
+ * <p>
+ * Maybe we could have this simple rule:
+ *
+ * <p>
+ * If a user is going to use non trans:
* <ul>
* <li>mark that i ra deployment descr
* <li>Use a JmsProviderAdapter with non XA factorys
* <li>Mark session as non transacted (this defeats the purpose of specifying
* <li>trans attrinbutes in deploy descr NOT GOOD
* </ul>
- *
- * <p>From the JMS tutorial:
- * "When you create a session in an enterprise bean, the container ignores
- * the arguments you specify, because it manages all transactional
- * properties for enterprise beans."
- *
- * <p>And further:
- * "You do not specify a message acknowledgment mode when you create a
- * message-driven bean that uses container-managed transactions. The
- * container handles acknowledgment automatically."
- *
- * <p>On Session or Connection:
- * <p>From Tutorial:
- * "A JMS API resource is a JMS API connection or a JMS API session." But in
- * the J2EE spec only connection is considered a resource.
- *
- * <p>Not resolved: connectionErrorOccurred: it is verry hard to know from the
- * exceptions thrown if it is a connection error. Should we register an
- * ExceptionListener and mark al handles as errounous? And then let them
- * send the event and throw an exception?
- *
+ *
+ * <p>
+ * From the JMS tutorial: "When you create a session in an enterprise bean, the
+ * container ignores the arguments you specify, because it manages all
+ * transactional properties for enterprise beans."
+ *
+ * <p>
+ * And further: "You do not specify a message acknowledgment mode when you
+ * create a message-driven bean that uses container-managed transactions. The
+ * container handles acknowledgment automatically."
+ *
+ * <p>
+ * On Session or Connection:
+ * <p>
+ * From Tutorial:
+ * "A JMS API resource is a JMS API connection or a JMS API session." But in the
+ * J2EE spec only connection is considered a resource.
+ *
+ * <p>
+ * Not resolved: connectionErrorOccurred: it is verry hard to know from the
+ * exceptions thrown if it is a connection error. Should we register an
+ * ExceptionListener and mark al handles as errounous? And then let them send
+ * the event and throw an exception?
+ *
* @author <a href="mailto:peter.antman at tim.se">Peter Antman</a>.
* @author <a href="mailto:jason at planet57.com">Jason Dillon</a>
* @author <a href="mailto:adrian at jboss.com">Adrian Brock</a>
* @version $Revision$
*/
-public class JmsManagedConnection
- implements ManagedConnection, ExceptionListener
+public class JmsManagedConnection implements ManagedConnection, ExceptionListener
{
- private static final Logger log = Logger.getLogger(JmsManagedConnection.class);
+ private static final Logger log = Logger.getLogger(JmsManagedConnection.class);
- private JmsManagedConnectionFactory mcf;
- private JmsConnectionRequestInfo info;
- private String user;
- private String pwd;
- private boolean isDestroyed;
+ private JmsManagedConnectionFactory mcf;
+ private JmsConnectionRequestInfo info;
+ private String user;
+ private String pwd;
+ private boolean isDestroyed;
- private ReentrantLock lock = new ReentrantLock();
-
- // Physical JMS connection stuff
- private Connection con;
- private Session session;
- private TopicSession topicSession;
- private QueueSession queueSession;
- private XASession xaSession;
- private XATopicSession xaTopicSession;
- private XAQueueSession xaQueueSession;
- private XAResource xaResource;
- private boolean xaTransacted;
+ private ReentrantLock lock = new ReentrantLock(true);
- /** Holds all current JmsSession handles. */
- private Set handles = Collections.synchronizedSet(new HashSet());
+ // Physical JMS connection stuff
+ private Connection con;
+ private Session session;
+ private TopicSession topicSession;
+ private QueueSession queueSession;
+ private XASession xaSession;
+ private XATopicSession xaTopicSession;
+ private XAQueueSession xaQueueSession;
+ private XAResource xaResource;
+ private boolean xaTransacted;
- /** The event listeners */
- private Vector listeners = new Vector();
+ /** Holds all current JmsSession handles. */
+ private Set handles = Collections.synchronizedSet(new HashSet());
- /**
- * Create a <tt>JmsManagedConnection</tt>.
- *
- * @param mcf
- * @param info
- * @param user
- * @param pwd
- *
- * @throws ResourceException
- */
- public JmsManagedConnection(final JmsManagedConnectionFactory mcf,
- final ConnectionRequestInfo info,
- final String user,
- final String pwd)
- throws ResourceException
- {
- this.mcf = mcf;
+ /** The event listeners */
+ private Vector listeners = new Vector();
- // seem like its asking for trouble here
- this.info = (JmsConnectionRequestInfo)info;
- this.user = user;
- this.pwd = pwd;
+ /**
+ * Create a <tt>JmsManagedConnection</tt>.
+ *
+ * @param mcf
+ * @param info
+ * @param user
+ * @param pwd
+ *
+ * @throws ResourceException
+ */
+ public JmsManagedConnection(final JmsManagedConnectionFactory mcf, final ConnectionRequestInfo info, final String user, final String pwd)
+ throws ResourceException
+ {
+ this.mcf = mcf;
- try
- {
- setup();
- }
- catch (Throwable t)
- {
- try
- {
- destroy();
- }
- catch (Throwable ignored)
- {
- }
- JBossResourceException.rethrowAsResourceException("Error during setup", t);
- }
- }
+ // seem like its asking for trouble here
+ this.info = (JmsConnectionRequestInfo) info;
+ this.user = user;
+ this.pwd = pwd;
- //---- ManagedConnection API ----
+ try
+ {
+ setup();
+ }
+ catch (Throwable t)
+ {
+ try
+ {
+ destroy();
+ }
+ catch (Throwable ignored)
+ {
+ }
+ JBossResourceException.rethrowAsResourceException("Error during setup", t);
+ }
+ }
- /**
- * Get the physical connection handler.
- *
- * <p>This bummer will be called in two situations:
- * <ol>
- * <li>When a new mc has bean created and a connection is needed
- * <li>When an mc has been fetched from the pool (returned in match*)
- * </ol>
- *
- * <p>It may also be called multiple time without a cleanup, to support
- * connection sharing.
- *
- * @param subject
- * @param info
- * @return A new connection object.
- *
- * @throws ResourceException
- */
- public Object getConnection(final Subject subject,
- final ConnectionRequestInfo info)
- throws ResourceException
- {
- // Check user first
- JmsCred cred = JmsCred.getJmsCred(mcf,subject,info);
+ // ---- ManagedConnection API ----
- // Null users are allowed!
- if (user != null && !user.equals(cred.name))
- throw new SecurityException
- ("Password credentials not the same, reauthentication not allowed");
- if (cred.name != null && user == null) {
- throw new SecurityException
- ("Password credentials not the same, reauthentication not allowed");
- }
+ /**
+ * Get the physical connection handler.
+ *
+ * <p>
+ * This bummer will be called in two situations:
+ * <ol>
+ * <li>When a new mc has bean created and a connection is needed
+ * <li>When an mc has been fetched from the pool (returned in match*)
+ * </ol>
+ *
+ * <p>
+ * It may also be called multiple time without a cleanup, to support
+ * connection sharing.
+ *
+ * @param subject
+ * @param info
+ * @return A new connection object.
+ *
+ * @throws ResourceException
+ */
+ public Object getConnection(final Subject subject, final ConnectionRequestInfo info) throws ResourceException
+ {
+ // Check user first
+ JmsCred cred = JmsCred.getJmsCred(mcf, subject, info);
- user = cred.name; // Basically meaningless
+ // Null users are allowed!
+ if (user != null && !user.equals(cred.name))
+ throw new SecurityException("Password credentials not the same, reauthentication not allowed");
+ if (cred.name != null && user == null)
+ {
+ throw new SecurityException("Password credentials not the same, reauthentication not allowed");
+ }
- if (isDestroyed)
- throw new IllegalStateException("ManagedConnection already destroyd");
+ user = cred.name; // Basically meaningless
- // Create a handle
- JmsSession handle = new JmsSession(this, (JmsConnectionRequestInfo) info);
- handles.add(handle);
- return handle;
- }
+ if (isDestroyed)
+ throw new IllegalStateException("ManagedConnection already destroyd");
- /**
- * Destroy all handles.
- *
- * @throws ResourceException Failed to close one or more handles.
- */
- private void destroyHandles() throws ResourceException
- {
- try
- {
- if (con != null)
- con.stop();
- }
- catch (Throwable t)
- {
- log.trace("Ignored error stopping connection", t);
- }
-
- Iterator iter = handles.iterator();
- while (iter.hasNext())
- ((JmsSession)iter.next()).destroy();
+ // Create a handle
+ JmsSession handle = new JmsSession(this, (JmsConnectionRequestInfo) info);
+ handles.add(handle);
+ return handle;
+ }
- // clear the handles map
- handles.clear();
- }
+ /**
+ * Destroy all handles.
+ *
+ * @throws ResourceException
+ * Failed to close one or more handles.
+ */
+ private void destroyHandles() throws ResourceException
+ {
+ try
+ {
+ if (con != null)
+ con.stop();
+ }
+ catch (Throwable t)
+ {
+ log.trace("Ignored error stopping connection", t);
+ }
- /**
- * Destroy the physical connection.
- *
- * @throws ResourceException Could not property close the session and
- * connection.
- */
- public void destroy() throws ResourceException
- {
- if (isDestroyed || con == null) return;
+ Iterator iter = handles.iterator();
+ while (iter.hasNext())
+ ((JmsSession) iter.next()).destroy();
- isDestroyed = true;
+ // clear the handles map
+ handles.clear();
+ }
- try
- {
- con.setExceptionListener(null);
- }
- catch (JMSException e)
- {
- log.debug("Error unsetting the exception listener " + this, e);
- }
-
- // destory handles
- destroyHandles();
-
- try
- {
- // Close session and connection
- try
- {
- if (info.getType() == JmsConnectionFactory.TOPIC)
- {
- if (topicSession != null)
- topicSession.close();
- if (xaTransacted && xaTopicSession != null) {
- xaTopicSession.close();
- }
- }
- else if (info.getType() == JmsConnectionFactory.QUEUE)
- {
- if (queueSession != null)
- queueSession.close();
- if (xaTransacted && xaQueueSession != null)
- xaQueueSession.close();
- }
- else
- {
- if (session != null)
- session.close();
- if (xaTransacted && xaSession != null)
- xaSession.close();
- }
- }
- catch (JMSException e)
- {
- log.debug("Error closing session " +this, e);
- }
- con.close();
- }
- catch (Throwable e)
- {
- throw new JBossResourceException
- ("Could not properly close the session and connection", e);
- }
- }
+ /**
+ * Destroy the physical connection.
+ *
+ * @throws ResourceException
+ * Could not property close the session and connection.
+ */
+ public void destroy() throws ResourceException
+ {
+ if (isDestroyed || con == null)
+ return;
- /**
- * Cleans up the, from the spec
- * - The cleanup of ManagedConnection instance resets its client specific
- * state.
- *
- * Does that mean that autentication should be redone. FIXME
- */
- public void cleanup() throws ResourceException
- {
- if (isDestroyed)
- throw new IllegalStateException("ManagedConnection already destroyed");
+ isDestroyed = true;
- // destory handles
- destroyHandles();
+ try
+ {
+ con.setExceptionListener(null);
+ }
+ catch (JMSException e)
+ {
+ log.debug("Error unsetting the exception listener " + this, e);
+ }
- // I'm recreating the lock object when we return to the pool
- // because it looks too nasty to expect the connection handle
- // to unlock properly in certain race conditions
- // where the dissociation of the managed connection is "random".
- lock = new ReentrantLock();
- }
+ // destory handles
+ destroyHandles();
- /**
- * Move a handler from one mc to this one.
- *
- * @param obj An object of type JmsSession.
- *
- * @throws ResourceException Failed to associate connection.
- * @throws IllegalStateException ManagedConnection in an illegal state.
- */
- public void associateConnection(final Object obj)
- throws ResourceException
- {
- //
- // Should we check auth, ie user and pwd? FIXME
- //
+ try
+ {
+ // Close session and connection
+ try
+ {
+ if (info.getType() == JmsConnectionFactory.TOPIC)
+ {
+ if (topicSession != null)
+ topicSession.close();
+ if (xaTransacted && xaTopicSession != null)
+ {
+ xaTopicSession.close();
+ }
+ }
+ else if (info.getType() == JmsConnectionFactory.QUEUE)
+ {
+ if (queueSession != null)
+ queueSession.close();
+ if (xaTransacted && xaQueueSession != null)
+ xaQueueSession.close();
+ }
+ else
+ {
+ if (session != null)
+ session.close();
+ if (xaTransacted && xaSession != null)
+ xaSession.close();
+ }
+ }
+ catch (JMSException e)
+ {
+ log.debug("Error closing session " + this, e);
+ }
+ con.close();
+ }
+ catch (Throwable e)
+ {
+ throw new JBossResourceException("Could not properly close the session and connection", e);
+ }
+ }
- if (!isDestroyed && obj instanceof JmsSession)
- {
- JmsSession h = (JmsSession)obj;
- h.setManagedConnection(this);
- handles.add(h);
- }
- else
- throw new IllegalStateException
- ("ManagedConnection in an illegal state");
- }
+ /**
+ * Cleans up the, from the spec - The cleanup of ManagedConnection instance
+ * resets its client specific state.
+ *
+ * Does that mean that autentication should be redone. FIXME
+ */
+ public void cleanup() throws ResourceException
+ {
+ if (isDestroyed)
+ throw new IllegalStateException("ManagedConnection already destroyed");
- protected void lock()
- {
- lock.lock();
- }
+ // destory handles
+ destroyHandles();
- protected void tryLock() throws JMSException
- {
- int tryLock = mcf.getUseTryLock();
- if (tryLock <= 0)
- {
- lock();
- return;
- }
- try
- {
- if (lock.tryLock(tryLock, TimeUnit.SECONDS) == false)
- throw new ResourceAllocationException("Unable to obtain lock in " + tryLock + " seconds: " + this);
- }
- catch (InterruptedException e)
- {
- throw new ResourceAllocationException("Interrupted attempting lock: " + this);
- }
- }
-
- protected void unlock()
- {
- lock.unlock();
- }
+ boolean isActive = false;
- /**
- * Add a connection event listener.
- *
- * @param l The connection event listener to be added.
- */
- public void addConnectionEventListener(final ConnectionEventListener l)
- {
- listeners.addElement(l);
+ if (lock.hasQueuedThreads())
+ {
+ Collection<Thread> threads = lock.getQueuedThreads();
+ for (Thread thread : threads)
+ {
+ Throwable t = new Throwable("Thread waiting for lock during cleanup");
+ t.setStackTrace(thread.getStackTrace());
- if (log.isTraceEnabled())
- log.trace("ConnectionEvent listener added: " + l);
- }
+ log.warn(t.getMessage(), t);
+ }
+ isActive = true;
+ }
- /**
- * Remove a connection event listener.
- *
- * @param l The connection event listener to be removed.
- */
- public void removeConnectionEventListener(final ConnectionEventListener l)
- {
- listeners.removeElement(l);
- }
+ if (lock.isLocked())
+ {
+ Throwable t = new Throwable("Lock owned during cleanup");
+ t.setStackTrace(lock.getOwner().getStackTrace());
- /**
- * Get the XAResource for the connection.
- *
- * @return The XAResource for the connection.
- *
- * @throws ResourceException XA transaction not supported
- */
- public XAResource getXAResource() throws ResourceException
- {
- //
- // Spec says a mc must allways return the same XA resource,
- // so we cache it.
- //
- if (!xaTransacted)
- throw new NotSupportedException("Non XA transaction not supported");
+ log.warn(t.getMessage(), t);
- if (xaResource == null)
- {
- if (info.getType() == JmsConnectionFactory.TOPIC)
- xaResource = xaTopicSession.getXAResource();
- else if (info.getType() == JmsConnectionFactory.QUEUE)
- xaResource = xaQueueSession.getXAResource();
- else
- xaResource = xaSession.getXAResource();
- }
+ isActive = true;
+ }
- if (log.isTraceEnabled())
- log.trace("XAResource=" + xaResource);
+ if (isActive)
+ {
+ // There are active lock - make sure that the JCA container kills
+ // this handle by throwing an exception
- xaResource = new JmsXAResource(this, xaResource);
- return xaResource;
- }
+ throw new ResourceException("Still active locks for " + this);
+ }
+ }
- /**
- * Get the location transaction for the connection.
- *
- * @return The local transaction for the connection.
- *
- * @throws ResourceException
- */
- public LocalTransaction getLocalTransaction() throws ResourceException
- {
- LocalTransaction tx = new JmsLocalTransaction(this);
- if (log.isTraceEnabled())
- log.trace("LocalTransaction=" + tx);
- return tx;
- }
+ /**
+ * Move a handler from one mc to this one.
+ *
+ * @param obj
+ * An object of type JmsSession.
+ *
+ * @throws ResourceException
+ * Failed to associate connection.
+ * @throws IllegalStateException
+ * ManagedConnection in an illegal state.
+ */
+ public void associateConnection(final Object obj) throws ResourceException
+ {
+ //
+ // Should we check auth, ie user and pwd? FIXME
+ //
- /**
- * Get the meta data for the connection.
- *
- * @return The meta data for the connection.
- *
- * @throws ResourceException
- * @throws IllegalStateException ManagedConnection already destroyed.
- */
- public ManagedConnectionMetaData getMetaData() throws ResourceException
- {
- if (isDestroyed)
- throw new IllegalStateException("ManagedConnection already destroyd");
+ if (!isDestroyed && obj instanceof JmsSession)
+ {
+ JmsSession h = (JmsSession) obj;
+ h.setManagedConnection(this);
+ handles.add(h);
+ }
+ else
+ throw new IllegalStateException("ManagedConnection in an illegal state");
+ }
- return new JmsMetaData(this);
- }
+ protected void lock()
+ {
+ lock.lock();
+ }
- /**
- * Set the log writer for this connection.
- *
- * @param out The log writer for this connection.
- *
- * @throws ResourceException
- */
- public void setLogWriter(final PrintWriter out) throws ResourceException
- {
- //
- // jason: screw the logWriter stuff for now it sucks ass
- //
- }
+ protected void tryLock() throws JMSException
+ {
+ int tryLock = mcf.getUseTryLock();
+ if (tryLock <= 0)
+ {
+ lock();
+ return;
+ }
+ try
+ {
+ if (lock.tryLock(tryLock, TimeUnit.SECONDS) == false)
+ throw new ResourceAllocationException("Unable to obtain lock in " + tryLock + " seconds: " + this);
+ }
+ catch (InterruptedException e)
+ {
+ throw new ResourceAllocationException("Interrupted attempting lock: " + this);
+ }
+ }
- /**
- * Get the log writer for this connection.
- *
- * @return Always null
- */
- public PrintWriter getLogWriter() throws ResourceException
- {
- //
- // jason: screw the logWriter stuff for now it sucks ass
- //
+ protected void unlock()
+ {
+ if (lock.isLocked())
+ {
+ lock.unlock();
+ }
+ else
+ {
+ log.warn("Owner is null");
+ Throwable t = new Throwable("Thread trying to unlock");
+ t.setStackTrace(Thread.currentThread().getStackTrace());
- return null;
- }
+ log.warn(t.getMessage(), t);
+ }
+ }
- // --- Exception listener implementation
-
- public void onException(JMSException exception)
- {
- if (isDestroyed)
- {
- if (log.isTraceEnabled())
- log.trace("Ignoring error on already destroyed connection " + this, exception);
- return;
- }
+ /**
+ * Add a connection event listener.
+ *
+ * @param l
+ * The connection event listener to be added.
+ */
+ public void addConnectionEventListener(final ConnectionEventListener l)
+ {
+ listeners.addElement(l);
- log.warn("Handling jms exception failure: " + this, exception);
+ if (log.isTraceEnabled())
+ log.trace("ConnectionEvent listener added: " + l);
+ }
- try
- {
- con.setExceptionListener(null);
- }
- catch (JMSException e)
- {
- log.debug("Unable to unset exception listener", e);
- }
-
- ConnectionEvent event = new ConnectionEvent(this, ConnectionEvent.CONNECTION_ERROR_OCCURRED, exception);
- sendEvent(event);
- }
-
- // --- Api to JmsSession
+ /**
+ * Remove a connection event listener.
+ *
+ * @param l
+ * The connection event listener to be removed.
+ */
+ public void removeConnectionEventListener(final ConnectionEventListener l)
+ {
+ listeners.removeElement(l);
+ }
- /**
- * Get the session for this connection.
- *
- * @return Either a topic or queue connection.
- */
- protected Session getSession()
- {
- if (info.getType() == JmsConnectionFactory.TOPIC)
- return topicSession;
- else if (info.getType() == JmsConnectionFactory.QUEUE)
- return queueSession;
- else
- return session;
- }
+ /**
+ * Get the XAResource for the connection.
+ *
+ * @return The XAResource for the connection.
+ *
+ * @throws ResourceException
+ * XA transaction not supported
+ */
+ public XAResource getXAResource() throws ResourceException
+ {
+ //
+ // Spec says a mc must allways return the same XA resource,
+ // so we cache it.
+ //
+ if (!xaTransacted)
+ throw new NotSupportedException("Non XA transaction not supported");
- /**
- * Send an event.
- *
- * @param event The event to send.
- */
- protected void sendEvent(final ConnectionEvent event)
- {
- int type = event.getId();
+ if (xaResource == null)
+ {
+ if (info.getType() == JmsConnectionFactory.TOPIC)
+ xaResource = xaTopicSession.getXAResource();
+ else if (info.getType() == JmsConnectionFactory.QUEUE)
+ xaResource = xaQueueSession.getXAResource();
+ else
+ xaResource = xaSession.getXAResource();
+ }
- if (log.isTraceEnabled())
- log.trace("Sending connection event: " + type);
+ if (log.isTraceEnabled())
+ log.trace("XAResource=" + xaResource);
- // convert to an array to avoid concurrent modification exceptions
- ConnectionEventListener[] list =
- (ConnectionEventListener[])listeners.toArray(new ConnectionEventListener[listeners.size()]);
+ xaResource = new JmsXAResource(this, xaResource);
+ return xaResource;
+ }
- for (int i=0; i<list.length; i++)
- {
- switch (type) {
- case ConnectionEvent.CONNECTION_CLOSED:
- list[i].connectionClosed(event);
- break;
+ /**
+ * Get the location transaction for the connection.
+ *
+ * @return The local transaction for the connection.
+ *
+ * @throws ResourceException
+ */
+ public LocalTransaction getLocalTransaction() throws ResourceException
+ {
+ LocalTransaction tx = new JmsLocalTransaction(this);
+ if (log.isTraceEnabled())
+ log.trace("LocalTransaction=" + tx);
+ return tx;
+ }
- case ConnectionEvent.LOCAL_TRANSACTION_STARTED:
- list[i].localTransactionStarted(event);
- break;
+ /**
+ * Get the meta data for the connection.
+ *
+ * @return The meta data for the connection.
+ *
+ * @throws ResourceException
+ * @throws IllegalStateException
+ * ManagedConnection already destroyed.
+ */
+ public ManagedConnectionMetaData getMetaData() throws ResourceException
+ {
+ if (isDestroyed)
+ throw new IllegalStateException("ManagedConnection already destroyd");
- case ConnectionEvent.LOCAL_TRANSACTION_COMMITTED:
- list[i].localTransactionCommitted(event);
- break;
+ return new JmsMetaData(this);
+ }
- case ConnectionEvent.LOCAL_TRANSACTION_ROLLEDBACK:
- list[i].localTransactionRolledback(event);
- break;
+ /**
+ * Set the log writer for this connection.
+ *
+ * @param out
+ * The log writer for this connection.
+ *
+ * @throws ResourceException
+ */
+ public void setLogWriter(final PrintWriter out) throws ResourceException
+ {
+ //
+ // jason: screw the logWriter stuff for now it sucks ass
+ //
+ }
- case ConnectionEvent.CONNECTION_ERROR_OCCURRED:
- list[i].connectionErrorOccurred(event);
- break;
+ /**
+ * Get the log writer for this connection.
+ *
+ * @return Always null
+ */
+ public PrintWriter getLogWriter() throws ResourceException
+ {
+ //
+ // jason: screw the logWriter stuff for now it sucks ass
+ //
- default:
- throw new IllegalArgumentException("Illegal eventType: " + type);
- }
- }
- }
+ return null;
+ }
- /**
- * Remove a handle from the handle map.
- *
- * @param handle The handle to remove.
- */
- protected void removeHandle(final JmsSession handle)
- {
- handles.remove(handle);
- }
+ // --- Exception listener implementation
- // --- Used by MCF
+ public void onException(JMSException exception)
+ {
+ if (isDestroyed)
+ {
+ if (log.isTraceEnabled())
+ log.trace("Ignoring error on already destroyed connection " + this, exception);
+ return;
+ }
- /**
- * Get the request info for this connection.
- *
- * @return The request info for this connection.
- */
- protected ConnectionRequestInfo getInfo()
- {
- return info;
- }
+ log.warn("Handling jms exception failure: " + this, exception);
- /**
- * Get the connection factory for this connection.
- *
- * @return The connection factory for this connection.
- */
- protected JmsManagedConnectionFactory getManagedConnectionFactory()
- {
- return mcf;
- }
+ // We need to unlock() before sending the connection error to the
+ // event listeners. Otherwise the lock won't be in sync once
+ // cleanup() is called
+ if (lock.isLocked() && Thread.currentThread().equals(lock.getOwner()))
+ unlock();
- void start() throws JMSException
- {
- con.start();
- }
+ try
+ {
+ con.setExceptionListener(null);
+ }
+ catch (JMSException e)
+ {
+ log.debug("Unable to unset exception listener", e);
+ }
- void stop() throws JMSException
- {
- con.stop();
- }
-
- // --- Used by MetaData
+ ConnectionEvent event = new ConnectionEvent(this, ConnectionEvent.CONNECTION_ERROR_OCCURRED, exception);
+ sendEvent(event);
+ }
- /**
- * Get the user name for this connection.
- *
- * @return The user name for this connection.
- */
- protected String getUserName()
- {
- return user;
- }
+ // --- Api to JmsSession
- // --- Private helper methods
+ /**
+ * Get the session for this connection.
+ *
+ * @return Either a topic or queue connection.
+ */
+ protected Session getSession()
+ {
+ if (info.getType() == JmsConnectionFactory.TOPIC)
+ return topicSession;
+ else if (info.getType() == JmsConnectionFactory.QUEUE)
+ return queueSession;
+ else
+ return session;
+ }
- /**
- * Get the JMS provider adapter that will be used to create JMS
- * resources.
- *
- * @return A JMS provider adapter.
- *
- * @throws NamingException Failed to lookup provider adapter.
- */
- private JMSProviderAdapter getProviderAdapter() throws NamingException
- {
- JMSProviderAdapter adapter;
+ /**
+ * Send an event.
+ *
+ * @param event
+ * The event to send.
+ */
+ protected void sendEvent(final ConnectionEvent event)
+ {
+ int type = event.getId();
- if (mcf.getJmsProviderAdapterJNDI() != null)
- {
- // lookup the adapter from JNDI
- Context ctx = new InitialContext();
- try
- {
- adapter = (JMSProviderAdapter)
- ctx.lookup(mcf.getJmsProviderAdapterJNDI());
- }
- finally
- {
- ctx.close();
- }
- }
- else
- adapter = mcf.getJmsProviderAdapter();
+ if (log.isTraceEnabled())
+ log.trace("Sending connection event: " + type);
- return adapter;
- }
+ // convert to an array to avoid concurrent modification exceptions
+ ConnectionEventListener[] list = (ConnectionEventListener[]) listeners.toArray(new ConnectionEventListener[listeners.size()]);
- /**
- * Setup the connection.
- *
- * @throws ResourceException
- */
- private void setup() throws ResourceException
- {
- boolean trace = log.isTraceEnabled();
+ for (int i = 0; i < list.length; i++)
+ {
+ switch (type)
+ {
+ case ConnectionEvent.CONNECTION_CLOSED:
+ list[i].connectionClosed(event);
+ break;
- try
- {
- JMSProviderAdapter adapter = getProviderAdapter();
- Context context = adapter.getInitialContext();
- Object factory;
- boolean transacted = info.isTransacted();
- int ack = Session.AUTO_ACKNOWLEDGE;
+ case ConnectionEvent.LOCAL_TRANSACTION_STARTED:
+ list[i].localTransactionStarted(event);
+ break;
- if (info.getType() == JmsConnectionFactory.TOPIC)
- {
- String jndi = adapter.getTopicFactoryRef();
- if (jndi == null)
- throw new IllegalStateException("No configured 'TopicFactoryRef' on the jms provider " + mcf.getJmsProviderAdapterJNDI());
- factory = context.lookup(jndi);
- con = ConnectionFactoryHelper.createTopicConnection(factory, user, pwd);
- if (info.getClientID() != null)
- con.setClientID(info.getClientID());
- con.setExceptionListener(this);
- if (trace)
- log.trace("created connection: " + con);
+ case ConnectionEvent.LOCAL_TRANSACTION_COMMITTED:
+ list[i].localTransactionCommitted(event);
+ break;
- if (con instanceof XATopicConnection)
- {
- xaTopicSession = ((XATopicConnection)con).createXATopicSession();
- topicSession = xaTopicSession.getTopicSession();
- xaTransacted = true;
- }
- else if (con instanceof TopicConnection)
- {
- topicSession =
- ((TopicConnection)con).createTopicSession(transacted, ack);
- if (trace)
- log.trace("Using a non-XA TopicConnection. " +
- "It will not be able to participate in a Global UOW");
- }
- else
- throw new JBossResourceException("Connection was not recognizable: " + con);
+ case ConnectionEvent.LOCAL_TRANSACTION_ROLLEDBACK:
+ list[i].localTransactionRolledback(event);
+ break;
- if (trace)
- log.trace("xaTopicSession=" + xaTopicSession + ", topicSession=" + topicSession);
- }
- else if (info.getType() == JmsConnectionFactory.QUEUE)
- {
- String jndi = adapter.getQueueFactoryRef();
- if (jndi == null)
- throw new IllegalStateException("No configured 'QueueFactoryRef' on the jms provider " + mcf.getJmsProviderAdapterJNDI());
- factory = context.lookup(jndi);
- con = ConnectionFactoryHelper.createQueueConnection(factory, user, pwd);
- if (info.getClientID() != null)
- con.setClientID(info.getClientID());
- con.setExceptionListener(this);
- if (trace)
- log.debug("created connection: " + con);
+ case ConnectionEvent.CONNECTION_ERROR_OCCURRED:
+ list[i].connectionErrorOccurred(event);
+ break;
- if (con instanceof XAQueueConnection)
- {
- xaQueueSession =
- ((XAQueueConnection)con).createXAQueueSession();
- queueSession = xaQueueSession.getQueueSession();
- xaTransacted = true;
- }
- else if (con instanceof QueueConnection)
- {
- queueSession =
- ((QueueConnection)con).createQueueSession(transacted, ack);
- if (trace)
- log.trace("Using a non-XA QueueConnection. " +
- "It will not be able to participate in a Global UOW");
- }
- else
- throw new JBossResourceException("Connection was not reconizable: " + con);
+ default:
+ throw new IllegalArgumentException("Illegal eventType: " + type);
+ }
+ }
+ }
- if (trace)
- log.trace("xaQueueSession=" + xaQueueSession + ", queueSession=" + queueSession);
- }
- else
- {
- String jndi = adapter.getFactoryRef();
- if (jndi == null)
- throw new IllegalStateException("No configured 'FactoryRef' on the jms provider " + mcf.getJmsProviderAdapterJNDI());
- factory = context.lookup(jndi);
- con = ConnectionFactoryHelper.createConnection(factory, user, pwd);
- if (info.getClientID() != null)
- con.setClientID(info.getClientID());
- con.setExceptionListener(this);
- if (trace)
- log.trace("created connection: " + con);
+ /**
+ * Remove a handle from the handle map.
+ *
+ * @param handle
+ * The handle to remove.
+ */
+ protected void removeHandle(final JmsSession handle)
+ {
+ handles.remove(handle);
+ }
- if (con instanceof XAConnection)
- {
- xaSession =
- ((XAConnection)con).createXASession();
- session = xaSession.getSession();
- xaTransacted = true;
- }
- else
- {
- session = con.createSession(transacted, ack);
- if (trace)
- log.trace("Using a non-XA Connection. " +
- "It will not be able to participate in a Global UOW");
- }
+ // --- Used by MCF
- if (trace)
- log.debug("xaSession=" + xaQueueSession + ", Session=" + session);
- }
+ /**
+ * Get the request info for this connection.
+ *
+ * @return The request info for this connection.
+ */
+ protected ConnectionRequestInfo getInfo()
+ {
+ return info;
+ }
- if (trace)
- log.debug("transacted=" + transacted + ", ack=" + ack);
- }
- catch (NamingException e)
- {
- throw new JBossResourceException("Unable to setup connection", e);
- }
- catch (JMSException e)
- {
- throw new JBossResourceException("Unable to setup connection", e);
- }
- }
+ /**
+ * Get the connection factory for this connection.
+ *
+ * @return The connection factory for this connection.
+ */
+ protected JmsManagedConnectionFactory getManagedConnectionFactory()
+ {
+ return mcf;
+ }
+
+ void start() throws JMSException
+ {
+ con.start();
+ }
+
+ void stop() throws JMSException
+ {
+ con.stop();
+ }
+
+ // --- Used by MetaData
+
+ /**
+ * Get the user name for this connection.
+ *
+ * @return The user name for this connection.
+ */
+ protected String getUserName()
+ {
+ return user;
+ }
+
+ // --- Private helper methods
+
+ /**
+ * Get the JMS provider adapter that will be used to create JMS resources.
+ *
+ * @return A JMS provider adapter.
+ *
+ * @throws NamingException
+ * Failed to lookup provider adapter.
+ */
+ private JMSProviderAdapter getProviderAdapter() throws NamingException
+ {
+ JMSProviderAdapter adapter;
+
+ if (mcf.getJmsProviderAdapterJNDI() != null)
+ {
+ // lookup the adapter from JNDI
+ Context ctx = new InitialContext();
+ try
+ {
+ adapter = (JMSProviderAdapter) ctx.lookup(mcf.getJmsProviderAdapterJNDI());
+ }
+ finally
+ {
+ ctx.close();
+ }
+ }
+ else
+ adapter = mcf.getJmsProviderAdapter();
+
+ return adapter;
+ }
+
+ /**
+ * Setup the connection.
+ *
+ * @throws ResourceException
+ */
+ private void setup() throws ResourceException
+ {
+ boolean trace = log.isTraceEnabled();
+
+ try
+ {
+ JMSProviderAdapter adapter = getProviderAdapter();
+ Context context = adapter.getInitialContext();
+ Object factory;
+ boolean transacted = info.isTransacted();
+ int ack = Session.AUTO_ACKNOWLEDGE;
+
+ if (info.getType() == JmsConnectionFactory.TOPIC)
+ {
+ String jndi = adapter.getTopicFactoryRef();
+ if (jndi == null)
+ throw new IllegalStateException("No configured 'TopicFactoryRef' on the jms provider " + mcf.getJmsProviderAdapterJNDI());
+ factory = context.lookup(jndi);
+ con = ConnectionFactoryHelper.createTopicConnection(factory, user, pwd);
+ if (info.getClientID() != null)
+ con.setClientID(info.getClientID());
+ con.setExceptionListener(this);
+ if (trace)
+ log.trace("created connection: " + con);
+
+ if (con instanceof XATopicConnection)
+ {
+ xaTopicSession = ((XATopicConnection) con).createXATopicSession();
+ topicSession = xaTopicSession.getTopicSession();
+ xaTransacted = true;
+ }
+ else if (con instanceof TopicConnection)
+ {
+ topicSession = ((TopicConnection) con).createTopicSession(transacted, ack);
+ if (trace)
+ log.trace("Using a non-XA TopicConnection. " + "It will not be able to participate in a Global UOW");
+ }
+ else
+ throw new JBossResourceException("Connection was not recognizable: " + con);
+
+ if (trace)
+ log.trace("xaTopicSession=" + xaTopicSession + ", topicSession=" + topicSession);
+ }
+ else if (info.getType() == JmsConnectionFactory.QUEUE)
+ {
+ String jndi = adapter.getQueueFactoryRef();
+ if (jndi == null)
+ throw new IllegalStateException("No configured 'QueueFactoryRef' on the jms provider " + mcf.getJmsProviderAdapterJNDI());
+ factory = context.lookup(jndi);
+ con = ConnectionFactoryHelper.createQueueConnection(factory, user, pwd);
+ if (info.getClientID() != null)
+ con.setClientID(info.getClientID());
+ con.setExceptionListener(this);
+ if (trace)
+ log.debug("created connection: " + con);
+
+ if (con instanceof XAQueueConnection)
+ {
+ xaQueueSession = ((XAQueueConnection) con).createXAQueueSession();
+ queueSession = xaQueueSession.getQueueSession();
+ xaTransacted = true;
+ }
+ else if (con instanceof QueueConnection)
+ {
+ queueSession = ((QueueConnection) con).createQueueSession(transacted, ack);
+ if (trace)
+ log.trace("Using a non-XA QueueConnection. " + "It will not be able to participate in a Global UOW");
+ }
+ else
+ throw new JBossResourceException("Connection was not reconizable: " + con);
+
+ if (trace)
+ log.trace("xaQueueSession=" + xaQueueSession + ", queueSession=" + queueSession);
+ }
+ else
+ {
+ String jndi = adapter.getFactoryRef();
+ if (jndi == null)
+ throw new IllegalStateException("No configured 'FactoryRef' on the jms provider " + mcf.getJmsProviderAdapterJNDI());
+ factory = context.lookup(jndi);
+ con = ConnectionFactoryHelper.createConnection(factory, user, pwd);
+ if (info.getClientID() != null)
+ con.setClientID(info.getClientID());
+ con.setExceptionListener(this);
+ if (trace)
+ log.trace("created connection: " + con);
+
+ if (con instanceof XAConnection)
+ {
+ xaSession = ((XAConnection) con).createXASession();
+ session = xaSession.getSession();
+ xaTransacted = true;
+ }
+ else
+ {
+ session = con.createSession(transacted, ack);
+ if (trace)
+ log.trace("Using a non-XA Connection. " + "It will not be able to participate in a Global UOW");
+ }
+
+ if (trace)
+ log.debug("xaSession=" + xaQueueSession + ", Session=" + session);
+ }
+
+ if (trace)
+ log.debug("transacted=" + transacted + ", ack=" + ack);
+ }
+ catch (NamingException e)
+ {
+ throw new JBossResourceException("Unable to setup connection", e);
+ }
+ catch (JMSException e)
+ {
+ throw new JBossResourceException("Unable to setup connection", e);
+ }
+ }
}
Modified: branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jms/JmsSession.java
===================================================================
--- branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jms/JmsSession.java 2015-02-12 18:08:11 UTC (rev 114833)
+++ branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jms/JmsSession.java 2015-02-13 05:36:48 UTC (rev 114834)
@@ -65,799 +65,818 @@
*/
public class JmsSession implements Session, QueueSession, TopicSession
{
- private static final Logger log = Logger.getLogger(JmsSession.class);
+ private static final Logger log = Logger.getLogger(JmsSession.class);
- /** The managed connection for this session. */
- private JmsManagedConnection mc; // = null;
+ /** The managed connection for this session. */
+ private volatile JmsManagedConnection mc;
- /** The connection request info */
- private JmsConnectionRequestInfo info;
+ private JmsManagedConnection lockedMC;
- /** The session factory for this session */
- private JmsSessionFactoryImpl sf;
-
- /** The message consumers */
- private HashSet consumers = new HashSet();
-
- /** The message producers */
- private HashSet producers = new HashSet();
-
- /** Whether trace is enabled */
- private boolean trace = log.isTraceEnabled();
-
- /**
- * Construct a <tt>JmsSession</tt>.
- *
- * @param mc The managed connection for this session.
- */
- public JmsSession(final JmsManagedConnection mc, JmsConnectionRequestInfo info)
- {
- this.mc = mc;
- this.info = info;
- if (trace)
- log.trace("new JmsSession " + this + " mc=" + mc + " cri=" + info);
- }
+ private int lockCount;
- public void setJmsSessionFactory(JmsSessionFactoryImpl sf)
- {
- this.sf = sf;
- }
-
- protected void lock() throws JMSException
- {
- JmsManagedConnection mc = this.mc;
- if (mc != null)
- mc.tryLock();
- else
- throw new IllegalStateException("Connection is not associated with a managed connection. " + this);
- }
+ /** The connection request info */
+ private JmsConnectionRequestInfo info;
- protected void unlock()
- {
- JmsManagedConnection mc = this.mc;
- if (mc != null)
- mc.unlock();
- // We recreate the lock when returned to the pool
- // so missing the unlock after disassociation is not important
- }
-
- /**
- * Ensure that the session is opened.
- *
- * @return The session
- *
- * @throws IllegalStateException The session is closed
- */
- Session getSession() throws JMSException
- {
- // ensure that the connection is opened
- if (mc == null)
- throw new IllegalStateException("The session is closed");
-
- checkTransactionActive();
-
- Session session = mc.getSession();
- if (trace)
- log.trace("getSession " + session + " for " + this);
- return session;
- }
+ /** The session factory for this session */
+ private JmsSessionFactoryImpl sf;
- void checkTransactionActive() throws IllegalStateException
- {
- if (sf != null)
- sf.checkTransactionActive();
- }
-
- // ---- Session API
+ /** The message consumers */
+ private HashSet consumers = new HashSet();
- public BytesMessage createBytesMessage() throws JMSException
- {
- Session session = getSession();
- if (trace)
- log.trace("createBytesMessage" + session);
- return session.createBytesMessage();
- }
+ /** The message producers */
+ private HashSet producers = new HashSet();
- public MapMessage createMapMessage() throws JMSException
- {
- Session session = getSession();
- if (trace)
- log.trace("createMapMessage" + session);
- return session.createMapMessage();
- }
+ /** Whether trace is enabled */
+ private boolean trace = log.isTraceEnabled();
- public Message createMessage() throws JMSException
- {
- Session session = getSession();
- if (trace)
- log.trace("createMessage" + session);
- return session.createMessage();
- }
+ /**
+ * Construct a <tt>JmsSession</tt>.
+ *
+ * @param mc
+ * The managed connection for this session.
+ */
+ public JmsSession(final JmsManagedConnection mc, JmsConnectionRequestInfo info)
+ {
+ this.mc = mc;
+ this.lockedMC = null;
+ this.lockCount = 0;
+ this.info = info;
+ if (trace)
+ log.trace("new JmsSession " + this + " mc=" + mc + " cri=" + info);
+ }
- public ObjectMessage createObjectMessage() throws JMSException
- {
- Session session = getSession();
- if (trace)
- log.trace("createObjectMessage" + session);
- return session.createObjectMessage();
- }
+ public void setJmsSessionFactory(JmsSessionFactoryImpl sf)
+ {
+ this.sf = sf;
+ }
- public ObjectMessage createObjectMessage(Serializable object) throws JMSException
- {
- Session session = getSession();
- if (trace)
- log.trace("createObjectMessage(Object)" + session);
- return session.createObjectMessage(object);
- }
+ protected void lock() throws JMSException
+ {
+ JmsManagedConnection mc = this.mc;
+ if (mc != null)
+ {
+ mc.tryLock();
- public StreamMessage createStreamMessage() throws JMSException
- {
- Session session = getSession();
- if (trace)
- log.trace("createStreamMessage" + session);
- return session.createStreamMessage();
- }
+ if (lockedMC == null)
+ lockedMC = mc;
- public TextMessage createTextMessage() throws JMSException
- {
- Session session = getSession();
- if (trace)
- log.trace("createTextMessage" + session);
- return session.createTextMessage();
- }
+ lockCount++;
+ }
+ else
+ throw new IllegalStateException("Connection is not associated with a managed connection. " + this);
+ }
- public TextMessage createTextMessage(String string) throws JMSException
- {
- Session session = getSession();
- if (trace)
- log.trace("createTextMessage(String)" + session);
- return session.createTextMessage(string);
- }
+ protected void unlock()
+ {
+ JmsManagedConnection mc = this.lockedMC;
+ if (--lockCount == 0)
+ lockedMC = null;
+ if (mc != null)
+ mc.unlock();
+ }
- public boolean getTransacted() throws JMSException
- {
- getSession(); // check closed
- return info.isTransacted();
- }
+ /**
+ * Ensure that the session is opened.
+ *
+ * @return The session
+ *
+ * @throws IllegalStateException
+ * The session is closed
+ */
+ Session getSession() throws JMSException
+ {
+ // ensure that the connection is opened
+ if (mc == null)
+ throw new IllegalStateException("The session is closed");
- /**
- * Always throws an Exception.
- *
- * @throws IllegalStateException Method not allowed.
- */
- public MessageListener getMessageListener() throws JMSException
- {
- throw new IllegalStateException("Method not allowed");
- }
+ checkTransactionActive();
- /**
- * Always throws an Exception.
- *
- * @throws IllegalStateException Method not allowed.
- */
- public void setMessageListener(MessageListener listener) throws JMSException
- {
- throw new IllegalStateException("Method not allowed");
- }
+ Session session = mc.getSession();
+ if (trace)
+ log.trace("getSession " + session + " for " + this);
+ return session;
+ }
- /**
- * Always throws an Error.
- *
- * @throws Error Method not allowed.
- */
- public void run()
- {
- // should this really throw an Error?
- throw new Error("Method not allowed");
- }
+ void checkTransactionActive() throws IllegalStateException
+ {
+ if (sf != null)
+ sf.checkTransactionActive();
+ }
- /**
- * Closes the session. Sends a ConnectionEvent.CONNECTION_CLOSED to the
- * managed connection.
- *
- * @throws JMSException Failed to close session.
- */
- public void close() throws JMSException
- {
- sf.closeSession(this);
- closeSession();
- }
+ // ---- Session API
- // FIXME - is this really OK, probably not
- public void commit() throws JMSException
- {
- lock();
- try
- {
- Session session = getSession();
- if (info.isTransacted() == false)
- throw new IllegalStateException("Session is not transacted");
- if (trace)
- log.trace("Commit session " + this);
- session.commit();
- }
- finally
- {
- unlock();
- }
- }
+ public BytesMessage createBytesMessage() throws JMSException
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createBytesMessage" + session);
+ return session.createBytesMessage();
+ }
- public void rollback() throws JMSException
- {
- lock();
- try
- {
- Session session = getSession();
- if (info.isTransacted() == false)
- throw new IllegalStateException("Session is not transacted");
- if (trace)
- log.trace("Rollback session " + this);
- session.rollback();
- }
- finally
- {
- unlock();
- }
- }
+ public MapMessage createMapMessage() throws JMSException
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createMapMessage" + session);
+ return session.createMapMessage();
+ }
- public void recover() throws JMSException
- {
- lock();
- try
- {
- Session session = getSession();
- if (info.isTransacted())
- throw new IllegalStateException("Session is transacted");
- if (trace)
- log.trace("Recover session " + this);
- session.recover();
- }
- finally
- {
- unlock();
- }
- }
+ public Message createMessage() throws JMSException
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createMessage" + session);
+ return session.createMessage();
+ }
- // --- TopicSession API
+ public ObjectMessage createObjectMessage() throws JMSException
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createObjectMessage" + session);
+ return session.createObjectMessage();
+ }
- public Topic createTopic(String topicName) throws JMSException
- {
- if(info.getType() == JmsConnectionFactory.QUEUE)
- {
- throw new IllegalStateException("Cannot create topic for javax.jms.QueueSession");
- }
+ public ObjectMessage createObjectMessage(Serializable object) throws JMSException
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createObjectMessage(Object)" + session);
+ return session.createObjectMessage(object);
+ }
- Session session = getSession();
- if (trace)
- log.trace("createTopic " + session + " topicName=" + topicName);
- Topic result = session.createTopic(topicName);
- if (trace)
- log.trace("createdTopic " + session + " topic=" + result);
- return result;
- }
+ public StreamMessage createStreamMessage() throws JMSException
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createStreamMessage" + session);
+ return session.createStreamMessage();
+ }
- public TopicSubscriber createSubscriber(Topic topic) throws JMSException
- {
- lock();
- try
- {
- TopicSession session = getTopicSession();
- if (trace)
- log.trace("createSubscriber " + session + " topic=" + topic);
- TopicSubscriber result = session.createSubscriber(topic);
- result = new JmsTopicSubscriber(result, this);
- if (trace)
- log.trace("createdSubscriber " + session + " JmsTopicSubscriber=" + result);
- addConsumer(result);
- return result;
- }
- finally
- {
- unlock();
- }
- }
+ public TextMessage createTextMessage() throws JMSException
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createTextMessage" + session);
+ return session.createTextMessage();
+ }
- public TopicSubscriber createSubscriber(Topic topic, String messageSelector, boolean noLocal) throws JMSException
- {
- lock();
- try
- {
- TopicSession session = getTopicSession();
- if (trace)
- log.trace("createSubscriber " + session + " topic=" + topic + " selector=" + messageSelector + " noLocal=" + noLocal);
- TopicSubscriber result = session.createSubscriber(topic, messageSelector, noLocal);
- result = new JmsTopicSubscriber(result, this);
- if (trace)
- log.trace("createdSubscriber " + session + " JmsTopicSubscriber=" + result);
- addConsumer(result);
- return result;
- }
- finally
- {
- unlock();
- }
- }
+ public TextMessage createTextMessage(String string) throws JMSException
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createTextMessage(String)" + session);
+ return session.createTextMessage(string);
+ }
- public TopicSubscriber createDurableSubscriber(Topic topic, String name) throws JMSException
- {
- if(info.getType() == JmsConnectionFactory.QUEUE)
- {
- throw new IllegalStateException("Cannot create durable subscriber from javax.jms.QueueSession");
- }
-
- lock();
- try
- {
- Session session = getSession();
- if (trace)
- log.trace("createDurableSubscriber " + session + " topic=" + topic + " name=" + name);
- TopicSubscriber result = session.createDurableSubscriber(topic, name);
- result = new JmsTopicSubscriber(result, this);
- if (trace)
- log.trace("createdDurableSubscriber " + session + " JmsTopicSubscriber=" + result);
- addConsumer(result);
- return result;
- }
- finally
- {
- unlock();
- }
- }
+ public boolean getTransacted() throws JMSException
+ {
+ getSession(); // check closed
+ return info.isTransacted();
+ }
- public TopicSubscriber createDurableSubscriber(Topic topic, String name, String messageSelector, boolean noLocal)
- throws JMSException
- {
- lock();
- try
- {
- Session session = getSession();
- if (trace)
- log.trace("createDurableSubscriber " + session + " topic=" + topic + " name=" + name + " selector=" + messageSelector + " noLocal=" + noLocal);
- TopicSubscriber result = session.createDurableSubscriber(topic, name, messageSelector, noLocal);
- result = new JmsTopicSubscriber(result, this);
- if (trace)
- log.trace("createdDurableSubscriber " + session + " JmsTopicSubscriber=" + result);
- addConsumer(result);
- return result;
- }
- finally
- {
- unlock();
- }
- }
+ /**
+ * Always throws an Exception.
+ *
+ * @throws IllegalStateException
+ * Method not allowed.
+ */
+ public MessageListener getMessageListener() throws JMSException
+ {
+ throw new IllegalStateException("Method not allowed");
+ }
- public TopicPublisher createPublisher(Topic topic) throws JMSException
- {
- lock();
- try
- {
- TopicSession session = getTopicSession();
- if (trace)
- log.trace("createPublisher " + session + " topic=" + topic);
- TopicPublisher result = session.createPublisher(topic);
- result = new JmsTopicPublisher(result, this);
- if (trace)
- log.trace("createdPublisher " + session + " publisher=" + result);
- addProducer(result);
- return result;
- }
- finally
- {
- unlock();
- }
- }
+ /**
+ * Always throws an Exception.
+ *
+ * @throws IllegalStateException
+ * Method not allowed.
+ */
+ public void setMessageListener(MessageListener listener) throws JMSException
+ {
+ throw new IllegalStateException("Method not allowed");
+ }
- public TemporaryTopic createTemporaryTopic() throws JMSException
- {
- if(info.getType() == JmsConnectionFactory.QUEUE)
- {
- throw new IllegalStateException("Cannot create temporary topic for javax.jms.QueueSession");
- }
-
- lock();
- try
- {
- Session session = getSession();
- if (trace)
- log.trace("createTemporaryTopic " + session);
- TemporaryTopic temp = session.createTemporaryTopic();
- if (trace)
- log.trace("createdTemporaryTopic " + session + " temp=" + temp);
- sf.addTemporaryTopic(temp);
- return temp;
- }
- finally
- {
- unlock();
- }
- }
+ /**
+ * Always throws an Error.
+ *
+ * @throws Error
+ * Method not allowed.
+ */
+ public void run()
+ {
+ // should this really throw an Error?
+ throw new Error("Method not allowed");
+ }
- public void unsubscribe(String name) throws JMSException
- {
- if(info.getType() == JmsConnectionFactory.QUEUE)
- {
- throw new IllegalStateException("Cannot unsubscribe for javax.jms.QueueSession");
- }
+ /**
+ * Closes the session. Sends a ConnectionEvent.CONNECTION_CLOSED to the
+ * managed connection.
+ *
+ * @throws JMSException
+ * Failed to close session.
+ */
+ public void close() throws JMSException
+ {
+ sf.closeSession(this);
+ closeSession();
+ }
- lock();
- try
- {
- Session session = getSession();
- if (trace)
- log.trace("unsubscribe " + session + " name=" + name);
- session.unsubscribe(name);
- }
- finally
- {
- unlock();
- }
- }
+ // FIXME - is this really OK, probably not
+ public void commit() throws JMSException
+ {
+ lock();
+ try
+ {
+ Session session = getSession();
+ if (info.isTransacted() == false)
+ throw new IllegalStateException("Session is not transacted");
+ if (trace)
+ log.trace("Commit session " + this);
+ session.commit();
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- //--- QueueSession API
+ public void rollback() throws JMSException
+ {
+ lock();
+ try
+ {
+ Session session = getSession();
+ if (info.isTransacted() == false)
+ throw new IllegalStateException("Session is not transacted");
+ if (trace)
+ log.trace("Rollback session " + this);
+ session.rollback();
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public QueueBrowser createBrowser(Queue queue) throws JMSException
- {
-
- if(info.getType() == JmsConnectionFactory.TOPIC)
- {
- throw new IllegalStateException("Cannot create browser for javax.jms.TopicSession");
-
- }
+ public void recover() throws JMSException
+ {
+ lock();
+ try
+ {
+ Session session = getSession();
+ if (info.isTransacted())
+ throw new IllegalStateException("Session is transacted");
+ if (trace)
+ log.trace("Recover session " + this);
+ session.recover();
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- Session session = getSession();
- if (trace)
- log.trace("createBrowser " + session + " queue=" + queue);
- QueueBrowser result = session.createBrowser(queue);
- if (trace)
- log.trace("createdBrowser " + session + " browser=" + result);
- return result;
- }
+ // --- TopicSession API
- public QueueBrowser createBrowser(Queue queue, String messageSelector) throws JMSException
- {
- Session session = getSession();
- if (trace)
- log.trace("createBrowser " + session + " queue=" + queue + " selector=" + messageSelector);
- QueueBrowser result = session.createBrowser(queue, messageSelector);
- if (trace)
- log.trace("createdBrowser " + session + " browser=" + result);
- return result;
- }
+ public Topic createTopic(String topicName) throws JMSException
+ {
+ if (info.getType() == JmsConnectionFactory.QUEUE)
+ {
+ throw new IllegalStateException("Cannot create topic for javax.jms.QueueSession");
+ }
- public Queue createQueue(String queueName) throws JMSException
- {
- if(info.getType() == JmsConnectionFactory.TOPIC)
- {
- throw new IllegalStateException("Cannot create browser or javax.jms.TopicSession");
-
- }
+ Session session = getSession();
+ if (trace)
+ log.trace("createTopic " + session + " topicName=" + topicName);
+ Topic result = session.createTopic(topicName);
+ if (trace)
+ log.trace("createdTopic " + session + " topic=" + result);
+ return result;
+ }
- Session session = getSession();
- if (trace)
- log.trace("createQueue " + session + " queueName=" + queueName);
- Queue result = session.createQueue(queueName);
- if (trace)
- log.trace("createdQueue " + session + " queue=" + result);
- return result;
- }
+ public TopicSubscriber createSubscriber(Topic topic) throws JMSException
+ {
+ lock();
+ try
+ {
+ TopicSession session = getTopicSession();
+ if (trace)
+ log.trace("createSubscriber " + session + " topic=" + topic);
+ TopicSubscriber result = session.createSubscriber(topic);
+ result = new JmsTopicSubscriber(result, this);
+ if (trace)
+ log.trace("createdSubscriber " + session + " JmsTopicSubscriber=" + result);
+ addConsumer(result);
+ return result;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public QueueReceiver createReceiver(Queue queue) throws JMSException
- {
- lock();
- try
- {
- QueueSession session = getQueueSession();
- if (trace)
- log.trace("createReceiver " + session + " queue=" + queue);
- QueueReceiver result = session.createReceiver(queue);
- result = new JmsQueueReceiver(result, this);
- if (trace)
- log.trace("createdReceiver " + session + " receiver=" + result);
- addConsumer(result);
- return result;
- }
- finally
- {
- unlock();
- }
- }
+ public TopicSubscriber createSubscriber(Topic topic, String messageSelector, boolean noLocal) throws JMSException
+ {
+ lock();
+ try
+ {
+ TopicSession session = getTopicSession();
+ if (trace)
+ log.trace("createSubscriber " + session + " topic=" + topic + " selector=" + messageSelector + " noLocal=" + noLocal);
+ TopicSubscriber result = session.createSubscriber(topic, messageSelector, noLocal);
+ result = new JmsTopicSubscriber(result, this);
+ if (trace)
+ log.trace("createdSubscriber " + session + " JmsTopicSubscriber=" + result);
+ addConsumer(result);
+ return result;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public QueueReceiver createReceiver(Queue queue, String messageSelector) throws JMSException
- {
- lock();
- try
- {
- QueueSession session = getQueueSession();
- if (trace)
- log.trace("createReceiver " + session + " queue=" + queue + " selector=" + messageSelector);
- QueueReceiver result = session.createReceiver(queue, messageSelector);
- result = new JmsQueueReceiver(result, this);
- if (trace)
- log.trace("createdReceiver " + session + " receiver=" + result);
- addConsumer(result);
- return result;
- }
- finally
- {
- unlock();
- }
- }
+ public TopicSubscriber createDurableSubscriber(Topic topic, String name) throws JMSException
+ {
+ if (info.getType() == JmsConnectionFactory.QUEUE)
+ {
+ throw new IllegalStateException("Cannot create durable subscriber from javax.jms.QueueSession");
+ }
- public QueueSender createSender(Queue queue) throws JMSException
- {
- lock();
- try
- {
- QueueSession session = getQueueSession();
- if (trace)
- log.trace("createSender " + session + " queue=" + queue);
- QueueSender result = session.createSender(queue);
- result = new JmsQueueSender(result, this);
- if (trace)
- log.trace("createdSender " + session + " sender=" + result);
- addProducer(result);
- return result;
- }
- finally
- {
- unlock();
- }
- }
+ lock();
+ try
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createDurableSubscriber " + session + " topic=" + topic + " name=" + name);
+ TopicSubscriber result = session.createDurableSubscriber(topic, name);
+ result = new JmsTopicSubscriber(result, this);
+ if (trace)
+ log.trace("createdDurableSubscriber " + session + " JmsTopicSubscriber=" + result);
+ addConsumer(result);
+ return result;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public TemporaryQueue createTemporaryQueue() throws JMSException
- {
- if(info.getType() == JmsConnectionFactory.TOPIC)
- {
- throw new IllegalStateException("Cannot create temporary queue for javax.jms.TopicSession");
-
- }
+ public TopicSubscriber createDurableSubscriber(Topic topic, String name, String messageSelector, boolean noLocal) throws JMSException
+ {
+ lock();
+ try
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createDurableSubscriber " + session + " topic=" + topic + " name=" + name + " selector=" + messageSelector + " noLocal=" + noLocal);
+ TopicSubscriber result = session.createDurableSubscriber(topic, name, messageSelector, noLocal);
+ result = new JmsTopicSubscriber(result, this);
+ if (trace)
+ log.trace("createdDurableSubscriber " + session + " JmsTopicSubscriber=" + result);
+ addConsumer(result);
+ return result;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- lock();
- try
- {
- Session session = getSession();
- if (trace)
- log.trace("createTemporaryQueue " + session);
- TemporaryQueue temp = session.createTemporaryQueue();
- if (trace)
- log.trace("createdTemporaryQueue " + session + " temp=" + temp);
- sf.addTemporaryQueue(temp);
- return temp;
- }
- finally
- {
- unlock();
- }
- }
+ public TopicPublisher createPublisher(Topic topic) throws JMSException
+ {
+ lock();
+ try
+ {
+ TopicSession session = getTopicSession();
+ if (trace)
+ log.trace("createPublisher " + session + " topic=" + topic);
+ TopicPublisher result = session.createPublisher(topic);
+ result = new JmsTopicPublisher(result, this);
+ if (trace)
+ log.trace("createdPublisher " + session + " publisher=" + result);
+ addProducer(result);
+ return result;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- // -- JMS 1.1
+ public TemporaryTopic createTemporaryTopic() throws JMSException
+ {
+ if (info.getType() == JmsConnectionFactory.QUEUE)
+ {
+ throw new IllegalStateException("Cannot create temporary topic for javax.jms.QueueSession");
+ }
- public MessageConsumer createConsumer(Destination destination) throws JMSException
- {
- lock();
- try
- {
- Session session = getSession();
- if (trace)
- log.trace("createConsumer " + session + " dest=" + destination);
- MessageConsumer result = session.createConsumer(destination);
- result = new JmsMessageConsumer(result, this);
- if (trace)
- log.trace("createdConsumer " + session + " consumer=" + result);
- addConsumer(result);
- return result;
- }
- finally
- {
- unlock();
- }
- }
+ lock();
+ try
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createTemporaryTopic " + session);
+ TemporaryTopic temp = session.createTemporaryTopic();
+ if (trace)
+ log.trace("createdTemporaryTopic " + session + " temp=" + temp);
+ sf.addTemporaryTopic(temp);
+ return temp;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public MessageConsumer createConsumer(Destination destination, String messageSelector) throws JMSException
- {
- lock();
- try
- {
- Session session = getSession();
- if (trace)
- log.trace("createConsumer " + session + " dest=" + destination + " messageSelector=" + messageSelector);
- MessageConsumer result = session.createConsumer(destination, messageSelector);
- result = new JmsMessageConsumer(result, this);
- if (trace)
- log.trace("createdConsumer " + session + " consumer=" + result);
- addConsumer(result);
- return result;
- }
- finally
- {
- unlock();
- }
- }
+ public void unsubscribe(String name) throws JMSException
+ {
+ if (info.getType() == JmsConnectionFactory.QUEUE)
+ {
+ throw new IllegalStateException("Cannot unsubscribe for javax.jms.QueueSession");
+ }
- public MessageConsumer createConsumer(Destination destination, String messageSelector, boolean noLocal)
- throws JMSException
- {
- lock();
- try
- {
- Session session = getSession();
- if (trace)
- log.trace("createConsumer " + session + " dest=" + destination + " messageSelector=" + messageSelector + " noLocal=" + noLocal);
- MessageConsumer result = session.createConsumer(destination, messageSelector, noLocal);
- result = new JmsMessageConsumer(result, this);
- if (trace)
- log.trace("createdConsumer " + session + " consumer=" + result);
- addConsumer(result);
- return result;
- }
- finally
- {
- unlock();
- }
- }
+ lock();
+ try
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("unsubscribe " + session + " name=" + name);
+ session.unsubscribe(name);
+ }
+ finally
+ {
+ unlock();
+ }
+ }
- public MessageProducer createProducer(Destination destination) throws JMSException
- {
- lock();
- try
- {
- Session session = getSession();
- if (trace)
- log.trace("createProducer " + session + " dest=" + destination);
- MessageProducer result = getSession().createProducer(destination);
- result = new JmsMessageProducer(result, this);
- if (trace)
- log.trace("createdProducer " + session + " producer=" + result);
- addProducer(result);
- return result;
- }
- finally
- {
- unlock();
- }
- }
+ // --- QueueSession API
- public int getAcknowledgeMode() throws JMSException
- {
- getSession(); // check closed
- return info.getAcknowledgeMode();
- }
+ public QueueBrowser createBrowser(Queue queue) throws JMSException
+ {
- // --- JmsManagedConnection api
+ if (info.getType() == JmsConnectionFactory.TOPIC)
+ {
+ throw new IllegalStateException("Cannot create browser for javax.jms.TopicSession");
- void setManagedConnection(final JmsManagedConnection mc)
- {
- if (this.mc != null)
- this.mc.removeHandle(this);
- this.mc = mc;
- }
+ }
- void destroy()
- {
- mc = null;
- }
+ Session session = getSession();
+ if (trace)
+ log.trace("createBrowser " + session + " queue=" + queue);
+ QueueBrowser result = session.createBrowser(queue);
+ if (trace)
+ log.trace("createdBrowser " + session + " browser=" + result);
+ return result;
+ }
- void start() throws JMSException
- {
- if (mc != null)
- mc.start();
- }
+ public QueueBrowser createBrowser(Queue queue, String messageSelector) throws JMSException
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createBrowser " + session + " queue=" + queue + " selector=" + messageSelector);
+ QueueBrowser result = session.createBrowser(queue, messageSelector);
+ if (trace)
+ log.trace("createdBrowser " + session + " browser=" + result);
+ return result;
+ }
- void stop() throws JMSException
- {
- if (mc != null)
- mc.stop();
- }
+ public Queue createQueue(String queueName) throws JMSException
+ {
+ if (info.getType() == JmsConnectionFactory.TOPIC)
+ {
+ throw new IllegalStateException("Cannot create browser or javax.jms.TopicSession");
- void checkStrict() throws JMSException
- {
- if (mc != null && mc.getManagedConnectionFactory().isStrict())
- throw new IllegalStateException(JmsSessionFactory.ISE);
- }
-
- void closeSession() throws JMSException
- {
- if (mc != null)
- {
- log.trace("Closing session");
+ }
- try
- {
- mc.stop();
- }
- catch (Throwable t)
- {
- log.trace("Error stopping managed connection", t);
- }
-
- synchronized (consumers)
- {
- for (Iterator i = consumers.iterator(); i.hasNext();)
- {
- JmsMessageConsumer consumer = (JmsMessageConsumer) i.next();
- try
- {
- consumer.closeConsumer();
- }
- catch (Throwable t)
- {
- log.trace("Error closing consumer", t);
- }
- i.remove();
- }
- }
+ Session session = getSession();
+ if (trace)
+ log.trace("createQueue " + session + " queueName=" + queueName);
+ Queue result = session.createQueue(queueName);
+ if (trace)
+ log.trace("createdQueue " + session + " queue=" + result);
+ return result;
+ }
- synchronized (producers)
- {
- for (Iterator i = producers.iterator(); i.hasNext();)
- {
- JmsMessageProducer producer = (JmsMessageProducer) i.next();
- try
- {
- producer.closeProducer();
- }
- catch (Throwable t)
- {
- log.trace("Error closing producer", t);
- }
- i.remove();
- }
- }
-
- mc.removeHandle(this);
- ConnectionEvent ev = new ConnectionEvent(mc, ConnectionEvent.CONNECTION_CLOSED);
- ev.setConnectionHandle(this);
- mc.sendEvent(ev);
- mc = null;
- }
- }
-
- void addConsumer(MessageConsumer consumer)
- {
- synchronized (consumers)
- {
- consumers.add(consumer);
- }
- }
-
- void removeConsumer(MessageConsumer consumer)
- {
- synchronized (consumers)
- {
- consumers.remove(consumer);
- }
- }
-
- void addProducer(MessageProducer producer)
- {
- synchronized (producers)
- {
- producers.add(producer);
- }
- }
-
- void removeProducer(MessageProducer producer)
- {
- synchronized (producers)
- {
- producers.remove(producer);
- }
- }
-
- QueueSession getQueueSession() throws JMSException
- {
- Session s = getSession();
- if( !(s instanceof QueueSession) )
- throw new InvalidDestinationException("Attempting to use QueueSession methods on: "+this);
- return (QueueSession) s;
- }
-
- TopicSession getTopicSession() throws JMSException
- {
- Session s = getSession();
- if( !(s instanceof TopicSession) )
- throw new InvalidDestinationException("Attempting to use TopicSession methods on: "+this);
- return (TopicSession) s;
- }
+ public QueueReceiver createReceiver(Queue queue) throws JMSException
+ {
+ lock();
+ try
+ {
+ QueueSession session = getQueueSession();
+ if (trace)
+ log.trace("createReceiver " + session + " queue=" + queue);
+ QueueReceiver result = session.createReceiver(queue);
+ result = new JmsQueueReceiver(result, this);
+ if (trace)
+ log.trace("createdReceiver " + session + " receiver=" + result);
+ addConsumer(result);
+ return result;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
+
+ public QueueReceiver createReceiver(Queue queue, String messageSelector) throws JMSException
+ {
+ lock();
+ try
+ {
+ QueueSession session = getQueueSession();
+ if (trace)
+ log.trace("createReceiver " + session + " queue=" + queue + " selector=" + messageSelector);
+ QueueReceiver result = session.createReceiver(queue, messageSelector);
+ result = new JmsQueueReceiver(result, this);
+ if (trace)
+ log.trace("createdReceiver " + session + " receiver=" + result);
+ addConsumer(result);
+ return result;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
+
+ public QueueSender createSender(Queue queue) throws JMSException
+ {
+ lock();
+ try
+ {
+ QueueSession session = getQueueSession();
+ if (trace)
+ log.trace("createSender " + session + " queue=" + queue);
+ QueueSender result = session.createSender(queue);
+ result = new JmsQueueSender(result, this);
+ if (trace)
+ log.trace("createdSender " + session + " sender=" + result);
+ addProducer(result);
+ return result;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
+
+ public TemporaryQueue createTemporaryQueue() throws JMSException
+ {
+ if (info.getType() == JmsConnectionFactory.TOPIC)
+ {
+ throw new IllegalStateException("Cannot create temporary queue for javax.jms.TopicSession");
+
+ }
+
+ lock();
+ try
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createTemporaryQueue " + session);
+ TemporaryQueue temp = session.createTemporaryQueue();
+ if (trace)
+ log.trace("createdTemporaryQueue " + session + " temp=" + temp);
+ sf.addTemporaryQueue(temp);
+ return temp;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
+
+ // -- JMS 1.1
+
+ public MessageConsumer createConsumer(Destination destination) throws JMSException
+ {
+ lock();
+ try
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createConsumer " + session + " dest=" + destination);
+ MessageConsumer result = session.createConsumer(destination);
+ result = new JmsMessageConsumer(result, this);
+ if (trace)
+ log.trace("createdConsumer " + session + " consumer=" + result);
+ addConsumer(result);
+ return result;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
+
+ public MessageConsumer createConsumer(Destination destination, String messageSelector) throws JMSException
+ {
+ lock();
+ try
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createConsumer " + session + " dest=" + destination + " messageSelector=" + messageSelector);
+ MessageConsumer result = session.createConsumer(destination, messageSelector);
+ result = new JmsMessageConsumer(result, this);
+ if (trace)
+ log.trace("createdConsumer " + session + " consumer=" + result);
+ addConsumer(result);
+ return result;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
+
+ public MessageConsumer createConsumer(Destination destination, String messageSelector, boolean noLocal) throws JMSException
+ {
+ lock();
+ try
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createConsumer " + session + " dest=" + destination + " messageSelector=" + messageSelector + " noLocal=" + noLocal);
+ MessageConsumer result = session.createConsumer(destination, messageSelector, noLocal);
+ result = new JmsMessageConsumer(result, this);
+ if (trace)
+ log.trace("createdConsumer " + session + " consumer=" + result);
+ addConsumer(result);
+ return result;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
+
+ public MessageProducer createProducer(Destination destination) throws JMSException
+ {
+ lock();
+ try
+ {
+ Session session = getSession();
+ if (trace)
+ log.trace("createProducer " + session + " dest=" + destination);
+ MessageProducer result = getSession().createProducer(destination);
+ result = new JmsMessageProducer(result, this);
+ if (trace)
+ log.trace("createdProducer " + session + " producer=" + result);
+ addProducer(result);
+ return result;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
+
+ public int getAcknowledgeMode() throws JMSException
+ {
+ getSession(); // check closed
+ return info.getAcknowledgeMode();
+ }
+
+ // --- JmsManagedConnection api
+
+ void setManagedConnection(final JmsManagedConnection mc)
+ {
+ if (this.mc != null)
+ this.mc.removeHandle(this);
+ this.mc = mc;
+ }
+
+ void destroy()
+ {
+ mc = null;
+ lockedMC = null;
+ lockCount = 0;
+ }
+
+ void start() throws JMSException
+ {
+ if (mc != null)
+ mc.start();
+ }
+
+ void stop() throws JMSException
+ {
+ if (mc != null)
+ mc.stop();
+ }
+
+ void checkStrict() throws JMSException
+ {
+ if (mc != null && mc.getManagedConnectionFactory().isStrict())
+ throw new IllegalStateException(JmsSessionFactory.ISE);
+ }
+
+ void closeSession() throws JMSException
+ {
+ if (mc != null)
+ {
+ log.trace("Closing session");
+
+ try
+ {
+ mc.stop();
+ }
+ catch (Throwable t)
+ {
+ log.trace("Error stopping managed connection", t);
+ }
+
+ synchronized (consumers)
+ {
+ for (Iterator i = consumers.iterator(); i.hasNext();)
+ {
+ JmsMessageConsumer consumer = (JmsMessageConsumer) i.next();
+ try
+ {
+ consumer.closeConsumer();
+ }
+ catch (Throwable t)
+ {
+ log.trace("Error closing consumer", t);
+ }
+ i.remove();
+ }
+ }
+
+ synchronized (producers)
+ {
+ for (Iterator i = producers.iterator(); i.hasNext();)
+ {
+ JmsMessageProducer producer = (JmsMessageProducer) i.next();
+ try
+ {
+ producer.closeProducer();
+ }
+ catch (Throwable t)
+ {
+ log.trace("Error closing producer", t);
+ }
+ i.remove();
+ }
+ }
+
+ mc.removeHandle(this);
+ ConnectionEvent ev = new ConnectionEvent(mc, ConnectionEvent.CONNECTION_CLOSED);
+ ev.setConnectionHandle(this);
+ mc.sendEvent(ev);
+ mc = null;
+ }
+ }
+
+ void addConsumer(MessageConsumer consumer)
+ {
+ synchronized (consumers)
+ {
+ consumers.add(consumer);
+ }
+ }
+
+ void removeConsumer(MessageConsumer consumer)
+ {
+ synchronized (consumers)
+ {
+ consumers.remove(consumer);
+ }
+ }
+
+ void addProducer(MessageProducer producer)
+ {
+ synchronized (producers)
+ {
+ producers.add(producer);
+ }
+ }
+
+ void removeProducer(MessageProducer producer)
+ {
+ synchronized (producers)
+ {
+ producers.remove(producer);
+ }
+ }
+
+ QueueSession getQueueSession() throws JMSException
+ {
+ Session s = getSession();
+ if (!(s instanceof QueueSession))
+ throw new InvalidDestinationException("Attempting to use QueueSession methods on: " + this);
+ return (QueueSession) s;
+ }
+
+ TopicSession getTopicSession() throws JMSException
+ {
+ Session s = getSession();
+ if (!(s instanceof TopicSession))
+ throw new InvalidDestinationException("Attempting to use TopicSession methods on: " + this);
+ return (TopicSession) s;
+ }
}
Added: branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jms/ReentrantLock.java
===================================================================
--- branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jms/ReentrantLock.java (rev 0)
+++ branches/JBPAPP_5_0_1_GA_JBPAPP-11214/connector/src/main/org/jboss/resource/adapter/jms/ReentrantLock.java 2015-02-13 05:36:48 UTC (rev 114834)
@@ -0,0 +1,59 @@
+/*
+ * JBoss, Home of Professional Open Source.
+ * Copyright 2006, Red Hat Middleware LLC, and individual contributors
+ * as indicated by the @author tags. See the copyright.txt file in the
+ * distribution for a full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.jboss.resource.adapter.jms;
+
+import java.util.Collection;
+
+/**
+ * ReentrantLock override
+ * @author <a href="mailto:jesper.pedersen at jboss.org">Jesper Pedersen</a>
+ */
+public class ReentrantLock extends java.util.concurrent.locks.ReentrantLock
+{
+ /** Serial version uid */
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * Constructor
+ * @param fair Fair locking
+ */
+ public ReentrantLock(boolean fair)
+ {
+ super(fair);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public Thread getOwner()
+ {
+ return super.getOwner();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public Collection<Thread> getQueuedThreads()
+ {
+ return super.getQueuedThreads();
+ }
+}
\ No newline at end of file
More information about the jboss-cvs-commits
mailing list