how to implement MessageListener to use sessions?

[lazzha]

Hi you all,

Ok now from the blackboard to the byte code, we are now in the implementation phase for this idea http://forum.springframework.org/showthread.php?t=18182, and this is the big picture:

…the main idea is that a client (ivr), talks to an Activation Server (composed by Spring + Pojos + Hibernate) and that the client sends every data that we need for the activation.

But, if for some reason, once we have all our data collected, the activation can’t be completed, because the databases are down, the ivr lost the connection with the activation server, or whatever other problems, we need to store all that data somewhere and wait until the server, databases or activation resources are available so we can complete the activation.

So summing up, we have a point to point jms; we have an asynchronous message producer, the activation server, queuing messages into one single queue, and an asynchronous message consumer, lets call this guy ActivationPojo, that would implement the MessageListener interface, kind of following the expertise/example of James Strachan from this blog http://jroller.com/page/habuma/20050...e_driven_pojos

Ok our doubts so far are:

1. When we send the ActivationPojo to the queue, it’s because something in our service layer failed, we are paying special attention to database transaction failures, and here’s where the first question arises, should we use a transaction to send the ActivationPojo to the queue, paraphrasing, should a Database Transaction be involved with a JMS Transaction?. We are getting a little bit messed up here because I think my team and I are confusing Database Transactions with Message Transactions, so we thing that it’s not necessary to use transactions to send the ActivationPojo to the queue, since it’s only one message, but just to be sure what do you think about it?

2. Ok, our ActivationPojo would try to make the activation (consume the message) until the activation is complete (no more errors from our service layer). This would be done by calling the onMessage() method. But here the second question arises, how could we re queue the message, if the consumption was not a successful consumption?

According to some readings I’ve made, this is done by invoking the rollback method from the session that’s holding the message, but in Spring, how could we do this?. It’s also related to what kind of acknowledgment mode we are using, how could we inject the session to our ActivationPojo, since it’s only implementing the MessageListener? should we just inject connection that we used to send the message?, but since it’s a single connection it opens and closes every time a message is sent, so I think the session would be lost!, would it not?


Ok any thoughts, ideas, or comments are welcome…

Cheers

[jstrachan]

1. here's a few links on JMS and transactions...

http://activemq.org/Should+I+use+transactions
http://activemq.org/Should+I+use+XA

2. Use a transacted session then you can commit() whenever the message is consumed properly without throwing an exception - then its automatically requeued by your JMS provider.

Even easier, just use Jencks to consume your JMS messages and it does all this for you...

http://jencks.org/Message+Driven+POJOs

[lazzha]

Hi James, well as you suggested in your reply we are trying to use Jencks, and we are following your Message Driven POJOs example, but our configuration is exactly the same as in your examples but we are getting this error:

Exception in thread "main" org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'holaMdp' defined in class path resource [com/telcel/activaciones/aplicacion/jms/applicationContext-mdp-jencks.xml]: Initialization of bean failed; nested exception is java.lang.IllegalArgumentException: resourceAdapter property must be set on the activationSpec object
java.lang.IllegalArgumentException: resourceAdapter property must be set on the activationSpec object
	at org.jencks.JCAConnector.afterPropertiesSet(JCAConnector.java:72)…

We think this is maybe a jar version problem, because when trying to make the cvs’s test cases run they doesn’t work either.

Because we also pulled out BootstrapContextFactoryBean and resourceAdapter beans, and put them as the error is requiring, but then we get that the BootstrapContextFactoryBean is not completely initialized…

Would you be so kind to tell us which versions you are using?, or do you have a complete working example that we can follow?

Also instead of the default transaction manager of Jenks should we use a spring transaction manager, in order for us to continue using our declarative databases transactions? We are a little messed up :cry:

Thanks in advance...

Ps(we wanted to ask you this on the server side forum but the forum is showing us many erros :wink: )

[lazzha]

Hi once again, well as we comment in the last post, we pooled out the BootstrapContextFactoryBean and resourceAdapter beans and ended with something like this:

 <?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN"
    "http&#58;//www.springframework.org/dtd/spring-beans.dtd">

