publish-subscribe-channel tries to use QueueSession
Hi,
I configured some queues an one topic. For some time the messages proceed as expected, but after a while messages published via the topic fail, since a queue session is used:
Can some please help to check was happens here?
Here is my configuration (partially):
Code:
<?xml version="1.0" encoding="UTF-8"?>
<bean:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:bean="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:jms="http://www.springframework.org/schema/integration/jms"
xmlns:stream="http://www.springframework.org/schema/integration/stream"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms-2.0.xsd
http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd">
<bean:bean id="connectionFactory"
class="org.springframework.jms.connection.CachingConnectionFactory">
<bean:property name="targetConnectionFactory">
<bean:bean class="org.apache.activemq.ActiveMQConnectionFactory">
<bean:property name="brokerURL" value="${jms.brokerURL}" />
</bean:bean>
</bean:property>
<bean:property name="sessionCacheSize" value="${jms.sessionCacheSize}" />
<bean:property name="cacheProducers" value="${jms.cacheProducers}" />
<bean:property name="cacheConsumers" value="${jms.cacheConsumers}" />
</bean:bean>
<bean:bean id="requestQueue" class="org.apache.activemq.command.ActiveMQQueue">
<bean:constructor-arg value="${jms.queueName.request}" />
</bean:bean>
<bean:bean id="replyQueue" class="org.apache.activemq.command.ActiveMQQueue">
<bean:constructor-arg value="${jms.queueName.reply}" />
</bean:bean>
<bean:bean id="asyncRequestQueue" class="org.apache.activemq.command.ActiveMQQueue">
<bean:constructor-arg value="${jms.queueName.async}" />
</bean:bean>
<chain input-channel="requests" output-channel="jms.replies">
<transformer>
<bean:bean
class="org.springframework.integration.json.ObjectToJsonTransformer" />
</transformer>
<jms:outbound-gateway request-destination="requestQueue"
reply-destination="replyQueue" explicit-qos-enabled="true"
time-to-live="${jms.request.timeToLive}" receive-timeout="${jms.receive.timeout}" />
</chain>
<channel id="async.map.requests" />
<chain input-channel="async.requests" output-channel="async.map.requests">
<transformer>
<bean:bean
class="org.springframework.integration.json.ObjectToJsonTransformer" />
</transformer>
</chain>
<jms:outbound-channel-adapter id="jmsAsyncOut"
destination="asyncRequestQueue" channel="async.map.requests" />
<!-- activeMQ topic -->
<bean:bean id="cacheMgmtTopic" class="org.apache.activemq.command.ActiveMQTopic">
<bean:constructor-arg value="${jms.cacheMgmt.cacheMgmtTopic}" />
</bean:bean>
<!-- Channel to publish cache management events -->
<jms:publish-subscribe-channel id="toCacheMgmtTopic"
auto-startup="true" connection-factory="connectionFactory" topic="cacheMgmtTopic"
selector="FALSE" />
<!-- Channel to receive cache management events -->
<jms:message-driven-channel-adapter
channel="fromCacheMgmtTopic" destination="cacheMgmtTopic"
connection-factory="connectionFactory" extract-payload="true" />
</bean:beans>
And here is the (shortened) stacktrace:
Code:
Caused by: org.springframework.integration.MessageDeliveryException: failed to send Message to channel 'toCacheMgmtTopic'
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:165) ~[spring-integration-core-2.0.5.RELEASE.jar:na]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:128) ~[spring-integration-core-2.0.5.RELEASE.jar:na]
at com.gamigo.gas.general.service.CacheActionPublisher.notifySubscribersToPurgeCaches(CacheActionPublisher.java:29) ~[CacheActionPublisher.class:na]
at com.gamigo.gas.general.service.CacheActionPublisher$$FastClassByCGLIB$$2084119b.invoke(<generated>) ~[cglib-2.2.jar:na]
at net.sf.cglib.proxy.MethodProxy.invoke(MethodProxy.java:191) ~[cglib-2.2.jar:na]
at org.springframework.aop.framework.Cglib2AopProxy$CglibMethodInvocation.invokeJoinpoint(Cglib2AopProxy.java:688) ~[spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:150) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:166) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:89) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.framework.Cglib2AopProxy$DynamicAdvisedInterceptor.intercept(Cglib2AopProxy.java:621) ~[spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at com.gamigo.gas.general.service.CacheActionPublisher$$EnhancerByCGLIB$$fc0f42a.notifySubscribersToPurgeCaches(<generated>) ~[cglib-2.2.jar:na]
at com.gamigo.gas.general.service.SystemMgmtServiceImpl.clearAllCaches(SystemMgmtServiceImpl.java:53) ~[SystemMgmtServiceImpl.class:na]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.6.0_21]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39) ~[na:1.6.0_21]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) ~[na:1.6.0_21]
at java.lang.reflect.Method.invoke(Method.java:597) ~[na:1.6.0_21]
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:309) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:183) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:150) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:166) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.aspectj.MethodInvocationProceedingJoinPoint.proceed(MethodInvocationProceedingJoinPoint.java:80) ~[spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at com.gamigo.gas.general.ValidationAspect.invoke(ValidationAspect.java:30) ~[ValidationAspect.class:na]
at sun.reflect.GeneratedMethodAccessor174.invoke(Unknown Source) ~[na:na]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) ~[na:1.6.0_21]
at java.lang.reflect.Method.invoke(Method.java:597) ~[na:1.6.0_21]
at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethodWithGivenArgs(AbstractAspectJAdvice.java:621) ~[spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethod(AbstractAspectJAdvice.java:610) ~[spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.aspectj.AspectJAroundAdvice.invoke(AspectJAroundAdvice.java:65) ~[spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:161) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:166) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:166) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:89) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:202) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at $Proxy564.clearAllCaches(Unknown Source) ~[na:na]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.6.0_21]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39) ~[na:1.6.0_21]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) ~[na:1.6.0_21]
at java.lang.reflect.Method.invoke(Method.java:597) ~[na:1.6.0_21]
at org.apache.cxf.service.invoker.AbstractInvoker.performInvocation(AbstractInvoker.java:173) ~[cxf-rt-core-2.3.3.jar:2.3.3]
at org.apache.cxf.service.invoker.AbstractInvoker.invoke(AbstractInvoker.java:89) ~[cxf-rt-core-2.3.3.jar:2.3.3]
... 76 common frames omitted ...
