創建或者自定義你應用的transactions.properties（或者jta.properties）文件（事務管理器的配置），然后將它放入到classpath中，安裝文件夾中包涵一個實例文件；在properties文件中注釋(#)后面的是默認值，取消一行并且改變默認值。 # SAMPLE PROPERTIES FILE FOR THE TRANSACTION SERVICE
# THIS FILE ILLUSTRATES THE DIFFERENT SETTINGS FOR THE TRANSACTION MANAGER
# UNCOMMENT THE ASSIGNMENTS TO OVERRIDE DEFAULT VALUES;# Required: factory implementation class of the transaction core.
# NOTE: there is no default for this, so it MUST be specified!
#
com.atomikos.icatch.service=com.atomikos.icatch.standalone.UserTransactionServiceFactory# Set base name of file where messages are output
# (also known as the 'console file').
#
# com.atomikos.icatch.console_file_name = tm.out# Size limit (in bytes) for the console file;
# negative means unlimited.
#
# com.atomikos.icatch.console_file_limit=-1# For size-limited console files, this option
# specifies a number of rotating files to
# maintain.
#
# com.atomikos.icatch.console_file_count=1# Set the number of log writes between checkpoints
#
# com.atomikos.icatch.checkpoint_interval=500# Set output directory where console file and other files are to be put
# make sure this directory exists!
#
# com.atomikos.icatch.output_dir = ./# Set directory of log files; make sure this directory exists!
#
# com.atomikos.icatch.log_base_dir = ./# Set base name of log file
# this name will be used as the first part of
# the system-generated log file name
#
# com.atomikos.icatch.log_base_name = tmlog# Set the max number of active local transactions
# or -1 for unlimited.
#
# com.atomikos.icatch.max_actives = 50# Set the default timeout (in milliseconds) for local transactions
#
# com.atomikos.icatch.default_jta_timeout = 10000# Set the max timeout (in milliseconds) for local transactions
#
# com.atomikos.icatch.max_timeout = 300000# The globally unique name of this transaction manager process
# override this value with a globally unique name
#
# com.atomikos.icatch.tm_unique_name = tm# Do we want to use parallel subtransactions? JTA's default
# is NO for J2EE compatibility
#
# com.atomikos.icatch.serial_jta_transactions=true# If you want to do explicit resource registration then
# you need to set this value to false.
#
# com.atomikos.icatch.automatic_resource_registration=true # Set this to WARN, INFO or DEBUG to control the granularity
# of output to the console file.
#
# com.atomikos.icatch.console_log_level=WARN# Do you want transaction logging to be enabled or not?
# If set to false, then no logging overhead will be done
# at the risk of losing data after restart or crash.
#
# com.atomikos.icatch.enable_logging=true# Should two-phase commit be done in (multi-)threaded mode or not?
# Set this to false if you want commits to be ordered according
# to the order in which resources are added to the transaction.
#
# NOTE: threads are reused on JDK 1.5 or higher.
# For JDK 1.4, thread reuse is enabled as soon as the
# concurrent backport is in the classpath - see
# http://mirrors.ibiblio.org/pub/mirrors/maven2/backport-util-concurrent/backport-util-concurrent/
#
# com.atomikos.icatch.threaded_2pc=false# Should shutdown of the VM trigger shutdown of the transaction core too?
#
# com.atomikos.icatch.force_shutdown_on_vm_exit=false
?
在你的應用程序中，創建一個實例com.atomikos.icatch.jta.UserTransactionImp或者com.atomikos.icatch.jta.UserTransactionManager（使用默認的無參數構造函數） /*** Copyright (C) 2000-2010 Atomikos <info@atomikos.com>** This code ("Atomikos TransactionsEssentials"), by itself,* is being distributed under the* Apache License, Version 2.0 ("License"), a copy of which may be found at* http://www.atomikos.com/licenses/apache-license-2.0.txt .* You may not use this file except in compliance with the License.** While the License grants certain patent license rights,* those patent license rights only extend to the use of* Atomikos TransactionsEssentials by itself.** This code (Atomikos TransactionsEssentials) contains certain interfaces* in package (namespace) com.atomikos.icatch* (including com.atomikos.icatch.Participant) which, if implemented, may* infringe one or more patents held by Atomikos.* It should be appreciated that you may NOT implement such interfaces;* licensing to implement these interfaces must be obtained separately from Atomikos.** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.*/package com.atomikos.icatch.jta;import java.io.Serializable;import javax.naming.NamingException;
import javax.naming.Reference;
import javax.naming.Referenceable;
import javax.transaction.NotSupportedException;
import javax.transaction.SystemException;
import javax.transaction.TransactionManager;
import javax.transaction.UserTransaction;import com.atomikos.icatch.admin.imp.SimpleLogAdministrator;
import com.atomikos.icatch.config.TSInitInfo;
import com.atomikos.icatch.config.UserTransactionService;
import com.atomikos.icatch.config.UserTransactionServiceImp;
import com.atomikos.util.SerializableObjectFactory;/*** * * Our UserTransaction implementation for J2SE transactions. This class is* special in that it automatically starts up and recover the transaction* service on first use. <b>Note: don't use this class in J2EE applications in* order to avoid starting different transaction engines in the same application* server! J2EE applications should use J2eeUserTransaction instead.</b>*/public class UserTransactionImp implements UserTransaction, Serializable,Referenceable
{private transient TransactionManager txmgr_;/*** No-argument constructor.*/public UserTransactionImp (){}/*** Referenceable mechanism requires later setup of txmgr_, otherwise binding* into JNDI already requires that TM is running.*/private void checkSetup (){// REMOVED FOLLOWING IF CHECK: DON'T CACHE THE TXMGR TO MAKE INSTANCES// RESILIENT TO RESTART IN TOMCAT. OTHERWISE, CLIENT APPS SEE THEIR// USERTX REFERENCES INVALIDATED AND THIS IS INTOLERABLE// if ( txmgr_ == null ) {// txmgr_ = TransactionManagerImp.getTransactionManager();synchronized ( TransactionManagerImp.class ) {txmgr_ = TransactionManagerImp.getTransactionManager ();// FOLLOWING COMMENTED OUT: NEW RECOVERY IN 2.0 ALLOWS US TO START// THE TM// IF NOT ALREADY RUNNING!!!// if ( txmgr_ == null )// throw new RuntimeException ( "No transaction monitor installed?"// );// NEW FROM 2.0: if TM is not running, just start it. Any resources// can be registered later.if ( txmgr_ == null ) {UserTransactionService uts = new UserTransactionServiceImp ();TSInitInfo info = uts.createTSInitInfo ();uts.registerLogAdministrator ( SimpleLogAdministrator.getInstance () );uts.init ( info );txmgr_ = TransactionManagerImp.getTransactionManager ();}}// }}/*** @see javax.transaction.UserTransaction*/public void begin () throws NotSupportedException, SystemException{checkSetup ();txmgr_.begin ();}/*** @see javax.transaction.UserTransaction*/public void commit () throws javax.transaction.RollbackException,javax.transaction.HeuristicMixedException,javax.transaction.HeuristicRollbackException,javax.transaction.SystemException, java.lang.IllegalStateException,java.lang.SecurityException{checkSetup ();txmgr_.commit ();}/*** @see javax.transaction.UserTransaction*/public void rollback () throws IllegalStateException, SystemException,SecurityException{checkSetup ();txmgr_.rollback ();}/*** @see javax.transaction.UserTransaction*/public void setRollbackOnly () throws IllegalStateException,SystemException{checkSetup ();txmgr_.setRollbackOnly ();}/*** @see javax.transaction.UserTransaction*/public int getStatus () throws SystemException{checkSetup ();return txmgr_.getStatus ();}/*** @see javax.transaction.UserTransaction*/public void setTransactionTimeout ( int seconds ) throws SystemException{checkSetup ();txmgr_.setTransactionTimeout ( seconds );}////// IMPLEMENTATION OF REFERENCEABLE////public Reference getReference () throws NamingException{return SerializableObjectFactory.createReference ( this );}
} /*** Copyright (C) 2000-2010 Atomikos <info@atomikos.com>** This code ("Atomikos TransactionsEssentials"), by itself,* is being distributed under the* Apache License, Version 2.0 ("License"), a copy of which may be found at* http://www.atomikos.com/licenses/apache-license-2.0.txt .* You may not use this file except in compliance with the License.** While the License grants certain patent license rights,* those patent license rights only extend to the use of* Atomikos TransactionsEssentials by itself.** This code (Atomikos TransactionsEssentials) contains certain interfaces* in package (namespace) com.atomikos.icatch* (including com.atomikos.icatch.Participant) which, if implemented, may* infringe one or more patents held by Atomikos.* It should be appreciated that you may NOT implement such interfaces;* licensing to implement these interfaces must be obtained separately from Atomikos.** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.*/package com.atomikos.icatch.jta;import java.io.Serializable;import javax.naming.NamingException;
import javax.naming.Reference;
import javax.naming.Referenceable;
import javax.transaction.HeuristicMixedException;
import javax.transaction.HeuristicRollbackException;
import javax.transaction.InvalidTransactionException;
import javax.transaction.NotSupportedException;
import javax.transaction.RollbackException;
import javax.transaction.SystemException;
import javax.transaction.Transaction;
import javax.transaction.TransactionManager;
import javax.transaction.UserTransaction;import com.atomikos.icatch.config.TSInitInfo;
import com.atomikos.icatch.config.UserTransactionService;
import com.atomikos.icatch.config.UserTransactionServiceImp;
import com.atomikos.util.SerializableObjectFactory;/*** * * * * * A straightforward, zero-setup implementation of a transaction manager. J2SE* applications can use an instance of this class to get a handle to the* transaction manager, and automatically startup or recover the transaction* service on first use. <b>J2EE applications should NOT use this class in order* to avoid the concurrent use of different transaction services. For J2EE* applications, we have the class J2eeTransactionManager instead.</b>*/
public class UserTransactionManager implements TransactionManager,Serializable, Referenceable, UserTransaction
{private static final long serialVersionUID = -655789038710288096L;private transient TransactionManagerImp tm;private UserTransactionService uts;private boolean forceShutdown;private boolean startupTransactionService;private boolean closed;private void checkSetup () throws SystemException{if ( closed ) throw new SystemException ( "This UserTransactionManager instance was closed already. Call init() to reuse if desired." );synchronized ( TransactionManagerImp.class ) {tm = (TransactionManagerImp) TransactionManagerImp.getTransactionManager ();if ( tm == null ) {// not initialized -> startup TM// System.out.println ( "STARTING UP TM!!!!!!");if ( getStartupTransactionService() ) {uts = new UserTransactionServiceImp ();TSInitInfo info = uts.createTSInitInfo ();uts.init ( info );tm = (TransactionManagerImp) TransactionManagerImp.getTransactionManager ();}else {throw new SystemException ( "Transaction service not running" );}}}}public UserTransactionManager(){//startup by default, to have backward compatibilitythis.startupTransactionService = true;this.closed = false;}/*** Sets whether the transaction service should be * started if not already running. * @param startup*/public void setStartupTransactionService ( boolean startup ){this.startupTransactionService = startup;}/*** Returns true if the transaction service will * be started if not already running.* @return*/public boolean getStartupTransactionService(){return this.startupTransactionService;}/*** Performs initialization if necessary.* This will startup the TM (if not running)* and perform recovery, unless <b>getStartupTransactionService</b>* returns false.* * @throws SystemException*/public void init() throws SystemException{closed = false;checkSetup();}/*** @see javax.transaction.TransactionManager#begin()*/public void begin () throws NotSupportedException, SystemException{checkSetup ();tm.begin ();}public boolean getForceShutdown(){return forceShutdown;}/*** Sets the force shutdown mode to use during close.* @param value */public void setForceShutdown ( boolean value ){this.forceShutdown = value;}/*** @see javax.transaction.TransactionManager#commit()*/public void commit () throws RollbackException, HeuristicMixedException,HeuristicRollbackException, SecurityException,IllegalStateException, SystemException{checkSetup ();tm.commit ();}/*** @see javax.transaction.TransactionManager#getStatus()*/public int getStatus () throws SystemException{checkSetup ();return tm.getStatus ();}/*** @see javax.transaction.TransactionManager#getTransaction()*/public Transaction getTransaction () throws SystemException{checkSetup ();return tm.getTransaction ();}/*** @see javax.transaction.TransactionManager#resume(javax.transaction.Transaction)*/public void resume ( Transaction tx ) throws InvalidTransactionException,IllegalStateException, SystemException{checkSetup ();tm.resume ( tx );}/*** @see javax.transaction.TransactionManager#rollback()*/public void rollback () throws IllegalStateException, SecurityException,SystemException{checkSetup ();tm.rollback ();}/*** @see javax.transaction.TransactionManager#setRollbackOnly()*/public void setRollbackOnly () throws IllegalStateException,SystemException{checkSetup ();tm.setRollbackOnly ();}/*** @see javax.transaction.TransactionManager#setTransactionTimeout(int)*/public void setTransactionTimeout ( int secs ) throws SystemException{checkSetup ();tm.setTransactionTimeout ( secs );}/*** @see javax.transaction.TransactionManager#suspend()*/public Transaction suspend () throws SystemException{checkSetup ();return tm.suspend ();}/*** @see javax.naming.Referenceable#getReference()*/public Reference getReference () throws NamingException{return SerializableObjectFactory.createReference ( this );}/*** Closes the transaction service, but only if it was * implicitly started via this instance.* In other words, if the transaction service was started* in another way then this method will not do anything.**/public void close(){if ( uts != null ) {uts.shutdown ( forceShutdown );uts = null;}closed = true;}}
?
對于JDBC，使用我們的一個實例com.atomikos.jdbc.AtomikosDataSourceBean或者，對于non-XA驅動，可以使用com.atomikos.jdbc.nonxa.AtomikosNonXADataSourceBean /*** Copyright (C) 2000-2010 Atomikos <info@atomikos.com>** This code ("Atomikos TransactionsEssentials"), by itself,* is being distributed under the* Apache License, Version 2.0 ("License"), a copy of which may be found at* http://www.atomikos.com/licenses/apache-license-2.0.txt .* You may not use this file except in compliance with the License.** While the License grants certain patent license rights,* those patent license rights only extend to the use of* Atomikos TransactionsEssentials by itself.** This code (Atomikos TransactionsEssentials) contains certain interfaces* in package (namespace) com.atomikos.icatch* (including com.atomikos.icatch.Participant) which, if implemented, may* infringe one or more patents held by Atomikos.* It should be appreciated that you may NOT implement such interfaces;* licensing to implement these interfaces must be obtained separately from Atomikos.** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.*/package com.atomikos.jdbc;
import java.util.Enumeration;
import java.util.Properties;import javax.sql.XADataSource;import com.atomikos.beans.PropertyUtils;
import com.atomikos.datasource.RecoverableResource;
import com.atomikos.datasource.xa.jdbc.JdbcTransactionalResource;
import com.atomikos.icatch.system.Configuration;
import com.atomikos.jdbc.AbstractDataSourceBean;
import com.atomikos.util.ClassLoadingHelper;/*** The preferred class for using Atomikos connection pooling. Use an instance of* this class if you want to use Atomikos JTA-enabled connection pooling. All* you need to do is construct an instance and set the required properties as* outlined below. The resulting bean will automatically register with the* transaction service (for recovery) and take part in active transactions.* All SQL done over connections (gotten from this class) will participate in JTA transactions.*/public class AtomikosDataSourceBean
extends AbstractDataSourceBean
{private static final long serialVersionUID = 1L;private Properties xaProperties = null;private String xaDataSourceClassName;private transient XADataSource xaDataSource;public AtomikosDataSourceBean() {this.xaProperties = new Properties();}protected String printXaProperties(){StringBuffer ret = new StringBuffer();if ( xaProperties != null ) {Enumeration it = xaProperties.propertyNames();ret.append ( "[" );boolean first = true;while ( it.hasMoreElements() ) {if ( ! first ) ret.append ( "," );String name = ( String ) it.nextElement();String value = xaProperties.getProperty( name);ret.append ( name ); ret.append ( "=" ); ret.append ( value );first = false;}ret.append ( "]" );}return ret.toString();}/*** Gets the properties used to* configure the XADataSource. */public Properties getXaProperties(){return xaProperties;}/*** Sets the properties (name,value pairs) used to* configure the XADataSource. Required, unless you call setXaDataSource directly.* * @param xaProperties * **/public void setXaProperties ( Properties xaProperties ) {this.xaProperties = xaProperties;}/*** Get the XADataSource class name.*/public String getXaDataSourceClassName() {return xaDataSourceClassName;}/*** Sets the fully qualified underlying XADataSource class name. Required, unless you * call setXaDataSource directly.* * @param xaDataSourceClassName*/public void setXaDataSourceClassName ( String xaDataSourceClassName ) {this.xaDataSourceClassName = xaDataSourceClassName;}/*** Gets the configured XADataSource (if any).* @return The instance, or null if none.*/public XADataSource getXaDataSource(){return xaDataSource;}/*** Sets the XADataSource directly - instead of providing the xaDataSourceClassName and xaProperties.* @param xaDataSource*/public void setXaDataSource(XADataSource xaDataSource){this.xaDataSource = xaDataSource;}protected com.atomikos.datasource.pool.ConnectionFactory doInit() throws Exception {if (xaDataSource == null){if (xaDataSourceClassName == null)throwAtomikosSQLException("Property 'xaDataSourceClassName' cannot be null");if (xaProperties == null)throwAtomikosSQLException("Property 'xaProperties' cannot be null");}if ( Configuration.isInfoLoggingEnabled() ) Configuration.logInfo(this + ": initializing with [" +" xaDataSourceClassName=" + xaDataSourceClassName + "," +" uniqueResourceName=" + getUniqueResourceName() + "," +" maxPoolSize=" + getMaxPoolSize() + "," +" minPoolSize=" + getMinPoolSize() + "," +" borrowConnectionTimeout=" + getBorrowConnectionTimeout() + "," +" maxIdleTime=" + getMaxIdleTime() + "," +" reapTimeout=" + getReapTimeout() + "," +" maintenanceInterval=" + getMaintenanceInterval() + "," +" testQuery=" + getTestQuery() + "," +" xaProperties=" + printXaProperties() + " loginTimeout=" + getLoginTimeout() +"]");if (xaDataSource == null){Class xadsClass = null;try {xadsClass = ClassLoadingHelper.loadClass ( getXaDataSourceClassName() );} catch ( ClassNotFoundException nf ) {AtomikosSQLException.throwAtomikosSQLException ( "The class '" + getXaDataSourceClassName() +"' specified by property 'xaDataSourceClassName' could not be found in the classpath. Please make sure the spelling is correct, and that the required jar(s) are in the classpath." , nf );}Object driver = xadsClass.newInstance();if ( ! ( driver instanceof XADataSource ) ) {AtomikosSQLException.throwAtomikosSQLException ("The class '" + getXaDataSourceClassName() +"' specified by property 'xaDataSourceClassName' does not implement the required interface javax.jdbc.XADataSource. Please make sure the spelling is correct, and check your JDBC driver vendor's documentation." );}xaDataSource = (XADataSource) driver;xaDataSource.setLoginTimeout ( getLoginTimeout() );xaDataSource.setLogWriter ( getLogWriter() );PropertyUtils.setProperties(xaDataSource, xaProperties );}JdbcTransactionalResource tr = new JdbcTransactionalResource(getUniqueResourceName() , xaDataSource);com.atomikos.datasource.pool.ConnectionFactory cf = new com.atomikos.jdbc.AtomikosXAConnectionFactory(xaDataSource, tr, this);Configuration.addResource ( tr );return cf;}protected void doClose() {RecoverableResource res = Configuration.getResource ( getUniqueResourceName() );if ( res != null ) {Configuration.removeResource ( getUniqueResourceName() );//fix for case 26005res.close();}} public String toString() {String ret = "AtomikosDataSoureBean";String name = getUniqueResourceName();if ( name != null ) {ret = ret + " '" + name + "'";}return ret;}} /*** Copyright (C) 2000-2010 Atomikos <info@atomikos.com>** This code ("Atomikos TransactionsEssentials"), by itself,* is being distributed under the* Apache License, Version 2.0 ("License"), a copy of which may be found at* http://www.atomikos.com/licenses/apache-license-2.0.txt .* You may not use this file except in compliance with the License.** While the License grants certain patent license rights,* those patent license rights only extend to the use of* Atomikos TransactionsEssentials by itself.** This code (Atomikos TransactionsEssentials) contains certain interfaces* in package (namespace) com.atomikos.icatch* (including com.atomikos.icatch.Participant) which, if implemented, may* infringe one or more patents held by Atomikos.* It should be appreciated that you may NOT implement such interfaces;* licensing to implement these interfaces must be obtained separately from Atomikos.** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.*/package com.atomikos.jdbc.nonxa;import java.sql.Connection;
import java.sql.SQLException;import com.atomikos.datasource.pool.ConnectionFactory;
import com.atomikos.icatch.HeuristicMessage;
import com.atomikos.icatch.system.Configuration;
import com.atomikos.jdbc.AbstractDataSourceBean;
import com.atomikos.util.DynamicProxy;/*** * A Bean class for DataSource access to non-XA JDBC implementations. * Instances are JTA transaction-aware and can rollback the work done * over multiple connections (provided that all work was done in one and the same thread).* **/
public class AtomikosNonXADataSourceBean extends AbstractDataSourceBean
{private static final long serialVersionUID = 1L;private String url;private String user;private String password;private String driverClassName;private boolean readOnly;/*** Sets the URL to use for getting connections. Required.* * @param url*/public void setUrl ( String url ) {this.url = url;}/*** Gets the URL to connect. */public String getUrl(){return url;}/*** Marks this datasource as being used for read-only work. Optional.* * Setting this to true will avoid warnings/errors upon recovery. ReadOnly mode* is intended to avoid XA configuration of databases where no updates are* being done.* * @param readOnly Defaults to false.*/public void setReadOnly ( boolean readOnly ) {this.readOnly = readOnly;}/*** @return Whether or not this datasource is marked as readOnly.*/public boolean getReadOnly(){return readOnly;}/*** @return The password.*/public String getPassword (){return password;}/*** Sets the password to use.* * @param string*/public void setPassword ( String string ){password = string;}/*** Set the user name to get connections with.* * @param string*/public void setUser ( String string ){user = string;}/*** @return The URL to connect with.*/public String getUser (){return user;}/*** * @return The DriverManager class name.*/public String getDriverClassName (){return driverClassName;}/*** Sets the driver class name to be used by the DriverManager. Required.* * @param string*/public void setDriverClassName ( String string ){driverClassName = string;}protected void doClose() {//nothing to do}protected ConnectionFactory doInit() throws Exception {AtomikosNonXAConnectionFactory ret = null;if ( Configuration.isInfoLoggingEnabled() ) Configuration.logInfo(this + ": initializing with [" +" uniqueResourceName=" + getUniqueResourceName() + "," +" maxPoolSize=" + getMaxPoolSize() + "," +" minPoolSize=" + getMinPoolSize() + "," +" borrowConnectionTimeout=" + getBorrowConnectionTimeout() + "," +" maxIdleTime=" + getMaxIdleTime() + "," +" reapTimeout=" + getReapTimeout() + "," +" maintenanceInterval=" + getMaintenanceInterval() + "," +" testQuery=" + getTestQuery() + "," +" driverClassName=" + getDriverClassName() + "," +" user=" + getUser() + "," +" url=" + getUrl() + " loginTimeout=" + getLoginTimeout() +"]");ret = new com.atomikos.jdbc.nonxa.AtomikosNonXAConnectionFactory ( this , url , driverClassName , user , password , getLoginTimeout() , readOnly ) ;ret.init();return ret;}public synchronized Connection getConnection ( HeuristicMessage hmsg ) throws SQLException{if ( Configuration.isInfoLoggingEnabled() ) Configuration.logInfo ( this + ": getConnection ( " + hmsg + " )..." );init();//let pool take care of reusing an existing handleConnection proxy = super.getConnection ( hmsg ); // here we are certain that proxy is not null -> increase the use countDynamicProxy dproxy = ( DynamicProxy ) proxy;com.atomikos.jdbc.nonxa.AtomikosThreadLocalConnection previous = (AtomikosThreadLocalConnection) dproxy.getInvocationHandler();previous.incUseCount();previous.addHeuristicMessage ( hmsg );if ( Configuration.isDebugLoggingEnabled() ) Configuration.logDebug ( this + ": returning " + proxy );return proxy;}public String toString() {String ret = "AtomikosNonXADataSourceBean";String name = getUniqueResourceName();if ( name != null ) {ret = ret + " '" + name + "'";}return ret;}}
?
對于JMS，可以使用我們的實例com.atomikos.jms.AtomikosConnectionFactoryBean,com.atomikos.jms.extra.AbstractJmsSenderTemplate（發送信息時使用）com.atomikos.jms.extra.MessageDrivenContainer（接收時使用） /*** Copyright (C) 2000-2010 Atomikos <info@atomikos.com>** This code ("Atomikos TransactionsEssentials"), by itself,* is being distributed under the* Apache License, Version 2.0 ("License"), a copy of which may be found at* http://www.atomikos.com/licenses/apache-license-2.0.txt .* You may not use this file except in compliance with the License.** While the License grants certain patent license rights,* those patent license rights only extend to the use of* Atomikos TransactionsEssentials by itself.** This code (Atomikos TransactionsEssentials) contains certain interfaces* in package (namespace) com.atomikos.icatch* (including com.atomikos.icatch.Participant) which, if implemented, may* infringe one or more patents held by Atomikos.* It should be appreciated that you may NOT implement such interfaces;* licensing to implement these interfaces must be obtained separately from Atomikos.** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.*/package com.atomikos.jms;import java.io.Serializable;import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.naming.NamingException;
import javax.naming.Reference;
import javax.naming.Referenceable;import com.atomikos.util.SerializableObjectFactory;/*** * * Common logic for the connection factory beans.**/public abstract class AbstractConnectionFactoryBean
implements Serializable, Referenceable, ConnectionFactory
{protected String resourceName_;protected String xaFactoryJndiName_;protected AbstractConnectionFactoryBean ( ){this.xaFactoryJndiName_ = "";this.resourceName_ = "someUniqueName";}/*** Sets the JNDI name of the underlying XAConnectionFactory (optional). This is* optional and an alternative to directly supplying the required factory* through setXaConnectionFactory().* * @param name* The JNDI name where the XAConnectionFactory can be found.* It is up to the client to make sure that the name exists and* points to an existing XAConnectionFactory.*/public void setXaFactoryJndiName ( String name ) {xaFactoryJndiName_ = name;}/*** Retrieve the JNDI name where the XAConnectionFactory is expected.* * @return String the name or an empty String if not set.*/public String getXaFactoryJndiName() {return xaFactoryJndiName_;}/*** Set the unique resource name for this factory (required). A unique* resource name is needed by the transaction service in order to register* and recover the underlying XA transactions. * Note: the value you set here should not exceed 45 bytes in length.* * <p><b>MQSeries NOTE:</b> For* IBM MQSeries, the name should include MQSeries_XA_RMI or the XA routines* will not work properly! * * @param name* The unique resource name.*/public void setResourceName ( String name ) {resourceName_ = name;}/*** Get the resource name.* * @return String the unique resource name as previously set.*/public String getResourceName() {return resourceName_;}public Reference getReference() throws NamingException {return SerializableObjectFactory.createReference ( this );}/*** Initialization method to register the underlying resource for recovery* and other init code. * * @throws JMSException*/public void init() throws JMSException{checkSetup();}protected abstract void checkSetup() throws JMSException;} /*** Copyright (C) 2000-2010 Atomikos <info@atomikos.com>** This code ("Atomikos TransactionsEssentials"), by itself,* is being distributed under the* Apache License, Version 2.0 ("License"), a copy of which may be found at* http://www.atomikos.com/licenses/apache-license-2.0.txt .* You may not use this file except in compliance with the License.** While the License grants certain patent license rights,* those patent license rights only extend to the use of* Atomikos TransactionsEssentials by itself.** This code (Atomikos TransactionsEssentials) contains certain interfaces* in package (namespace) com.atomikos.icatch* (including com.atomikos.icatch.Participant) which, if implemented, may* infringe one or more patents held by Atomikos.* It should be appreciated that you may NOT implement such interfaces;* licensing to implement these interfaces must be obtained separately from Atomikos.** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.*/package com.atomikos.jms.extra;import java.io.Serializable;
import java.util.Map;import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.Topic;
import javax.transaction.Status;
import javax.transaction.SystemException;import com.atomikos.icatch.jta.UserTransactionManager;
import com.atomikos.icatch.system.Configuration;
import com.atomikos.jms.AtomikosConnectionFactoryBean;
import com.atomikos.jms.AtomikosJMSException;
import com.atomikos.jms.AtomikosTransactionRequiredJMSException;/*** Common functionality for the sender templates.**/public abstract class AbstractJmsSenderTemplate
{protected AtomikosConnectionFactoryBean connectionFactoryBean;private String user;protected String password;protected Destination destination;private String destinationName;private Destination replyToDestination;private String replyToDestinationName;private int deliveryMode;private int priority;private long timeToLive;protected boolean inited;protected AbstractJmsSenderTemplate(){// set default values according to Sun's JMS javadocssetTimeToLive ( 0 );setDeliveryMode ( javax.jms.DeliveryMode.PERSISTENT );setPriority ( 4 );}protected abstract Session getOrRefreshSession ( Connection c ) throws JMSException;protected abstract Connection getOrReuseConnection() throws JMSException;protected abstract void afterUseWithoutErrors ( Connection c , Session s ) throws JMSException;protected void destroy ( Connection c , Session s)throws JMSException {try {if ( s != null ) s.close();} catch ( JMSException warn ) {Configuration.logWarning ( this + ": error closing session" , warn);}try {if ( c != null ) c.close();} catch ( JMSException warn ) {Configuration.logWarning ( this + ": error closing connection" , warn);}} protected synchronized Connection refreshConnection() throws JMSException {Connection connection = null;if ( getDestinationName() == null )throw new JMSException ( "Please call setDestination or setDestinationName first!" );if ( user != null ) {connection = connectionFactoryBean.createConnection (user, password );} else {connection = connectionFactoryBean.createConnection ();}connection.start ();return connection;}/*** Initializes the session for sending. * Call this method first.*/public void init() throws JMSException{if ( ! inited ) {if ( connectionFactoryBean == null ) throw new IllegalStateException ( "Property 'atomikosConnectionFactoryBean' must be set first!" );if ( getDestinationName() == null ) {throw new IllegalStateException ( "Property 'destination' or 'destinationName' must be set first!" );}StringBuffer msg = new StringBuffer();msg.append ( this + ":configured with [" );msg.append ( "user=" ).append ( getUser() ).append ( ", " );msg.append ( "password=" ).append ( password ).append ( ", " );msg.append ( "deliveryMode=" ).append ( getDeliveryMode() ).append ( ", " );msg.append ( "timeToLive=" ).append ( getTimeToLive() ).append ( ", " );msg.append ( "priority=" ).append ( getPriority() ).append ( ", " );msg.append ( "destination=" ).append( getDestinationName() ).append ( ", " );msg.append ( "replyToDestination=" ).append ( getReplyToDestinationName() );msg.append ( "]" );if ( Configuration.isDebugLoggingEnabled() ) Configuration.logDebug ( msg.toString() );inited = true; }}private void retrieveDestinationIfNecessary() throws JMSException {if ( getDestination() == null ) {String dName = getDestinationName();RetrieveDestinationCallback cb = new RetrieveDestinationCallback ( dName );executeCallbackInternal ( cb );setDestination ( cb.getDestination() );}}private void retrieveReplyToDestinationIfNecessary() throws JMSException {if ( getReplyToDestination() == null ) {String dName = getReplyToDestinationName();if ( dName != null ) {RetrieveDestinationCallback cb = new RetrieveDestinationCallback ( dName );executeCallbackInternal ( cb );setReplyToDestination ( cb.getDestination() );}}}/*** Sets the connection factory to use. Required.* @param connectionFactory*/public void setAtomikosConnectionFactoryBean(AtomikosConnectionFactoryBean connectionFactory) {this.connectionFactoryBean = connectionFactory;}public AtomikosConnectionFactoryBean getAtomikosConnectionFactoryBean() {return connectionFactoryBean;}public Destination getDestination() {return destination;}/*** Sets the (provider-specific) destination name in order* to lookup the destination (rather than providing one directly).* * Required, unless you set the destination directly.* * @param destinationName*/public void setDestinationName ( String destinationName ) {this.destinationName = destinationName;}/*** Sets the destination to send to. Required, unless* you set the destinationName instead.* * @param destination*/public void setDestination(Destination destination) {this.destination = destination;}private String getName(Destination d, String destinationName ) {String ret = destinationName;if ( ret == null ) {if ( d instanceof Queue ) {Queue q = ( Queue ) d;try {ret = q.getQueueName();} catch ( JMSException e ) {if ( Configuration.isDebugLoggingEnabled() ) Configuration.logDebug ( this + ": error retrieving queue name" , e );}} else if ( d instanceof Topic ) {Topic t = ( Topic ) d;try {ret = t.getTopicName();} catch ( JMSException e ) {if ( Configuration.isDebugLoggingEnabled() ) Configuration.logDebug ( this + ": error retrieving topic name" , e );}}}return ret;}protected String getDestinationName() {return getName ( getDestination() , destinationName );}protected String getReplyToDestinationName() {return getName ( getReplyToDestination() , replyToDestinationName );}/*** @return The user to connect with, or null if no explicit authentication* is to be used.*/public String getUser() {return user;}/*** If this session is used for sending request/reply messages, then this* property indicates the destination where the replies are to be sent (optional). The* session uses this to set the JMSReplyTo header accordingly. This property* can be omitted if no reply is needed.* * <p>* The replyToDestination should be in the same JMS vendor domain as the send* queue. To cross domains, configure a bridge for both the request and the* reply channels.*/public void setReplyToDestination(Destination destination) {this.replyToDestination = destination;}/*** Sets the provider-specific replyToDestinationName. Optional.* * @param replyToDestinationName*/public void setReplyToDestinationName ( String replyToDestinationName ) {this.replyToDestinationName = replyToDestinationName;} /*** Gets the replyToDestination.* * @return*/public Destination getReplyToDestination() {return replyToDestination;}/*** Set the password for explicit authentication (optional). * This is only required if* the user has also been set.* * @param password* The password.*/public void setPassword(String password) {this.password = password;}/*** Set the user to use for explicit authentication (optional). If no explicit* authentication is required then this method should not be called.* * @param user*/public void setUser(String user) {this.user = user;}protected void executeCallbackInternal ( JmsSenderTemplateCallback callback ) throws JMSException {init();Session session = null;Connection conn = null;try {conn = getOrReuseConnection();session = getOrRefreshSession ( conn ); if ( Configuration.isDebugLoggingEnabled() ) Configuration.logDebug ( "Calling callback..." );callback.doInJmsSession ( session );if ( Configuration.isDebugLoggingEnabled() ) Configuration.logDebug ( "Callback done!" );afterUseWithoutErrors ( conn , session );} catch ( AtomikosTransactionRequiredJMSException notx ) {destroy ( conn , session );String msg = "The JMS session you are using requires a JTA transaction context for the calling thread and none was found." + "\n" +"Please correct your code to do one of the following: " + "\n" + "1. start a JTA transaction before sending any message, or" + "\n" + "2. increase the maxPoolSize of the AtomikosConnectionFactoryBean to avoid transaction timeout while waiting for a connection.";Configuration.logWarning ( msg );AtomikosTransactionRequiredJMSException.throwAtomikosTransactionRequiredJMSException ( msg );} catch ( JMSException e ) {e.printStackTrace();destroy ( conn , session );String msg = this + ": error in sending JMS message";AtomikosJMSException.throwAtomikosJMSException( msg , e );}}/*** Executes an application-level call-back within the managed session.* * @param callback* @throws JMSException*/public void executeCallback(JmsSenderTemplateCallback callback) throws JMSException {init(); retrieveDestinationIfNecessary();retrieveReplyToDestinationIfNecessary();UserTransactionManager tm = new UserTransactionManager ();try {if ( tm.getStatus () != Status.STATUS_ACTIVE )throw new JMSException ("This method requires an active transaction!" );} catch ( SystemException e ) {Configuration.logWarning ( this +": error in getting transaction status", e );throw new RuntimeException ( e.getMessage () );}executeCallbackInternal ( callback );}/*** @return The deliverymode for messages sent in this session.*/public int getDeliveryMode() {return deliveryMode;}/*** @return The priority for messages sent in this session.*/public int getPriority() {return priority;}/*** @return The timeToLive for messages sent in this session.*/public long getTimeToLive() {return timeToLive;}/*** * Set the deliverymode for messages sent in this session (optional). Defaults to* persistent.* * @param*/public void setDeliveryMode(int i) {deliveryMode = i;}/*** Set the priority for messages sent in this session (optional). Defaults to 4.* * @param*/public void setPriority(int i) {priority = i;}/*** Set the time to live for messages sent in this session (optional). Defaults to 0.* * @param*/public void setTimeToLive(long l) {timeToLive = l;}/*** Sends a TextMessage.* * @param content The text as a string.* @throws JMSException */public void sendTextMessage(String content) throws JMSException {retrieveDestinationIfNecessary();retrieveReplyToDestinationIfNecessary();SendTextMessageCallback cb = new SendTextMessageCallback ( content , getDestination() , getReplyToDestination() , getDeliveryMode() , getPriority() , getTimeToLive() );executeCallback ( cb );}/*** Sends a MapMessage.* * @param content The Map to get the content from.* * @throws JMSException*/public void sendMapMessage(Map content) throws JMSException {retrieveDestinationIfNecessary();retrieveReplyToDestinationIfNecessary();SendMapMessageCallback cb = new SendMapMessageCallback ( content , getDestination() , getReplyToDestination() , getDeliveryMode() , getPriority() , getTimeToLive() );executeCallback ( cb ); }/*** Sends an ObjectMessage.* * @param content The serializable object content.* @throws JMSException*/public void sendObjectMessage(Serializable content) throws JMSException {retrieveDestinationIfNecessary();retrieveReplyToDestinationIfNecessary();SendObjectMessageCallback cb = new SendObjectMessageCallback ( content , getDestination() , getReplyToDestination() , getDeliveryMode() , getPriority() , getTimeToLive() );executeCallback ( cb ); }/*** Sends a ByteMessage.* * @param content The content as a byte array.* @throws JMSException*/public void sendBytesMessage(byte[] content) throws JMSException {retrieveDestinationIfNecessary();retrieveReplyToDestinationIfNecessary();SendBytesMessageCallback cb = new SendBytesMessageCallback ( content , getDestination() , getReplyToDestination() , getDeliveryMode() , getPriority() , getTimeToLive() );executeCallback ( cb );}/*** Closes all resources.*/public void close() {try {Connection c = getOrReuseConnection();Session s = getOrRefreshSession(c);destroy(c, s);} catch (JMSException e) {Configuration.logWarning ( this + ": error closing" , e );}connectionFactoryBean.close();}} /*** Copyright (C) 2000-2010 Atomikos <info@atomikos.com>** This code ("Atomikos TransactionsEssentials"), by itself,* is being distributed under the* Apache License, Version 2.0 ("License"), a copy of which may be found at* http://www.atomikos.com/licenses/apache-license-2.0.txt .* You may not use this file except in compliance with the License.** While the License grants certain patent license rights,* those patent license rights only extend to the use of* Atomikos TransactionsEssentials by itself.** This code (Atomikos TransactionsEssentials) contains certain interfaces* in package (namespace) com.atomikos.icatch* (including com.atomikos.icatch.Participant) which, if implemented, may* infringe one or more patents held by Atomikos.* It should be appreciated that you may NOT implement such interfaces;* licensing to implement these interfaces must be obtained separately from Atomikos.** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.*/package com.atomikos.jms.extra;import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.MessageListener;import com.atomikos.icatch.system.Configuration;
import com.atomikos.jms.AtomikosConnectionFactoryBean;/*** * A message-driven container for asynchronously receiving JMS messages* from a topic or queue, within a managed JTA transaction context.* * Upon start, an instance of this class will create a number of* concurrent sessions that listen for incoming messages on the same destination.* MessageListener instances should be thread-safe if the pool size is larger* than one. Note: in general, after start() any changed properties are only* effective on the next start() event.* * <p>* <b>IMPORTANT:</b> the transactional behaviour guarantees redelivery after failures.* As a side-effect, this can lead to so-called <em>poison messages</em>: messages* whose processing repeatedly fails due to some recurring error (for instance, a primary* key violation in the database, a NullPointerException, ...). Poison messages are problematic* because they can prevent other messages from being processed, and block the system.* * To avoid poison messages, make sure that your MessageListener implementation * only throws a <b>RuntimeException</b> when the problem is <em>transient</em>. In that* case, the system will perform rollback and the message will be redelivered* facing a clean system state. All non-transient errors (i.e., those that happen* each time a message is delivered) indicate problems at the application level* and should be dealt with by writing better application code.*/public class MessageDrivenContainer
implements MessageConsumerSessionProperties
{private static final int DEFAULT_TIMEOUT = 30;private AtomikosConnectionFactoryBean connectionFactoryBean;private MessageListener messageListener;private String user;private String password;private Destination destination;private String destinationName;private int transactionTimeout;private int poolSize;private List sessions;private boolean daemonThreads;private boolean notifyListenerOnClose;private String messageSelector;private ExceptionListener exceptionListener;private String subscriberName;private boolean noLocal;private boolean unsubscribeOnClose;private String clientID;private int receiveTimeout;public MessageDrivenContainer(){sessions = new ArrayList ();notifyListenerOnClose = false;setPoolSize ( 1 );setTransactionTimeout ( DEFAULT_TIMEOUT );}private MessageConsumerSession createSession() {return new MessageConsumerSession ( this );}/*** Sets the clientID for durable subscriptions. Optional.* * @param clientID*/public void setClientID ( String clientID ) {this.clientID = clientID;}/*** Sets the connection factory to use. Required.* @param bean*/public void setAtomikosConnectionFactoryBean ( AtomikosConnectionFactoryBean bean ){this.connectionFactoryBean = bean;}public AtomikosConnectionFactoryBean getAtomikosConnectionFactoryBean(){return connectionFactoryBean;}/*** Gets the destination. * * @return The destination, or null if not set.*/public Destination getDestination(){return destination;}/*** Sets the JMS destination to listen on (required unless the destinationName is set instead).* * @param dest*/public void setDestination ( Destination dest ){this.destination = dest;}/*** Gets the destination name.* * @return The name, or null if not set.*/public String getDestinationName(){return destinationName;}/*** Sets the JMS provider-specific destination name* (required unless the destination is set directly). * * @param destinationName*/public void setDestinationName ( String destinationName ){this.destinationName = destinationName;}/*** Sets whether threads should be daemon threads or not (optional).* Default is false.* @param value If true then threads will be daemon threads.*/public void setDaemonThreads ( boolean value ) {this.daemonThreads = value;}/*** Tests whether threads are daemon threads.* @return True if threads are deamons.*/public boolean getDaemonThreads() {return daemonThreads;}/*** * Get the message listener if any.* * @return*/public MessageListener getMessageListener() {return messageListener;}/*** Get the transaction timeout.* * @return*/public int getTransactionTimeout() {return transactionTimeout;}/*** Get the user for connecting, or null if the default user should be used.* * @return*/public String getUser() {return user;}/*** Set the message listener to use (required). * The same instance will be used for each* session in the pool, meaning that instances need to be thread-safe. Only* one listener is allowed at a time. Call this method with a null argument* to unset the listener.* * @param listener*/public void setMessageListener ( MessageListener listener ) {messageListener = listener;Iterator it = sessions.iterator ();while ( it.hasNext () ) {MessageConsumerSession s = (MessageConsumerSession) it.next ();s.setMessageListener ( listener );}}/*** Set the password if explicit authentication is needed (optional). * You need to set this if the user is also set.* * @param string*/public void setPassword ( String string ) {password = string;}/*** Set the transaction timeout in seconds (optional).* * @param seconds*/public void setTransactionTimeout ( int seconds ) {transactionTimeout = seconds;}/*** Set the user to use for explicit authentication (optional). * Don't set this property* if you want to use the default authentication.* * @param string*/public void setUser ( String string ) {user = string;}/*** Get the message selector (if any)* * @return The selector, or null if none.*/public String getMessageSelector() {return this.messageSelector;}/*** Set the message selector to use (optional).* * @param selector*/public void setMessageSelector ( String selector ) {this.messageSelector = selector;}/*** Get the size of the pool.* * @return*/public int getPoolSize() {return poolSize;}/*** Sets the size of the session pool (optional).* Default is 1.* * @param size */public void setPoolSize ( int size ) {poolSize = size;}/*** Gets the exception listener (if any). * @return Null if no ExceptionListener was set.*/public ExceptionListener getExceptionListener() {return exceptionListener;}/*** Sets the exception listener (optional). The listener will be* notified of connection-level JMS errors.* * @param exceptionListener*/public void setExceptionListener ( ExceptionListener exceptionListener ) {this.exceptionListener = exceptionListener;}/*** Test if this instance will receive sends from the same connection.* * @return*/public boolean isNoLocal() {return noLocal;}/*** Sets whether or not this topic should receive sends from the * same connection (optional). * * @param noLocal */public void setNoLocal(boolean noLocal) {this.noLocal = noLocal;}/*** Gets the subscriber name (for durable subscribers).* @return The name, or null if not set (no durable subscriber).*/public String getSubscriberName() {return subscriberName;}/*** Sets the name to use for durable subscriptions (optional).* <br>* <b>Note: this name will be appended with a suffix to ensure uniqueness* among instances in the pool. Otherwise, the JMS back-end would see * multiple instances subscribing with the same name - an error.</b>* * @param subscriberName*/public void setSubscriberName(String subscriberName) {this.subscriberName = subscriberName;}protected boolean getNoLocal() {return isNoLocal();}/*** Start listening for messages.* * @throws JMSException*/public void start() throws JMSException {if ( destination == null && destinationName == null )throw new JMSException ("MessageDrivenContainer: destination not specified" );if ( connectionFactoryBean == null )throw new JMSException ("MessageDrivenContainer: factory not set" );if ( messageListener == null )throw new JMSException ("MessageDrivenContainer: messageListener not set" );for ( int i = 0; i < poolSize; i++ ) {MessageConsumerSession s = createSession();s.setMessageListener ( messageListener );s.setPassword ( password );s.setUser ( user );s.setDestination ( destination );s.setDestinationName ( destinationName );s.setAtomikosConnectionFactoryBean ( connectionFactoryBean );s.setDaemonThreads ( daemonThreads );s.setNotifyListenerOnClose ( notifyListenerOnClose );s.setMessageSelector ( getMessageSelector () );s.setExceptionListener ( exceptionListener );s.setNoLocal( noLocal );s.setSubscriberName( subscriberName );//set subscriber name with suffix to ensure unique namesif ( getSubscriberName() != null ) s.setSubscriberName ( getSubscriberName() + "-" + i );s.setNoLocal ( getNoLocal() );s.setClientID(clientID);try {s.startListening ();// System.out.println ( "MessageDrivenContainer: started// session");} catch ( Exception e ) {Configuration.logWarning ( "Error starting pool", e );}sessions.add ( s );}// set listener again to trigger listeningsetMessageListener ( messageListener );}/*** Stop listening for messages. If <b>notifyListenerOnClose</b> is set then* calling this method will notify the listener by calling its onMessage* method with a null argument (and also without transaction context).* * This method will wait for all active receive operations to unblock, which may take* up to <b>receiveTimeout</b> seconds per active thread.*/public void stop() {Iterator it = sessions.iterator ();while ( it.hasNext () ) {MessageConsumerSession s = (MessageConsumerSession) it.next ();s.stopListening ();}}/*** Getter to check whether the listener is notified on close.* * @return*/public boolean getNotifyListenerOnClose() {return notifyListenerOnClose;}/*** Set whether the listener should be notified of close events on the pool* (optional). Default is false.* * @param b* If true, then the listener will receive a null message if the* pool is closed.*/public void setNotifyListenerOnClose ( boolean b ) {notifyListenerOnClose = b;Iterator it = sessions.iterator ();while ( it.hasNext () ) {MessageConsumerSession s = (MessageConsumerSession) it.next ();s.setNotifyListenerOnClose ( b );}}/*** Sets whether unsubscribe should be done at closing time (optional). Default is false.* * @param b If true, then unsubscribe will be done at closing time. This only applies to * durable subscribers (i.e., cases where subscriberName is set).*/public void setUnsubscribeOnClose ( boolean b ) {this.unsubscribeOnClose = b;}/*** Getter to test if unsubscribe should be called on close.*/public boolean getUnsubscribeOnClose(){return unsubscribeOnClose;}/*** Gets the receive timeout in seconds.* * @return*/public int getReceiveTimeout() {int ret = receiveTimeout;if ( ret <=0 ) ret = getTransactionTimeout()/2;return ret;}/*** Sets the receive timeout in seconds, * i.e. the number of seconds to wait for incoming messages in the message listener thread's event loop.* * This property is optional and defaults to half the transactionTimeout, but typically this should be lower * because the time required to shutdown (stop) this container will be bound by this value multiplied by * the number of threads (as indicated by <b>poolSize</b>).* * @param seconds*/public void setReceiveTimeout(int seconds) {this.receiveTimeout = seconds; }}