<beans>

	<bean
		id="jencks"
		class="org.jencks.JCAContainer">
		<property
			name="bootstrapContext"
			ref="bootstrapContext" />
		<property
			name="resourceAdapter"
			ref="activeMQResourceAdapter" />
	</bean>

	<bean
		id="bootstrapContext"
		class="org.jencks.factory.BootstrapContextFactoryBean">
		<property
			name="threadPoolSize"
			value="25" />
	</bean>

	<bean
		id="activeMQResourceAdapter"
		class="org.activemq.ra.ActiveMQResourceAdapter">
		<property
			name="serverUrl"
			value="tcp&#58;//10.131.5.140&#58;61616" />
	</bean>

	<bean
		id="holaMdp"
		factory-method="addConnector"
		factory-bean="jencks">

		<property name="activationSpec">
			<bean class="org.activemq.ra.ActiveMQActivationSpec">
				<property
					name="destination"
					value="Hola.Queue" />
				<property
					name="destinationType"
					value="javax.jms.Queue" />
			</bean>
		</property>

		<property
			name="bootstrapContext"
			ref="bootstrapContext" />

		<property
			name="resourceAdapter"
			ref="activeMQResourceAdapter" />

		<property
			name="ref"
			value="holaBean" />
	</bean>

	<bean
		id="holaBean"
		class="com.telcel.activaciones.aplicacion.jms.HolaMundoBean"
		singleton="true" />

</beans>

and when running, WORKS but we get a DEBUG error, should it matter that is a debug error?:

 DEBUG: Ignoring BeanCreationException on FactoryBean type check
org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'holaMdp' defined in class path resource [com/telcel/activaciones/aplicacion/jms/applicationContext-mdp-jencks.xml]: Can't resolve reference to bean 'bootstrapContext' while setting property 'bootstrapContext'; nested exception is org.springframework.beans.factory.FactoryBeanNotInitializedException: Error creating bean with name 'bootstrapContext': FactoryBean returned null object: probably not fully initialized (maybe due to circular bean reference)
org.springframework.beans.factory.FactoryBeanNotInitializedException: Error creating bean with name 'bootstrapContext': FactoryBean returned null object: probably not fully initialized (maybe due to circular bean reference)
	at org.springframework.beans.factory.support.AbstractBeanFactory.getObjectForSharedInstance(AbstractBeanFactory.java:800)
	at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:191)
	at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:147)
	at org.springframework.beans.factory.support.BeanDefinitionValueResolver.resolveReference(BeanDefinitionValueResolver.java:176)
	at org.springframework.beans.factory.support.BeanDefinitionValueResolver.resolveValueIfNecessary(BeanDefinitionValueResolver.java:105)
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.applyPropertyValues(AbstractAutowireCapableBeanFactory.java:1012)
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.populateBean(AbstractAutowireCapableBeanFactory.java:823)
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:345)
	at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:226)
	at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:147)
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.getTypeForFactoryMethod(AbstractAutowireCapableBeanFactory.java:396)
	at org.springframework.beans.factory.support.AbstractBeanFactory.getType(AbstractBeanFactory.java:329)
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.isBeanTypeMatch(DefaultListableBeanFactory.java:249)
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.getBeanNamesForType(DefaultListableBeanFactory.java:144)
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.getBeansOfType(DefaultListableBeanFactory.java:198)
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.getBeansOfType(DefaultListableBeanFactory.java:192)
	at org.springframework.context.support.AbstractApplicationContext.getBeansOfType(AbstractApplicationContext.java:606)
	at org.jencks.factory.WorkManagerFactoryBean.getTransactionManager(WorkManagerFactoryBean.java:114)
	at org.jencks.factory.WorkManagerFactoryBean.createTransactionContextManager(WorkManagerFactoryBean.java:172)
	at org.jencks.factory.WorkManagerFactoryBean.getTransactionContextManager(WorkManagerFactoryBean.java:95)
	at org.jencks.factory.WorkManagerFactoryBean.createWorkManager(WorkManagerFactoryBean.java:176)
	at org.jencks.factory.WorkManagerFactoryBean.afterPropertiesSet(WorkManagerFactoryBean.java:74)
	at org.jencks.factory.WorkManagerFactoryBean.getWorkManager(WorkManagerFactoryBean.java:80)
	at org.jencks.factory.BootstrapContextFactoryBean.getWorkManager(BootstrapContextFactoryBean.java:76)
	at org.jencks.factory.BootstrapContextFactoryBean.afterPropertiesSet(BootstrapContextFactoryBean.java:63)
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.invokeInitMethods(AbstractAutowireCapableBeanFactory.java:1058)
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:363)
	at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:226)
	at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:147)
	at org.springframework.beans.factory.support.BeanDefinitionValueResolver.resolveReference(BeanDefinitionValueResolver.java:176)
	at org.springframework.beans.factory.support.BeanDefinitionValueResolver.resolveValueIfNecessary(BeanDefinitionValueResolver.java:105)
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.applyPropertyValues(AbstractAutowireCapableBeanFactory.java:1012)
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.populateBean(AbstractAutowireCapableBeanFactory.java:823)
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:345)
	at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:226)
	at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:147)
	at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:275)
	at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:318)
	at org.springframework.context.support.ClassPathXmlApplicationContext.<init>&#40;ClassPathXmlApplicationContext.java&#58;81&#41;
	at org.springframework.context.support.ClassPathXmlApplicationContext.<init>&#40;ClassPathXmlApplicationContext.java&#58;66&#41;
	at org.springframework.context.support.ClassPathXmlApplicationContext.<init>&#40;ClassPathXmlApplicationContext.java&#58;57&#41;
	at com.telcel.activaciones.aplicacion.jms.HolaMundoMdpApp.main&#40;HolaMundoMdpApp.java&#58;8&#41;

