Hello,
I'm trying to test the queue persistence of ActiveMQ.
I have an embedded ActiveMQ server with an unique consumer.
This embedded server receive JMS messages from many other JVM applications.
It works fine, consumer application receive the notifications.
So i've tried to test the persistence of the messages.
I've put a (remote) breakpoint on the MessageListener of the consumer so
that i can enqueue many messages and make the ActiveMQ server crash.
On server restart, i'd like all the enqueued messages to be able to be
consumed, and not to be lost.
And then i tried that test.
I got into that breakpoint on the first message send.
But for all messages i try to send, i get the following stacktrack on the
producer side:
Exception in thread "main"
org.springframework.jms.UncategorizedJmsException: Uncategorized exception
occured during JMS processing; nested exception is javax.jms.JMSException:
Wire format negotiation timeout: peer did not send his wire format.
at
org.springframework.jms.support.JmsUtils.convertJmsAccessException(JmsUtils.java:316)
at
org.springframework.jms.support.JmsAccessor.convertJmsAccessException(JmsAccessor.java:168)
at org.springframework.jms.core.JmsTemplate.execute(JmsTemplate.java:469)
at org.springframework.jms.core.JmsTemplate.send(JmsTemplate.java:534)
at
org.springframework.jms.core.JmsTemplate.convertAndSend(JmsTemplate.java:612)
at
org.springframework.jms.core.JmsTemplate.convertAndSend(JmsTemplate.java:604)
at
com.xxxxxxxxxxx.mobilepush.client.RealClientTest.main(RealClientTest.java:29)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
at java.lang.reflect.Method.invoke(Method.java:597)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:120)
Caused by: javax.jms.JMSException: Wire format negotiation timeout: peer
did not send his wire format.
at
org.apache.activemq.util.JMSExceptionSupport.create(JMSExceptionSupport.java:62)
at
org.apache.activemq.ActiveMQConnection.syncSendPacket(ActiveMQConnection.java:1380)
at
org.apache.activemq.ActiveMQConnection.ensureConnectionInfoSent(ActiveMQConnection.java:1466)
at
org.apache.activemq.ActiveMQConnection.createSession(ActiveMQConnection.java:308)
at
org.springframework.jms.support.JmsAccessor.createSession(JmsAccessor.java:196)
at org.springframework.jms.core.JmsTemplate.execute(JmsTemplate.java:457)
... 9 more
Caused by: java.io.IOException: Wire format negotiation timeout: peer did
not send his wire format.
at
org.apache.activemq.transport.WireFormatNegotiator.oneway(WireFormatNegotiator.java:98)
at
org.apache.activemq.transport.MutexTransport.oneway(MutexTransport.java:68)
at
org.apache.activemq.transport.ResponseCorrelator.asyncRequest(ResponseCorrelator.java:81)
at
org.apache.activemq.transport.ResponseCorrelator.request(ResponseCorrelator.java:86)
at
org.apache.activemq.ActiveMQConnection.syncSendPacket(ActiveMQConnection.java:1351)
... 13 more
I don't understand why my producer would be blocked when my consumer is in
my breakpoint.
My broker uri is: mobilepush.activemq.broker.transport.connector.uri=tcp://
0.0.0.0:61616
-------------------------------------------------------------------------------------------------------------------------------------------------
Any idea?
My configuration is pretty simple:
SERVER:
<!-- lets create an embedded ActiveMQ Broker -->
<amq:broker useJmx="false" persistent="true">
<amq:transportConnectors>
<amq:transportConnector
uri="${mobilepush.activemq.broker.transport.connector.uri}" />
</amq:transportConnectors>
<amq:persistenceAdapter>
<amq:kahaPersistenceAdapter
directory="${mobilepush.activemq.broker.queue.persistence.directory}"
maxDataFileLength="100 Mb"/>
</amq:persistenceAdapter>
</amq:broker>
CONSUMER:
<bean id="connectionFactory"
class="fr.xebia.management.jms.SpringManagedConnectionFactory">
<property name="connectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory" >
<property name="brokerURL"
value="${mobilepush.activemq.broker.uri}"/>
</bean>
</property>
</bean>
<bean id="pushConsumer"
class="com.xxxxxxxxxxxxxxx.mobilepush.messaging.jms.PushConsumer">
<property name="jmsPushMessageConverter"
ref="jmsPushMessageConverter"/>
<property name="pushDelegate" ref="directPushDelegate"/>
</bean>
<management:executor-service id="pushConsumerExecutor"
pool-size="${mobilepush.consumer.thread.min}-${mobilepush.consumer.thread.max}"
keep-alive="60" />
<jms:listener-container
task-executor="pushConsumerExecutor"
connection-factory="connectionFactory"
acknowledge="auto"
container-class="fr.xebia.springframework.jms.ManagedDefaultMessageListenerContainer">
<jms:listener destination="mobilepush.queue" ref="pushConsumer"
method="onMessage" />
</jms:listener-container>
(management namespace and xebia class it only a JMX decorator)
PRODUCER:
<bean id="connectionFactory"
class="org.apache.activemq.ActiveMQConnectionFactory" >
<property name="brokerURL"
value="${mobilepush.activemq.broker.uri}"/>
</bean>
<bean id="mobilePushJmsTemplate"
class="org.springframework.jms.core.JmsTemplate">
<property name="defaultDestination" ref="mobilePushQueue"/>
<property name="messageConverter" ref="jmsPushMessageConverter"/>
<property name="connectionFactory">
<!-- lets wrap in a pool to avoid creating a connection per
send -->
<bean
class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory">
<ref local="connectionFactory" />
</property>
</bean>
</property>
</bean>
Thanks
