/*
* JBoss, Home of Professional Open Source.
* Copyright 2008-2009, 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.jca.core.connectionmanager;
import org.jboss.jca.common.JBossResourceException;
import org.jboss.jca.common.api.metadata.common.FlushStrategy;
import org.jboss.jca.core.api.connectionmanager.ccm.CachedConnectionManager;
import org.jboss.jca.core.connectionmanager.listener.ConnectionListener;
import org.jboss.jca.core.connectionmanager.listener.ConnectionState;
import org.jboss.jca.core.connectionmanager.pool.api.Pool;
import org.jboss.jca.core.spi.transaction.TransactionIntegration;
import org.jboss.jca.core.spi.transaction.usertx.UserTransactionRegistry;
import java.io.IOException;
import java.io.NotSerializableException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.ObjectStreamException;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.resource.ResourceException;
import javax.resource.spi.ConnectionRequestInfo;
import javax.resource.spi.ManagedConnectionFactory;
import javax.resource.spi.security.PasswordCredential;
import javax.security.auth.Subject;
import javax.transaction.SystemException;
import javax.transaction.Transaction;
import org.jboss.logging.Logger;
import org.jboss.security.SubjectFactory;
/**
* AbstractConnectionManager.
*
* @author <a href="mailto:gurkanerdogdu@yahoo.com">Gurkan Erdogdu</a>
* @author <a href="mailto:jesper.pedersen@jboss.org">Jesper Pedersen</a>
*/
public abstract class AbstractConnectionManager implements ConnectionManager
{
/** Log instance */
private final Logger log = Logger.getLogger(getClass());
/** Log trace */
private final boolean trace;
/** The pool */
private Pool pool;
/** Security domain */
private String securityDomain;
/** SubjectFactory */
private SubjectFactory subjectFactory;
/** The flush strategy */
private FlushStrategy flushStrategy;
/** Number of retry to allocate connection */
private int allocationRetry;
/** Interval between retries */
private long allocationRetryWaitMillis;
/** Startup/ShutDown flag */
private final AtomicBoolean shutdown = new AtomicBoolean(false);
/** User transaction registry */
private UserTransactionRegistry userTransactionRegistry;
/** Cached connection manager */
private CachedConnectionManager cachedConnectionManager;
/** Jndi name */
private String jndiName;
/**
* Creates a new instance of connection manager.
*/
protected AbstractConnectionManager()
{
this.trace = log.isTraceEnabled();
}
/**
* Gets log.
* @return log instance
*/
protected Logger getLog()
{
return log;
}
/**
* Set the pool.
* @param pool the pool
*/
public void setPool(Pool pool)
{
this.pool = pool;
}
/**
* Get the pool.
* @return the pool
*/
public Pool getPool()
{
return pool;
}
/**
* Set the user transaction registry
* @param utr The value
*/
public void setUserTransactionRegistry(UserTransactionRegistry utr)
{
this.userTransactionRegistry = utr;
}
/**
* Sets cached connection manager.
* @param cachedConnectionManager cached connection manager
*/
public void setCachedConnectionManager(CachedConnectionManager cachedConnectionManager)
{
this.cachedConnectionManager = cachedConnectionManager;
if (userTransactionRegistry != null && cachedConnectionManager != null)
userTransactionRegistry.addListener(cachedConnectionManager);
}
/**
* Gets cached connection manager.
* @return cached connection manager
*/
public CachedConnectionManager getCachedConnectionManager()
{
return cachedConnectionManager;
}
/**
* Sets shut down flag.
* @param shutDown shut down flag
*/
public void setShutDown(boolean shutDown)
{
this.shutdown.set(shutDown);
if (shutDown)
{
if (userTransactionRegistry != null && cachedConnectionManager != null)
userTransactionRegistry.removeListener(cachedConnectionManager);
}
}
/**
* Gets jndi name.
* @return jndi name
*/
public String getJndiName()
{
return jndiName;
}
/**
* Sets jndi name.
* @param jndiName jndi name
*/
public void setJndiName(String jndiName)
{
this.jndiName = jndiName;
}
/**
* {@inheritDoc}
*/
public String getSecurityDomain()
{
return securityDomain;
}
/**
* Sets security domain
* @param securityDomain security domain
*/
public void setSecurityDomain(String securityDomain)
{
this.securityDomain = securityDomain;
}
/**
* {@inheritDoc}
*/
public SubjectFactory getSubjectFactory()
{
return subjectFactory;
}
/**
* Sets subject factory.
* @param subjectFactory subject factory
*/
public void setSubjectFactory(SubjectFactory subjectFactory)
{
this.subjectFactory = subjectFactory;
}
/**
* Get the flush strategy
* @return The value
*/
public FlushStrategy getFlushStrategy()
{
return flushStrategy;
}
/**
* Set the flush strategy
* @param v The value
*/
public void setFlushStrategy(FlushStrategy v)
{
this.flushStrategy = v;
}
/**
* Gets managed connection factory.
* @return managed connection factory
*/
public javax.resource.spi.ManagedConnectionFactory getManagedConnectionFactory()
{
if (pool == null)
{
if (trace)
{
log.trace("No pooling strategy found! for connection manager : " + this);
return null;
}
}
else
{
return pool.getManagedConnectionFactory();
}
return null;
}
/**
* Set the number of allocation retries
* @param number retry number
*/
public void setAllocationRetry(int number)
{
if (number >= 0)
allocationRetry = number;
}
/**
* Get the number of allocation retries
* @return The number of retries
*/
public int getAllocationRetry()
{
return allocationRetry;
}
/**
* Set the wait time between each allocation retry
* @param millis wait in ms
*/
public void setAllocationRetryWaitMillis(long millis)
{
if (millis > 0)
allocationRetryWaitMillis = millis;
}
/**
* Get the wait time between each allocation retry
* @return The millis
*/
public long getAllocationRetryWaitMillis()
{
return allocationRetryWaitMillis;
}
/**
* Public for use in testing pooling functionality by itself.
* called by both allocateConnection and reconnect.
*
* @param subject a <code>Subject</code> value
* @param cri a <code>ConnectionRequestInfo</code> value
* @return a <code>ManagedConnection</code> value
* @exception ResourceException if an error occurs
*/
public ConnectionListener getManagedConnection(Subject subject, ConnectionRequestInfo cri) throws ResourceException
{
return getManagedConnection(null, subject, cri);
}
/**
* Get the managed connection from the pool.
*
* @param transaction the transaction for track by transaction
* @param subject the subject
* @param cri the ConnectionRequestInfo
* @return a managed connection
* @exception ResourceException if an error occurs
*/
protected ConnectionListener getManagedConnection(Transaction transaction, Subject subject,
ConnectionRequestInfo cri) throws ResourceException
{
ResourceException failure = null;
if (shutdown.get())
{
throw new ResourceException("The connection manager is shutdown " + jndiName);
}
// First attempt
try
{
return pool.getConnection(transaction, subject, cri);
}
catch (ResourceException e)
{
failure = e;
// Retry?
if (allocationRetry != 0)
{
for (int i = 0; i < allocationRetry; i++)
{
if (shutdown.get())
{
throw new ResourceException("The connection manager is shutdown " + jndiName);
}
if (trace)
{
log.trace("Attempting allocation retry for cri=" + cri);
}
try
{
if (allocationRetryWaitMillis != 0)
{
Thread.sleep(allocationRetryWaitMillis);
}
return pool.getConnection(transaction, subject, cri);
}
catch (ResourceException re)
{
failure = re;
}
catch (InterruptedException ie)
{
JBossResourceException.rethrowAsResourceException("getManagedConnection retry wait was interrupted " +
jndiName, ie);
}
}
}
}
// If we get here all retries failed, throw the lastest failure
throw new ResourceException("Unable to get managed connection for " + jndiName, failure);
}
/**
* Kill given connection listener wrapped connection instance.
* @param bcl connection listener that wraps connection
* @param kill kill connection or not
*/
public void returnManagedConnection(org.jboss.jca.core.api.connectionmanager.listener.ConnectionListener bcl,
boolean kill)
{
// Hack - We know that we can type cast it
ConnectionListener cl = (ConnectionListener)bcl;
Pool localStrategy = cl.getPool();
if (localStrategy != pool)
{
kill = true;
}
try
{
if (!kill && cl.getState().equals(ConnectionState.NORMAL))
{
cl.tidyup();
}
}
catch (Throwable t)
{
log.warn("Error during tidy up connection" + cl, t);
kill = true;
}
try
{
localStrategy.returnConnection(cl, kill);
}
catch (ResourceException re)
{
// We can receive notification of an error on the connection
// before it has been assigned to the pool. Reduce the noise for
// these errors
if (kill)
{
log.debug("resourceException killing connection (error retrieving from pool?)", re);
}
else
{
log.warn("resourceException returning connection: " + cl.getManagedConnection(), re);
}
}
}
/**
* {@inheritDoc}
*/
public Object allocateConnection(ManagedConnectionFactory mcf, ConnectionRequestInfo cri) throws ResourceException
{
//Check for pooling!
if (pool == null)
{
throw new ResourceException("You are trying to use a connection factory that has been shut down: " +
"ManagedConnectionFactory is null.");
}
//it is an explicit spec requirement that equals be used for matching rather than ==.
if (!pool.getManagedConnectionFactory().equals(mcf))
{
throw new ResourceException("Wrong ManagedConnectionFactory sent to allocateConnection!");
}
// Pick a managed connection from the pool
Subject subject = getSubject();
ConnectionListener cl = getManagedConnection(subject, cri);
// Tell each connection manager the managed connection is active
reconnectManagedConnection(cl);
// Ask the managed connection for a connection
Object connection = null;
try
{
connection = cl.getManagedConnection().getConnection(subject, cri);
}
catch (Throwable t)
{
try
{
managedConnectionDisconnected(cl);
}
catch (ResourceException re)
{
log.trace("Get exception from managedConnectionDisconnected, maybe delist() have problem" + re);
returnManagedConnection(cl, true);
}
JBossResourceException.rethrowAsResourceException(
"Unchecked throwable in ManagedConnection.getConnection() cl=" + cl, t);
}
// Associate managed connection with the connection
registerAssociation(cl, connection);
if (cachedConnectionManager != null)
{
cachedConnectionManager.registerConnection(this, cl, connection, cri);
}
return connection;
}
/**
* {@inheritDoc}
*/
public void disconnect(Collection<ConnectionRecord> conRecords, Set<String> unsharableResources)
throws ResourceException
{
// if we have an unshareable connection do not remove the association
// nothing to do
if (unsharableResources.contains(jndiName))
{
log.trace("disconnect for unshareable connection: nothing to do");
return;
}
Set<ConnectionListener> cls = new HashSet<ConnectionListener>(conRecords.size());
for (Iterator<ConnectionRecord> i = conRecords.iterator(); i.hasNext();)
{
ConnectionRecord cr = i.next();
ConnectionListener cl = cr.getConnectionListener();
cr.setConnectionListener(null);
unregisterAssociation(cl, cr.getConnection());
if (!cls.contains(cl))
{
cls.add(cl);
}
}
for (Iterator<ConnectionListener> i = cls.iterator(); i.hasNext();)
{
disconnectManagedConnection(i.next());
}
}
/**
* {@inheritDoc}
*/
public void reconnect(Collection<ConnectionRecord> conns, Set<String> unsharableResources) throws ResourceException
{
// if we have an unshareable connection the association was not removed
// nothing to do
if (unsharableResources.contains(jndiName))
{
log.trace("reconnect for unshareable connection: nothing to do");
return;
}
Map<ConnectionRequestInfo, ConnectionListener> criToCLMap =
new HashMap<ConnectionRequestInfo, ConnectionListener>(conns.size());
for (Iterator<ConnectionRecord> i = conns.iterator(); i.hasNext();)
{
ConnectionRecord cr = i.next();
if (cr.getConnectionListener() != null)
{
//This might well be an error.
log.warn("reconnecting a connection handle that still has a managedConnection! "
+ cr.getConnectionListener().getManagedConnection() + " " + cr.getConnection());
}
ConnectionListener cl = criToCLMap.get(cr.getCri());
if (cl == null)
{
cl = getManagedConnection(getSubject(), cr.getCri());
criToCLMap.put(cr.getCri(), cl);
//only call once per managed connection, when we get it.
reconnectManagedConnection(cl);
}
cl.getManagedConnection().associateConnection(cr.getConnection());
registerAssociation(cl, cr.getConnection());
cr.setConnectionListener(cl);
}
}
/**
* Unregister association.
* @param cl connection listener
* @param c connection
*/
//does NOT put the mc back in the pool if no more handles. Doing so would introduce a race condition
//whereby the mc got back in the pool while still enlisted in the tx.
//The mc could be checked out again and used before the delist occured.
public void unregisterAssociation(ConnectionListener cl, Object c)
{
cl.unregisterConnection(c);
}
/**
* Invoked to reassociate a managed connection.
*
* @param cl the managed connection
* @throws ResourceException for exception
*/
protected void reconnectManagedConnection(ConnectionListener cl) throws ResourceException
{
try
{
managedConnectionReconnected(cl);
}
catch (Throwable t)
{
disconnectManagedConnection(cl);
JBossResourceException.rethrowAsResourceException("Unchecked throwable in managedConnectionReconnected() cl="
+ cl, t);
}
}
/**
* Invoked when a managed connection is no longer associated
*
* @param cl the managed connection
*/
protected void disconnectManagedConnection(ConnectionListener cl)
{
try
{
managedConnectionDisconnected(cl);
}
catch (Throwable t)
{
log.warn("Unchecked throwable in managedConnectionDisconnected() cl=" + cl, t);
}
}
/**
* For polymorphism.
* <p>
*
* Do not invoke directly, use reconnectManagedConnection
* which does the relevent exception handling
* @param cl connection listener
* @throws ResourceException for exception
*/
protected void managedConnectionReconnected(ConnectionListener cl) throws ResourceException
{
//Nothing as default
}
/**
* For polymorphism.
* <p>
*
* Do not invoke directly, use disconnectManagedConnection
* which does the relevent exception handling
* @param cl connection listener
* @throws ResourceException for exception
*/
protected void managedConnectionDisconnected(ConnectionListener cl) throws ResourceException
{
//Nothing as default
}
/**
* Register connection with connection listener.
* @param cl connection listener
* @param c connection
* @throws ResourceException exception
*/
private void registerAssociation(ConnectionListener cl, Object c) throws ResourceException
{
cl.registerConnection(c);
}
/**
* {@inheritDoc}
*/
public abstract void transactionStarted(Collection<ConnectionRecord> conns) throws SystemException;
/**
* {@inheritDoc}
*/
public abstract boolean isTransactional();
/**
* {@inheritDoc}
*/
public abstract TransactionIntegration getTransactionIntegration();
/**
* Gets subject.
* @return subject
*/
private Subject getSubject()
{
Subject subject = null;
if (subjectFactory != null && securityDomain != null)
{
subject = subjectFactory.createSubject(securityDomain);
Set<PasswordCredential> credentials = subject.getPrivateCredentials(PasswordCredential.class);
if (credentials != null && credentials.size() > 0)
{
ManagedConnectionFactory pcMcf = getManagedConnectionFactory();
for (PasswordCredential pc : credentials)
{
pc.setManagedConnectionFactory(pcMcf);
}
}
}
log.tracef("Subject: %s", subject);
return subject;
}
/**
* Write the object to the stream -- THIS IS NOT SUPPORTED
* @param out The stream
* @exception IOException Thrown in case of an error
*/
private void writeObject(ObjectOutputStream out) throws IOException
{
throw new IOException("This method is not supported");
}
/**
* Read the object from the stream -- THIS IS NOT SUPPORTED
* @param in The stream
* @exception IOException Thrown in case of an error
* @exception ClassNotFoundException Thrown if a class can't be resolved
*/
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException
{
throw new IOException("This method is not supported");
}
/**
* Read the object -- THIS IS NOT SUPPORTED
* @exception ObjectStreamException Thrown in case of an error
*/
private void readObjectNoData() throws ObjectStreamException
{
throw new NotSerializableException("This method is not supported");
}
}