Afterwards we took a look at the Jencks source code, and saw that we could pass the BootstrapContextFactoryBean and resourceAdapter as parameters to the JCAConnector constructor, so we ended up with a configuration like this:

 <?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN"
    "http&#58;//www.springframework.org/dtd/spring-beans.dtd">

<beans>

	<bean
		id="jencks"
		class="org.jencks.JCAContainer">
		<property
			name="bootstrapContext"
			ref="bootstrapContext" />
		<property
			name="resourceAdapter"
			ref="activeMQResourceAdapter" />
	</bean>

	<bean
		id="bootstrapContext"
		class="org.jencks.factory.BootstrapContextFactoryBean">
		<property
			name="threadPoolSize"
			value="25" />
	</bean>

	<bean
		id="activeMQResourceAdapter"
		class="org.activemq.ra.ActiveMQResourceAdapter">
		<property
			name="serverUrl"
			value="tcp&#58;//10.131.5.140&#58;61616" />
	</bean>

	<bean
		id="holaMdp"
		class="org.jencks.JCAConnector">
		<constructor-arg ref="bootstrapContext" />
		<constructor-arg ref="activeMQResourceAdapter" />

		<property name="activationSpec">
			<bean class="org.activemq.ra.ActiveMQActivationSpec">
				<property
					name="destination"
					value="Hola.Queue" />
				<property
					name="destinationType"
					value="javax.jms.Queue" />
			</bean>
		</property>

		<property
			name="ref"
			value="holaBean" />
	</bean>

	<bean
		id="holaBean"
		class="com.telcel.activaciones.aplicacion.jms.HolaMundoBean"
		singleton="true" />

</beans>

And everything works and we get no error, the question here is, is it correctly configured or there was something we were missing in our first configuration in order for it to work?

If these configurations are ok, or at least one of them, the next step is to implement or not a transacted session, as we red in the http://activemq.org/Should+I+use+transactions, we are supposed to use a transacted session, since we are using Spring DataSourceTransactionMannager and if get an exception from the Service Layer injected to the Message Consumer (Message Driven POJOs) we need to re-queue the message.

Is this done in the same way with Jencks?, or is there any example we can follow, should we use a local Jms Transaction? Or any guidance?


Thanks in advance once again…

[lazzha]

Ok now apparently we manage to get to a configuration that uses transactions, here’s the snippet:

 <?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN"
        "http&#58;//www.springframework.org/dtd/spring-beans.dtd">

