标签：
Atomikos 翻译文档（英文文档来源：下载安装包中START_HERE.html）
----译者：周枫
请尊重劳动成果，转载请标明，英语水平有限，如有不准确地方请在评论中指出，谢谢
官网地址：http://www.atomikos.com/Main/WebHome
使用版本：AtomikosTransactionsEssentials-3.7.2
感谢您使用Atomikos，下面的说明文档可以让您正确使用，如果您有任何问题或者反馈，请访问我们的帮助网页http://www.atomikos.com/Main/SupportOverview，或者给我们发送邮件sales@atomikos.com。
Atomikos TransactionsEssentials 是一个为Java平台提供增值服务的并且开源类事务管理器，以下是包括在这个开源版本中的一些功能：
l 全面崩溃 / 重启恢复
l 兼容标准的SUN公司JTA API
l 嵌套事务
l 为XA和非XA提供内置的JDBC适配器
注释：XA：XA协议由Tuxedo首先提出的，并交给X/Open组织，作为资源管理器（数据库）与事务管理器的接口标准。目前，Oracle、Informix、DB2和Sybase等各大数据库厂家都提供对XA的支持。XA协议采用两阶段提交方式来管理分布式事务。XA接口提供资源管理器与事务管理器之间进行通信的标准接口。XA协议包括两套函数，以xa_开头的及以ax_开头的。
以下的函数使事务管理器可以对资源管理器进行的操作：
1）xa_open,xa_close：建立和关闭与资源管理器的连接。
2）xa_start,xa_end：开始和结束一个本地事务。
3）xa_prepare,xa_commit,xa_rollback：预提交、提交和回滚一个本地事务。
4）xa_recover：回滚一个已进行预提交的事务。
5）ax_开头的函数使资源管理器可以动态地在事务管理器中进行注册，并可以对XID(TRANSACTION IDS)进行操作。
6）ax_reg,ax_unreg；允许一个资源管理器在一个TMS(TRANSACTION MANAGER SERVER)中动态注册或撤消注册。
l 内置的JMS适配器XA-capable JMS队列连接器
注释：JMS:jms即Java消息服务（Java Message Service）应用程序接口是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。
l 通过XA API兼容第三方适配器
l 更好的整合您的项目
l 集成Hibernate
Atomikos TransactionsEssentials 是一个可靠的库，可以加入到您的Java应用程序，也就是说为了使用这个产品，您必须添加一些jar文件（包括在dist和lib文件夹下）到您的应用程序或者应用程序服务器。
请注意：Atomikos TransactionsEssentials是一个非常快速的嵌入式事务管理器，这就意味着，您不需要另外启动一个单独的事务管理器进程（不要查找任何的bin文件夹）。相反，您的应用服务器将有它自己的intra-VM事务管理器。
配置需求：至少Java1.5 jdk，并且最少128M的内存
性能优化：尽管这个软件有着很大的优势，但是想要更好的发挥其作用，可以按以下的方法优化：
l 更高的内存，意味着更高的吞吐量（每秒的事务数目）
l 使连接池尽可能的大
l 一旦你不需要的连接请马上关闭它们。不要把你的应用程序放在缓存里，让内部连接池为你做这些，这将促使更高效的连接使用
l 不要让活动的事务闲置：终止所有情况下的事务，尤其是在异常报错情况下的事务。这将减少数据库的锁定时间，并且最大效率的处理启用的使用。
如果想获取这些细节的更多信息，也要参阅文档说明部分。
值得注意的是，在我们所有的压力测试中，Atomikos TransactionsEssentials比J2EE的web容器更高效的吞吐量。这些测量值包括日志记录的高效的事务状态，同样，在我们所有的测量中，包括XA和non-XA，高效的效率是一样的。
在J2SE中使用Atomikos Transactions Essentials,只需要按以下步骤
# 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
/**
* 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 compatibility
this.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;
}
}
/**
* 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 26005
res.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 handle
Connection proxy = super.getConnection ( hmsg );
// here we are certain that proxy is not null -> increase the use count
DynamicProxy 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;
}
}
/**
* 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 javadocs
setTimeToLive ( 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 names
if ( 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 listening
setMessageListener ( 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;
}
}
注意：虽然这个版本包含了特定的第三方产品，如RDBMS软件和JMS代理软件，请注意，AtomikosTransactionsEssentials绝不是仅限制在这些特定的产品，除了RDBMS软件和JMS代理软件大多数的其他软件依然兼容。
J2SE实例：examples/jse这个文件夹包含带有源代码的各种各样的例子。可以执行这个脚本来运行这个程序（示例程序只有文本输出，而没有图形化界面输出）
运行示例前最好安装Ant，然后打开命令窗口在 examples/jse 文件夹并且输入“ant”。
Linux/Unix/Mac OSX Note：在这些系统，你必须在终端输入‘chmod u+x *.sh‘命令，否则，示例将无法运行。
解决问题：偶尔，一些示例可能不能再继续启动使用（尽管在开始阶段可以正常运行），这个通常是由于进程原因引起的（比如JMS代理后台环境），关闭这期间的所有进程便会解决这个问题，或者，重新启动计算机，也可以解决这个问题。
在Spring中使用：Atomikos TransactionEssentials可以很方便的于Spring相结合，运行您程序使用企业级的J2EE应用程序，而不需要EJB甚至EJB容器。另外，我们额外提供了一个强大的功能，JMS消息驱动添加到Spring的内置特性。
查看examples/spring文件夹，将展示Atomikos TransactionEssentials如何配置基于Spring的应用程序。
在J2EE的web容器中使用：在您的J2EE应用程序环境中使用Atomikos TransactionsEssentials最简单的方法是通过Spring作为pico-container（组件容器），如果想获取更多信息，请到Spring处查看。
Javadoc：
http://www.atomikos.com/downloads/transactions-essentials/com/atomikos/AtomikosTransactionsEssentials/javadoc/3.7/index.html
标签：
原文地址：http://www.cnblogs.com/duanxz/p/4672551.html
