Real-Time Messaging: General Question

[axle123]

I'm evaluating Real-Time Messaging in an application in which the user initiates some request - runFooCheck() - which is an RPC call that does some processing on the server which could take some indefinite amount of time, at the end of which a message is pushed to the client using the ISCMessageDispatcher.

I'm currently configured to use "Enterprise messaging" mode, using ActiveMQ and Tomcat following the instructions here: https://isomorphic.atlassian.net/wik...t+and+ActiveMQ

I want to know if it's possible for a client to receive the server-pushed message without being currently subscribed at the time the message was pushed. That is, a user clicks a button which initiates the runFooCheck(), closes their browser window for some inordinate period of time, meanwhile runFooCheck() completes and the ISCMessageDispatcher has pushed the resulting message onto the channel/Topic. When the user re-opens their browser window, they are re-subscribed to the channel, and they immediately receive the message that was there waiting for them.

Is this possible to achieve with Real-Time Messaging? Or is this a pipe dream?

[Isomorphic]

This is straightforward to achieve, but Real-Time Messaging is only a part of how you do it.

Basically, if the user is currently subscribed via Real-Time Messaging, you go ahead and send the message via Messaging. If they aren't, you store it in a database table (or similar permanent storage).

Then when the user comes back to the application, you retrieve any messages that were stored to the table, and deliver them via Messaging or just via a normal DataSource fetch.

[axle123]

I know that ActiveMQ in itself supports something they call Subscription Recovery, and Retroactive Consumer: https://activemq.apache.org/subscrip...ry-policy.html

Is this not possible to achieve with Real-Time Messaging without having to store data externally (e.g. database table)?

If not, it seems that using ActiveMQ by itself would better support this use case, rather than using Real-Time Messaging with a JMS/ActiveMQ dispatcher. Would you agree or disagree?

To reiterate, the use case here is that a client/user makes an asynchronous RPC request to the server, which runs a process that could take up to, say, 30 minutes, during which the client may close their browser window, go home for the day, etc.. When the server-side procedure is finished, the String result is pushed back to the client, and they receive this message whenever they are using the webapp -- whether it's at the same moment the message is pushed, or a day later when they re-navigate to the webapp.

[Isomorphic]

From reading that page, it sounds as if this feature is client-agnostic, so it would work with SmartGWT's Real-time Messaging.

As far as using other clients for ActiveMQ, we're not familiar with the quality level there, but we do know that our own messaging solution includes a lot of workarounds for browser bugs like memory leaks, infinite throbber display, issues with web proxies and the like.

It also looks like you would have to wrap ActiveMQ's Ajax client in GWT yourself or use a third-party solution (gwt-activemq).

And you would also lose current and future built-in SmartGWT APIs that take advantage of our own Messaging module.

[axle123]

Fair enough. Well, my follow up question would be how to configure the subscription recovery policy in this case, using Real-Time Messaging with ActiveMQ and Tomcat. I don't think there are settings that I can put in server.properties that would define the policy. Is this something I'd configure in Tomcat's context.xml, in the <Resource> definition for that specific topic?

[Isomorphic]

Definitely not server.properties, we do not ship ActiveMQ or attempt to auto-configure it.

ActiveMQ settings should be configured via ActiveMQ-specific properties files. We're not sure whether it can be done in Tomcat's context.xml, you should refer to Tomcat and/or ActiveMQ documentation for this.

[axle123]

It looks like my subscription recovery policy was already defined in my ActiveMQ's activemq.xml, but it doesn't seem to be doing what I'm expecting. I'm expecting to receive the missed message upon re-subscribing.

This led me to look into using a Queue instead of a Topic. According to ActiveMQ documentation, a Queue would be exactly what I want here:

"A single message will be received by exactly one consumer. If there are no consumers available at the time the message is sent it will be kept until a consumer is available that can process the message."

Taken from https://activemq.apache.org/how-does...o-a-topic.html

But it doesn't seem that Queues are supported with Real-Time Messaging, correct? I get a ClassCastException if I try to define my <Resource> as an org.apache.activemq.command.ActiveMQQueue