<beans>
<!-- WE ADDED THIS TRANSACTION MANAGER, AND USED IT IN THE  BOOTSTRAPCONTEXT AND IN OUR HOLAMDP BEAN &#40;CONSUMER&#41; -->
    <bean id="transactionManager"
          class="org.jencks.factory.TransactionManagerFactoryBean"/>


    <bean id="jencks" class="org.jencks.JCAContainer">
        <property name="bootstrapContext"
                  ref="bootstrapContext"/>
        <property name="resourceAdapter"
                  ref="activeMQResourceAdapter"/>
    </bean>

    <bean id="bootstrapContext" class="org.jencks.factory.BootstrapContextFactoryBean">
        <property name="transactionManager"
                  ref="transactionManager"/>
        <property name="threadPoolSize"
                  value="25"/>
    </bean>

    <bean id="activeMQResourceAdapter" class="org.activemq.ra.ActiveMQResourceAdapter">
        <property name="serverUrl"
                  value="tcp&#58;//10.131.5.140&#58;61616"/>
    </bean>

    <bean id="holaMdp"
          factory-method="addConnector"
          factory-bean="jencks">

        <property name="activationSpec">
            <bean class="org.activemq.ra.ActiveMQActivationSpec">
                <property name="destination"
                          value="Hola.Queue"/>
                <property name="destinationType"
                          value="javax.jms.Queue"/>
            </bean>
        </property>

        <property name="bootstrapContext"
                  ref="bootstrapContext"/>

        <property name="resourceAdapter"
                  ref="activeMQResourceAdapter"/>

        <property name="transactionManager"
                  ref="transactionManager"/>

        <property name="ref"
                  value="holaBean"/>
    </bean>

    <bean id="holaBean"
          class="com.telcel.activaciones.aplicacion.jms.HolaMundoBean"
          singleton="true"/>

</beans>

If we are in debug mode still get the same Debug Exception we got in the previous post, still wondering if that matters or no…

Ok supposedly we have a transactions and it re-queues the messages if any exception is thrown, if we catch it the message is consumed “correctly”, here is where our questions arise, first, the commit process is done automatically if no exceptions are thrown?, or re-queued if any exception happens?. Trying to make some emphasis in the

Even easier, just use Jencks to consume your JMS messages and it does all this for you...

, this means we don’t have to make any “manual” or explicit commit to say the message was consumed correctly?

Ok the second question; we obtain the same behavior, automatic re-queued of messages if exceptions are thrown, if we remove the transactionManager bean, so what’s the difference if we use the transactionManager or not? According to documentations this is going to help us once we use our Service Layer to access databases etc… is that so?

Once again thanks…

Cheers

[jstrachan]

Firstly I'm so glad you got things working all by yourself

I don't yet completely understand when you do and don't get the Spring circularity error; we've never seen that yet - I guess we need to be a little careful over the order in which beans are declared and configured. I wonder if we can change Jencks in some way to make it harder to make the same mistake?

In terms of transactions and so forth...

If you don't specify a JTA transaction manager on a connector, then Jencks will use local JMS transactions. If you specify a transaction manager it will use XA transactions.

In either mode - throwing a runtime exception in the MessageListener will cause a rollback and for the message to be re-dispatched onto the queue.

The main difference between local and XA is that local is a transaction purely on the JMS session; it does not include any other resource. XA however can include multiple resources (e.g. your database). Though be aware XA is much much slower due to the persistence steps required to recover.

http://activemq.org/Should+I+use+XA
http://activemq.org/Should+I+use+transactions

In actual fact, most decent J2EE containers and Transaction Managers, if using XA and only having one resource, will often switch into local transaction mode, so you should only pay for XA if you actually use it - though its definitely worth testing that to be the case before assuming it is.

[lazzha]

Hi James, hi to everybody…

Well now we have our jms producer all set up and we would like to have some kind of daemon to consume the messages ASAP.
What we have thought of, is to put the consumer as a bean in the activeMQ start configuration… sounded like a good idea, as soon as the server starts up the consumer also starts up and waits for messages… but we try that and nothing’s happening, it looks like it’s not initializing the spring beans…

This is the configuration I have…

 <?xml version="1.0" encoding="iso-8859-1"?>
