From 9bb405b4de39aac7d011f0fc2f956647f4b3aa1d Mon Sep 17 00:00:00 2001 From: dsyer Date: Sat, 10 May 2008 00:19:40 +0000 Subject: [PATCH] Missing source file (Subclipse) --- .../jms/BatchMessageListenerContainer.java | 145 ++++++++++++++++++ 1 file changed, 145 insertions(+) create mode 100644 spring-batch-integration/src/main/java/org/springframework/batch/container/jms/BatchMessageListenerContainer.java diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/container/jms/BatchMessageListenerContainer.java b/spring-batch-integration/src/main/java/org/springframework/batch/container/jms/BatchMessageListenerContainer.java new file mode 100644 index 000000000..2ebafd916 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/batch/container/jms/BatchMessageListenerContainer.java @@ -0,0 +1,145 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.container.jms; + +import javax.jms.JMSException; +import javax.jms.MessageConsumer; +import javax.jms.Session; + +import org.aopalliance.aop.Advice; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.aop.support.DefaultPointcutAdvisor; +import org.springframework.aop.support.NameMatchMethodPointcut; +import org.springframework.batch.repeat.RepeatOperations; +import org.springframework.batch.repeat.interceptor.RepeatOperationsInterceptor; +import org.springframework.jms.connection.TransactionAwareConnectionFactoryProxy; +import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.transaction.interceptor.TransactionInterceptor; + +/** + * Message listener container adapted for intercepting the message reception + * with advice provided through configuration.
+ * + * To enable batching of messages in a single transaction, use the + * {@link TransactionInterceptor} and the {@link RepeatOperationsInterceptor} in + * the advice chain (with or without a transaction manager set in the base + * class). Instead of receiving a single message and processing it, the + * container will then use a {@link RepeatOperations} to receive multiple + * messages in the same thread. Use with a {@link RepeatOperations} and a + * transaction interceptor. If the transaction interceptor uses XA then use an + * XA connection factory, or else the + * {@link TransactionAwareConnectionFactoryProxy} to synchronize the JMS session + * with the ongoing transaction (opening up the possibility of duplicate + * messages after a failure). In the latter case you will not need to provide a + * transaction manager in the base class - it only gets on the way and prevents + * the JMS session from synchronizing with the database transaction. + * + * @author Dave Syer + * + */ +public class BatchMessageListenerContainer extends DefaultMessageListenerContainer { + + /** + * @author Dave Syer + * + */ + public static interface ContainerDelegate { + boolean receiveAndExecute(Session session, MessageConsumer consumer) throws JMSException; + } + + private Advice[] advices = new Advice[0]; + + private ContainerDelegate delegate = new ContainerDelegate() { + public boolean receiveAndExecute(Session session, MessageConsumer consumer) throws JMSException { + return BatchMessageListenerContainer.super.receiveAndExecute(session, consumer); + } + }; + + private ContainerDelegate proxy = delegate; + + /** + * Public setter for the {@link Advice}. + * @param advices the advice to set + */ + public void setAdvices(Advice[] advices) { + this.advices = advices; + } + + /** + * Set up interceptor with provided advice on the + * {@link #receiveAndExecute(Session, MessageConsumer)} method. + * + * @see org.springframework.jms.listener.AbstractJmsListeningContainer#afterPropertiesSet() + */ + public void afterPropertiesSet() { + super.afterPropertiesSet(); + initializeProxy(); + } + + /** + * Override base class to prevent exceptions from being swallowed. Should be + * an injectable strategy (see SPR-4733). + * + * @see org.springframework.jms.listener.AbstractMessageListenerContainer#handleListenerException(java.lang.Throwable) + */ + protected void handleListenerException(Throwable ex) { + if (!isSessionTransacted()) { + // Log the exceptions in base class if not transactional anyway + super.handleListenerException(ex); + return; + } + logger.debug("Re-throwing exception in container."); + if (ex instanceof RuntimeException) { + // We need to re-throw so that an enclosing non-JMS transaction can + // rollback... + throw (RuntimeException) ex; + } + else if (ex instanceof Error) { + // Just re-throw Error instances because otherwise unit tests just + // swallow exceptions from EasyMock and JUnit. + throw (Error) ex; + } + } + + /** + * Override base class method to wrap call in advice if provided. + * @see org.springframework.jms.listener.AbstractPollingMessageListenerContainer#receiveAndExecute(javax.jms.Session, + * javax.jms.MessageConsumer) + */ + protected boolean receiveAndExecute(final Session session, final MessageConsumer consumer) throws JMSException { + return proxy.receiveAndExecute(session, consumer); + } + + /** + * + */ + public void initializeProxy() { + ProxyFactory factory = new ProxyFactory(); + for (int i = 0; i < advices.length; i++) { + DefaultPointcutAdvisor advisor = new DefaultPointcutAdvisor(advices[i]); + NameMatchMethodPointcut pointcut = new NameMatchMethodPointcut(); + pointcut.addMethodName("receiveAndExecute"); + advisor.setPointcut(pointcut); + factory.addAdvisor(advisor); + } + factory.setProxyTargetClass(false); + factory.addInterface(ContainerDelegate.class); + factory.setTarget(delegate); + proxy = (ContainerDelegate) factory.getProxy(); + } + +}