Author: clebert.suconic(a)jboss.com Date: 2010-08-30 22:31:32 -0400 (Mon, 30 Aug 2010) New Revision: 9612 Modified: branches/Branch_2_1/docs/user-manual/zh/appserver-integration.xml branches/Branch_2_1/docs/user-manual/zh/persistence.xml branches/Branch_2_1/merge-activity.txt branches/Branch_2_1/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java branches/Branch_2_1/src/main/org/hornetq/jms/client/HornetQSession.java branches/Branch_2_1/src/main/org/hornetq/jms/client/JMSMessageListenerWrapper.java branches/Branch_2_1/src/main/org/hornetq/ra/inflow/HornetQMessageHandler.java Log: Merging -r9590:9598 (including https://jira.jboss.org/browse/HORNETQ-495) from trunk Modified: branches/Branch_2_1/docs/user-manual/zh/appserver-integration.xml =================================================================== --- branches/Branch_2_1/docs/user-manual/zh/appserver-integration.xml 2010-08-30 14:18:38 UTC (rev 9611) +++ branches/Branch_2_1/docs/user-manual/zh/appserver-integration.xml 2010-08-31 02:31:32 UTC (rev 9612) @@ -627,6 +627,17 @@ <entry>Integer</entry> <entry>线程池的大小</entry> </row> + <row> + <entry>SetupAttempts</entry> + <entry>Integer</entry> + <entry>尝试建立JMS连接的次数（默认值是10。-1表示无限次进行尝试）。有时MDB在部署时相关的JMS资源还没有准备好，这时通过多次的 +　　　　　　　　　　　　　　　　　　　尝试直到JMS资源连接上为止。只适用于内部（Inbound）连接的情况。</entry> + </row> + <row> + <entry>SetupInterval</entry> + <entry>Long</entry> + <entry>每两次相邻尝试之间的时间间隔，以毫秒为单位（默认值为2000毫秒）。只适用于内部（Inbound）连接的情况。</entry> + </row> </tbody> </tgroup> </informaltable> Modified: branches/Branch_2_1/docs/user-manual/zh/persistence.xml =================================================================== --- branches/Branch_2_1/docs/user-manual/zh/persistence.xml 2010-08-30 14:18:38 UTC (rev 9611) +++ branches/Branch_2_1/docs/user-manual/zh/persistence.xml 2010-08-31 02:31:32 UTC (rev 9612) @@ -56,23 +56,32 @@ </listitem> </itemizedlist> <para>标准的HornetQ核心服务器使用了两种日志：</para> - <itemizedlist> + <itemizedlist id="persistence.journallist"> <listitem> <para>绑定日志</para> <para>这个日志用来保存与绑定有关的数据。其中包括在HornetQ上部署的队列及其属性，还有ID序列计数器。 </para> <para>绑定日志是一个NIO型日志。与消息日志相比它的呑吐量是比较低的。</para> + <para>这种日志文件的名字采用<literal>hornetq-bindings</literal>作为前缀。每个文件都有 + <literal>bindings</literal>这样的扩展。文件大小是<literal + >1048576</literal>，它的位置在bindings文件夹下。</para> </listitem> <listitem> <para>JMS日志</para> <para>这个日志保存所有JMS相关的数据，包括JMS队列，话题及连接工厂，以及它们的JNDI绑定信息。</para> <para>通过管理接口创建的JMS资源将被保存在这个日志中。但是通过配置文件配置的资源则不保存。只有使用JMS时JMS的日志 才被创建。</para> + <para>这种日志文件的名字采用<literal>hornetq-jms</literal>作为前缀。每个文件都有 + <literal>jms</literal>这样的扩展。文件大小是<literal + >1048576</literal>，它的位置在bindings文件夹下。</para> </listitem> <listitem> <para>消息日志</para> <para>这个日志用来存贮所有消息相关的数据，包括消息本身和重复ID缓存。</para> <para>默认情况下HornetQ总是优先使用AIO型日志。如果AIO型日志不可用（比如在非Linux平台上运行，或系统内核版本不同） 它将自动使用NIO型日志。</para> + <para>这种日志文件的名字采用<literal>hornetq-data</literal>。作为前缀。每个文件都有 + 　<literal>hq</literal>作为扩展名。默认的文件大小是 <literal + >10485760</literal> (可配置)。文件保存在journal文件夹下。</para> </listitem> </itemizedlist> <para>对于超大消息，Hornet将它们保存在消息日志之外的地方。详见<xref linkend="large-messages"/>.</para> @@ -220,4 +229,35 @@ 参数设为<literal>false</literal>即可。 </para> <para>注意如果你将该参数设为 false来关闭持久化，就意味着所有的绑定数据、消息数据、超大消息数据、重复ID缓冲以及转移（paging）数据都将不会被持久。</para> </section> + <section id="persistence.importexport"> + <title>导入入/导出日志数据</title> + <para>有时你需要使用导入／导出工具来查看日志文件的记录。这个导入／导出工具类在hornetq-core.jar文件中。 +　　　 使用以下命令可以将日志文件导出为文本文件：</para> + <para><literal>java -cp hornetq-core.jar org.hornetq.core.journal.impl.ExportJournal + &lt;JournalDirectory> &lt;JournalPrefix> &lt;FileExtension> &lt;FileSize> + &lt;FileOutput></literal></para> + <para>要将日志文件导入，使用下面的命令（注意你需要netty.jar）：</para> + <para><literal>java -cp hornetq-core.jar:netty.jar org.hornetq.core.journal.impl.ImportJournal + &lt;JournalDirectory> &lt;JournalPrefix> &lt;FileExtension> &lt;FileSize> + &lt;FileInput></literal></para> + <itemizedlist> + <listitem> + <para>JournalDirectory：文件的位置，如./hornetq/data/journal</para> + </listitem> + <listitem> + <para>JournalPrefix： 日志文件的前缀。<link linkend="persistence.journallist">这里</link>有关于前缀的详细描述。</para> + </listitem> + <listitem> + <para>FileExtension： 文件的扩展名。详细讨论参见<link linkend="persistence.journallist">这里</link>。 + </para> + </listitem> + <listitem> + <para>FileSize：日志文件的大小。详细讨论参见<link linkend="persistence.journallist">这里</link>。</para> + </listitem> + <listitem> + <para>FileOutput：输出的文本文件名。</para> + </listitem> + </itemizedlist> + </section> + </chapter> Modified: branches/Branch_2_1/merge-activity.txt =================================================================== --- branches/Branch_2_1/merge-activity.txt 2010-08-30 14:18:38 UTC (rev 9611) +++ branches/Branch_2_1/merge-activity.txt 2010-08-31 02:31:32 UTC (rev 9612) @@ -7,3 +7,5 @@ - 24-aug-2010 - clebert - merge from trunk -r9588:9590 There was also a manual copy of JournalImpl.java on this merge, since there was a minor change before that needed to be applied + +- 30-aug-2010 - clebert - merge from trunk -r9590:9598 Modified: branches/Branch_2_1/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java =================================================================== --- branches/Branch_2_1/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2010-08-30 14:18:38 UTC (rev 9611) +++ branches/Branch_2_1/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2010-08-31 02:31:32 UTC (rev 9612) @@ -1087,7 +1087,7 @@ final boolean autoCommitSends, final boolean autoCommitAcks, final boolean preAcknowledge, - final int ackBatchSize) throws HornetQException + final int ackBatchSize_xx) throws HornetQException { if (closed) { Modified: branches/Branch_2_1/src/main/org/hornetq/jms/client/HornetQSession.java =================================================================== --- branches/Branch_2_1/src/main/org/hornetq/jms/client/HornetQSession.java 2010-08-30 14:18:38 UTC (rev 9611) +++ branches/Branch_2_1/src/main/org/hornetq/jms/client/HornetQSession.java 2010-08-31 02:31:32 UTC (rev 9612) @@ -213,6 +213,11 @@ return ackMode; } + + public boolean isXA() + { + return xa; + } public void commit() throws JMSException { Modified: branches/Branch_2_1/src/main/org/hornetq/jms/client/JMSMessageListenerWrapper.java =================================================================== --- branches/Branch_2_1/src/main/org/hornetq/jms/client/JMSMessageListenerWrapper.java 2010-08-30 14:18:38 UTC (rev 9611) +++ branches/Branch_2_1/src/main/org/hornetq/jms/client/JMSMessageListenerWrapper.java 2010-08-31 02:31:32 UTC (rev 9612) @@ -42,9 +42,9 @@ private final boolean transactedOrClientAck; protected JMSMessageListenerWrapper(final HornetQSession session, - final ClientConsumer consumer, - final MessageListener listener, - final int ackMode) + final ClientConsumer consumer, + final MessageListener listener, + final int ackMode) { this.session = session; @@ -52,7 +52,7 @@ this.listener = listener; - transactedOrClientAck = ackMode == Session.SESSION_TRANSACTED || ackMode == Session.CLIENT_ACKNOWLEDGE; + transactedOrClientAck = (ackMode == Session.SESSION_TRANSACTED || ackMode == Session.CLIENT_ACKNOWLEDGE) || session.isXA(); } /** Modified: branches/Branch_2_1/src/main/org/hornetq/ra/inflow/HornetQMessageHandler.java =================================================================== --- branches/Branch_2_1/src/main/org/hornetq/ra/inflow/HornetQMessageHandler.java 2010-08-30 14:18:38 UTC (rev 9611) +++ branches/Branch_2_1/src/main/org/hornetq/ra/inflow/HornetQMessageHandler.java 2010-08-31 02:31:32 UTC (rev 9612) @@ -68,11 +68,16 @@ private boolean useLocalTx; + private boolean transacted; + private final int sessionNr; private final TransactionManager tm; - public HornetQMessageHandler(final HornetQActivation activation, final TransactionManager tm, final ClientSession session, final int sessionNr) + public HornetQMessageHandler(final HornetQActivation activation, + final TransactionManager tm, + final ClientSession session, + final int sessionNr) { this.activation = activation; this.session = session; @@ -96,14 +101,16 @@ { String subscriptionName = spec.getSubscriptionName(); String clientID = spec.getClientID(); - + // Durable sub if (clientID == null) { - throw new InvalidClientIDException("Cannot create durable subscription for " + subscriptionName + " - client ID has not been set"); + throw new InvalidClientIDException("Cannot create durable subscription for " + subscriptionName + + " - client ID has not been set"); } - SimpleString queueName = new SimpleString(HornetQDestination.createQueueNameForDurableSubscription(clientID, subscriptionName)); + SimpleString queueName = new SimpleString(HornetQDestination.createQueueNameForDurableSubscription(clientID, + subscriptionName)); QueueQuery subResponse = session.queueQuery(queueName); @@ -123,9 +130,10 @@ SimpleString oldFilterString = subResponse.getFilterString(); boolean selectorChanged = selector == null && oldFilterString != null || - oldFilterString == null && selector != null || - (oldFilterString != null && selector != null && - !oldFilterString.toString().equals(selector)); + oldFilterString == null && + selector != null || + (oldFilterString != null && selector != null && !oldFilterString.toString() + .equals(selector)); SimpleString oldTopicName = subResponse.getAddress(); @@ -155,7 +163,7 @@ } else { - queueName = activation.getTopicTemporaryQueue(); + queueName = activation.getTopicTemporaryQueue(); } } else @@ -168,6 +176,7 @@ // Create the endpoint, if we are transacted pass the sesion so it is enlisted, unless using Local TX MessageEndpointFactory endpointFactory = activation.getMessageEndpointFactory(); useLocalTx = !activation.isDeliveryTransacted() && activation.getActivationSpec().isUseLocalTx(); + transacted = activation.isDeliveryTransacted(); if (activation.isDeliveryTransacted() && !activation.getActivationSpec().isUseLocalTx()) { endpoint = endpointFactory.createEndpoint(session); @@ -201,7 +210,7 @@ { HornetQMessageHandler.log.debug("Error releasing endpoint " + endpoint, t); } - + try { consumer.close(); @@ -246,15 +255,28 @@ try { - if(activation.getActivationSpec().getTransactionTimeout() > 0 && tm != null) + if (activation.getActivationSpec().getTransactionTimeout() > 0 && tm != null) { tm.setTransactionTimeout(activation.getActivationSpec().getTransactionTimeout()); } endpoint.beforeDelivery(HornetQActivation.ONMESSAGE); beforeDelivery = true; msg.doBeforeReceive(); + + //In the transacted case the message must be acked *before* onMessage is called + + if (transacted) + { + message.acknowledge(); + } + ((MessageListener)endpoint).onMessage(msg); - message.acknowledge(); + + if (!transacted) + { + message.acknowledge(); + } + try { endpoint.afterDelivery();