<!DOCTYPE beans PUBLIC  "-//ACTIVEMQ//DTD//EN" "http&#58;//activemq.org/dtd/activemq.dtd">
<beans>

    <!-- ==================================================================== -->
    <!-- ActiveMQ Broker Configuration -->
    <!-- ==================================================================== -->
    <broker>
        <connector>
            <tcpServerTransport uri="tcp&#58;//10.131.5.60&#58;61616" backlog="1000" useAsyncSend="true"
                                maxOutstandingMessages="50"/>
        </connector>


        <persistence>
            <cachePersistence>
                <journalPersistence directory="../var/journal">

                    <jdbcPersistence dataSourceRef="mysql-ds"
                                     adapterClass="org.activemq.store.jdbc.adapter.DefaultJDBCAdapter"/>

                </journalPersistence>
            </cachePersistence>
        </persistence>

        <redeliveryPolicy backOffMode="true"
                          backOffIncreaseRate="2"
                          initialRedeliveryTimeout="1000"
                          maximumRetryCount="1073741823"/>


    </broker>

    <!-- ==================================================================== -->
    <!-- JDBC DataSource Configurations -->
    <!-- ==================================================================== -->

    <!-- The MYSQL Datasource that will be used by the Broker -->
    <bean id="mysql-ds"
          class="org.apache.commons.dbcp.BasicDataSource"
          destroy-method="close">
        <property name="driverClassName">
            <value>com.mysql.jdbc.Driver</value>
        </property>
        <property name="url">
            <value>jdbc&#58;mysql&#58;//10.131.1.13/activemq</value>
        </property>
        <property name="username">
            <value>activemq</value>
        </property>
        <property name="password">
            <value>activemq</value>
        </property>
        <property name="poolPreparedStatements">
            <value>true</value>
        </property>
    </bean>

    <!-- ==================================================================== -->
    <!-- Jms Consumer Bean -->
    <!-- ==================================================================== -->

    <bean id="jencks" class="org.jencks.JCAContainer">
        <property name="bootstrapContext">
            <ref local="bootstrapContext"/>
        </property>
        <property name="resourceAdapter">
            <ref local="activeMQResourceAdapter"/>
        </property>

    </bean>

    <bean id="bootstrapContext"
          class="org.jencks.factory.BootstrapContextFactoryBean">
        <property name="threadPoolSize">
            <value>25</value>
        </property>
    </bean>

    <bean id="activeMQResourceAdapter"
          class="org.activemq.ra.ActiveMQResourceAdapter">
        <property
                name="serverUrl">
            <value>tcp&#58;//localhost&#58;61616</value>
        </property>
    </bean>

    <bean id="MdpConsumer"
          class="org.jencks.JCAConnector">


        <constructor-arg>
            <ref local="bootstrapContext"/>
        </constructor-arg>
        <constructor-arg>
            <ref local="activeMQResourceAdapter"/>
        </constructor-arg>

        <property name="activationSpec">
            <bean class="org.activemq.ra.ActiveMQActivationSpec">
                <property name="destination">
                    <value>Test.Queue</value>
                </property>

                <property name="destinationType">
                    <value>javax.jms.Queue</value>
                </property>
            </bean>
        </property>

        <property name="ref">
            <value>ConsumerBean</value>
        </property>
    </bean>

    <bean id="ConsumerBean"
          class="com.telcel.activaciones.aplicacion.jms.JmsConsumer"
          singleton="true"/>

</beans>

Do you, or anybody have any idea why the beans are not being initialized?…

and we have another question, what happens if the same consumer is loaded by accident two or more times, does the server makes a load balance in order to send each consumer a DIFFERENT message or it sends the SAME message to all of them?

Appreciate all your help…

cheers

[jstrachan]

I think that could be ActiveMQ 3.x's issue in that its using a BeanFactory not an ApplicationContext to parse the XML; so its not auto-creating any singleton beans.

An alternative is to load your own ApplicationContext with a normal Spring.xml, configure all your beans on startup - but then use the factory bean to configure ActiveMQ's broker...

http://activemq.org/Spring+Support

i.e. add this to your spring.xml

<bean id="broker" class="org.activemq.spring.BrokerFactoryBean">
<property name="config" value="classpath:foo/bar/broker.xml"/>
</bean>

note this situation is all solved nicely in ActiveMQ 4.x But until then I'd use the above and you should be fine.

Note that this is just a glitch in how the XML is parsed in 3.x; you can still easily have all this stuff in the same JVM and to start up together; its just easier if you configure ActiveMQ 3.x in a separate XML document.

[singh_darshan]

Still i am not able to do this. Can you please give me the client code which uses this app context xml to recieve messages.

Thanks
Darshan