=== 2014-03-20 12:33:03,336 [c-10] ERROR MessagingServlet - MessagingServlet top-level exception
java.lang.ClassCastException
	at com.sun.corba.se.impl.javax.rmi.PortableRemoteObject.narrow(PortableRemoteObject.java:245)
	at javax.rmi.PortableRemoteObject.narrow(PortableRemoteObject.java:150)
	at com.isomorphic.messaging.JMSMessageDispatcher.bindToChannel(JMSMessageDispatcher.java:110)
	at com.isomorphic.messaging.JMSMessageDispatcher.reconnect(JMSMessageDispatcher.java:92)
	at com.isomorphic.messaging.JMSMessageDispatcher.ensureConnected(JMSMessageDispatcher.java:98)
	at com.isomorphic.messaging.JMSMessageDispatcher.subscribe(JMSMessageDispatcher.java:150)
	at com.isomorphic.messaging.ISCMessageDispatcher.subscribe(ISCMessageDispatcher.java:181)
	at com.isomorphic.messaging.MessagingConnectionHandler.registerSubscriber(MessagingConnectionHandler.java:341)
	at com.isomorphic.messaging.MessagingConnectionHandler.process(MessagingConnectionHandler.java:107)
	at com.isomorphic.messaging.MessagingServlet.doPost(MessagingServlet.java:107)
	at com.isomorphic.messaging.MessagingServlet.doGet(MessagingServlet.java:119)
	at javax.servlet.http.HttpServlet.service(HttpServlet.java:621)
	at com.isomorphic.servlet.BaseServlet.service(BaseServlet.java:152)
	at javax.servlet.http.HttpServlet.service(HttpServlet.java:728)
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:305)
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:210)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:330)
	at org.springframework.security.web.access.intercept.FilterSecurityInterceptor.invoke(FilterSecurityInterceptor.java:118)
	at org.springframework.security.web.access.intercept.FilterSecurityInterceptor.doFilter(FilterSecurityInterceptor.java:84)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342)
	at org.springframework.security.web.access.ExceptionTranslationFilter.doFilter(ExceptionTranslationFilter.java:113)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342)
	at org.springframework.security.web.session.SessionManagementFilter.doFilter(SessionManagementFilter.java:103)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342)
	at org.springframework.security.web.authentication.AnonymousAuthenticationFilter.doFilter(AnonymousAuthenticationFilter.java:113)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342)
	at org.springframework.security.web.servletapi.SecurityContextHolderAwareRequestFilter.doFilter(SecurityContextHolderAwareRequestFilter.java:54)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342)
	at org.springframework.security.web.savedrequest.RequestCacheAwareFilter.doFilter(RequestCacheAwareFilter.java:45)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342)
	at org.springframework.security.web.authentication.www.BasicAuthenticationFilter.doFilter(BasicAuthenticationFilter.java:201)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342)
	at org.springframework.security.web.authentication.preauth.AbstractPreAuthenticatedProcessingFilter.doFilter(AbstractPreAuthenticatedProcessingFilter.java:88)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342)
	at org.springframework.security.web.context.SecurityContextPersistenceFilter.doFilter(SecurityContextPersistenceFilter.java:87)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:342)
	at org.springframework.security.web.FilterChainProxy.doFilterInternal(FilterChainProxy.java:192)
	at org.springframework.security.web.FilterChainProxy.doFilter(FilterChainProxy.java:160)
	at org.springframework.web.filter.DelegatingFilterProxy.invokeDelegate(DelegatingFilterProxy.java:346)
	at org.springframework.web.filter.DelegatingFilterProxy.doFilter(DelegatingFilterProxy.java:259)
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:243)
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:210)
	at org.jasig.cas.client.session.SingleSignOutFilter.doFilter(SingleSignOutFilter.java:110)
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:243)
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:210)
	at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:222)
	at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:123)
	at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:472)
	at org.apache.catalina.valves.CometConnectionManagerValve.invoke(CometConnectionManagerValve.java:186)
	at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:171)
	at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:99)
	at org.apache.catalina.valves.AccessLogValve.invoke(AccessLogValve.java:936)
	at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:118)
	at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:407)
	at org.apache.coyote.http11.AbstractHttp11Processor.process(AbstractHttp11Processor.java:1004)
	at org.apache.coyote.AbstractProtocol$AbstractConnectionHandler.process(AbstractProtocol.java:589)
	at org.apache.tomcat.util.net.JIoEndpoint$SocketProcessor.run(JIoEndpoint.java:310)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
	at java.lang.Thread.run(Thread.java:744)
Caused by: java.lang.ClassCastException: org.apache.activemq.command.ActiveMQQueue cannot be cast to org.omg.CORBA.Object
	at com.sun.corba.se.impl.javax.rmi.PortableRemoteObject.narrow(PortableRemoteObject.java:228)
	... 59 more

[Isomorphic]

We don't currently support JMS Queues. It looks like a relatively straightforward enhancement that you could sponsor, although it doesn't seem particularly difficult to build your own implementation of persistent for messages (again using a database table).