merge bug fix VYZ-3431-6686 JMSOutputAdapter ClassCastException when …

…HA enabled in SSR
espertechinc · Mar 8, 2018 · b98e8c5 · b98e8c5
1 parent 1620dca
commit b98e8c5
Show file tree

Hide file tree

Showing 3 changed files with 35 additions and 4 deletions.
diff --git a/esperio-springjms/etc/regression/jms_regression_output_spring.xml b/esperio-springjms/etc/regression/jms_regression_output_spring.xml
@@ -52,10 +52,12 @@
           <key><idref local="subscriptionOne"/></key>
           <ref bean="subscriptionOne"/>
         </entry>
+        <!-- Example with a custom marshaller 
         <entry>
           <key><idref local="subscriptionTwo"/></key>
           <ref bean="subscriptionTwo"/>
         </entry>
+        -->
       </map>
     </property>
     <property name="jmsMessageMarshaller">
@@ -67,12 +69,14 @@
     <property name="eventTypeName" value="MyOutputStream"/>
   </bean>
 
+  <!-- Example with a custom marshaller 
   <bean id="subscriptionTwo" class="com.espertech.esperio.jms.JMSSubscription">
     <property name="eventTypeName" value="MyMapEventTwo"/>
     <property name="jmsMessageMarshaller">
       <ref bean="myCustomMarshaller"/>
     </property>
   </bean>
+  -->
 
 </beans>
 
diff --git a/esperio-springjms/src/main/java/com/espertech/esperio/jms/JMSOutputAdapter.java b/esperio-springjms/src/main/java/com/espertech/esperio/jms/JMSOutputAdapter.java
@@ -11,9 +11,7 @@
 package com.espertech.esperio.jms;
 
 import com.espertech.esper.adapter.*;
-import com.espertech.esper.client.EPException;
-import com.espertech.esper.client.EPServiceProvider;
-import com.espertech.esper.client.EventBean;
+import com.espertech.esper.client.*;
 import com.espertech.esper.core.service.EPServiceProviderSPI;
 import com.espertech.esper.util.ExecutionPathDebugLog;
 import org.slf4j.Logger;
@@ -132,7 +130,32 @@ public void start() throws EPException {
         stateManager.start();
         Iterator<Map.Entry<String, Subscription>> it = subscriptionMap.entrySet().iterator();
         while (it.hasNext()) {
-            it.next().getValue().registerAdapter(this);
+            Map.Entry<String, Subscription> sub = it.next();
+            JMSSubscription destination = (JMSSubscription) sub.getValue();
+            destination.setJMSOutputAdapter(this);
+
+            String statementName = this.getClass().getSimpleName() + "-" + sub.getKey() + "-" + sub.getValue().getEventTypeName();
+
+            EPStatement statement = spi.getEPAdministrator().getStatement(statementName);
+            if (statement == null) {
+                try {
+                    statement = spi.getEPAdministrator().createEPL("select * from " + sub.getValue().getEventTypeName(), statementName);
+                } catch (Throwable t) {
+                    String message = "Exception starting adapter: " + t.getMessage();
+                    log.error(message, t);
+                    throw new RuntimeException(message, t);
+                }
+            }
+            statement.addListener(new UpdateListener() {
+                public void update(EventBean[] newEvents, EventBean[] oldEvents) {
+                    if (newEvents == null) {
+                        return;
+                    }
+                    for (EventBean event : newEvents) {
+                        destination.matchFound(event, null);
+                    }
+                }
+            });
         }
     }
 

diff --git a/esperio-springjms/src/main/java/com/espertech/esperio/jms/JMSSubscription.java b/esperio-springjms/src/main/java/com/espertech/esperio/jms/JMSSubscription.java
@@ -46,6 +46,10 @@ public void setJmsMessageMarshaller(JMSMessageMarshaller jmsMessageMarshaller) {
         this.jmsMessageMarshaller = jmsMessageMarshaller;
     }
 
+    public void setJMSOutputAdapter(JMSOutputAdapter adapter) {
+        this.adapter = adapter;
+    }
+
     public void matchFound(EventBean theEvent, Collection<FilterHandleCallback> allStmtMatches) {
 
         if (!(adapter instanceof JMSOutputAdapter)) {