You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vk...@apache.org on 2018/04/21 18:23:27 UTC
ignite git commit: IGNITE-8154 - Add an ability to provide
ExceptionListener to JmsStreamer - Fixes #3828
Repository: ignite
Updated Branches:
refs/heads/master 3f3239b90 -> ff7c4888f
IGNITE-8154 - Add an ability to provide ExceptionListener to JmsStreamer - Fixes #3828
Signed-off-by: Valentin Kulichenko <va...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/ff7c4888
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/ff7c4888
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/ff7c4888
Branch: refs/heads/master
Commit: ff7c4888f7d24c0d90fb77f3da6ecbbbe06039e4
Parents: 3f3239b
Author: Anton Kurbanov <an...@gmail.com>
Authored: Sat Apr 21 11:23:21 2018 -0700
Committer: Valentin Kulichenko <va...@gmail.com>
Committed: Sat Apr 21 11:23:21 2018 -0700
----------------------------------------------------------------------
.../apache/ignite/stream/jms11/JmsStreamer.java | 28 +++++++
.../stream/jms11/IgniteJmsStreamerTest.java | 81 ++++++++++++++++++--
2 files changed, 103 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/ff7c4888/modules/jms11/src/main/java/org/apache/ignite/stream/jms11/JmsStreamer.java
----------------------------------------------------------------------
diff --git a/modules/jms11/src/main/java/org/apache/ignite/stream/jms11/JmsStreamer.java b/modules/jms11/src/main/java/org/apache/ignite/stream/jms11/JmsStreamer.java
index 1b203ad..55441d7 100644
--- a/modules/jms11/src/main/java/org/apache/ignite/stream/jms11/JmsStreamer.java
+++ b/modules/jms11/src/main/java/org/apache/ignite/stream/jms11/JmsStreamer.java
@@ -29,6 +29,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
+import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
@@ -40,6 +41,7 @@ import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.util.typedef.internal.A;
+import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.stream.StreamAdapter;
/**
@@ -150,6 +152,9 @@ public class JmsStreamer<T extends Message, K, V> extends StreamAdapter<T, K, V>
/** Message listeners. */
private Set<IgniteJmsMessageListener> listeners = Collections.newSetFromMap(new ConcurrentHashMap<IgniteJmsMessageListener, Boolean>());
+ /** The Jms {@link ExceptionListener} to use. */
+ private ExceptionListener exceptionListener;
+
/** Scheduler for handling {@link #batchClosureMillis}. */
private ScheduledExecutorService scheduler;
@@ -215,6 +220,8 @@ public class JmsStreamer<T extends Message, K, V> extends StreamAdapter<T, K, V>
connection.setClientID(clientId.trim());
}
+ connection.setExceptionListener(new IgniteJmsExceptionListener());
+
// build the JMS objects
if (destinationType == Queue.class) {
initializeJmsObjectsForQueue();
@@ -424,6 +431,15 @@ public class JmsStreamer<T extends Message, K, V> extends StreamAdapter<T, K, V>
this.durableSubscriptionName = durableSubscriptionName;
}
+ /**
+ * Exception listener for queue/topic failures.
+ *
+ * @param exceptionListener ExceptionListener interface implementation.
+ */
+ public void setExceptionListener(ExceptionListener exceptionListener) {
+ this.exceptionListener = exceptionListener;
+ }
+
private void initializeJmsObjectsForTopic() throws JMSException {
Session session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
Topic topic = (Topic)destination;
@@ -536,4 +552,16 @@ public class JmsStreamer<T extends Message, K, V> extends StreamAdapter<T, K, V>
}
}
+ /**
+ * Exception listener for JmsExceptions.
+ */
+ private class IgniteJmsExceptionListener implements ExceptionListener {
+ /** {@inheritDoc} */
+ @Override public void onException(JMSException e) {
+ U.error(log, "Caught JMS internal exception.", e);
+
+ if (exceptionListener != null)
+ exceptionListener.onException(e);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ff7c4888/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/IgniteJmsStreamerTest.java
----------------------------------------------------------------------
diff --git a/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/IgniteJmsStreamerTest.java b/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/IgniteJmsStreamerTest.java
index 2011684..caf8f9e 100644
--- a/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/IgniteJmsStreamerTest.java
+++ b/modules/jms11/src/test/java/org/apache/ignite/stream/jms11/IgniteJmsStreamerTest.java
@@ -17,14 +17,19 @@
package org.apache.ignite.stream.jms11;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
+import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
@@ -34,6 +39,7 @@ import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerPlugin;
import org.apache.activemq.broker.BrokerRegistry;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.region.DestinationStatistics;
@@ -42,11 +48,14 @@ import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.security.SimpleAuthenticationPlugin;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.IgniteException;
import org.apache.ignite.events.CacheEvent;
import org.apache.ignite.lang.IgniteBiPredicate;
+import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.After;
import org.junit.Before;
@@ -71,17 +80,17 @@ public class IgniteJmsStreamerTest extends GridCommonAbstractTest {
/** */
private static final Map<String, String> TEST_DATA = new HashMap<>();
+ static {
+ for (int i = 1; i <= CACHE_ENTRY_COUNT; i++)
+ TEST_DATA.put(Integer.toString(i), "v" + i);
+ }
+
/** */
private BrokerService broker;
/** */
private ConnectionFactory connFactory;
- static {
- for (int i = 1; i <= CACHE_ENTRY_COUNT; i++)
- TEST_DATA.put(Integer.toString(i), "v" + i);
- }
-
/** Constructor. */
public IgniteJmsStreamerTest() {
super(true);
@@ -116,7 +125,6 @@ public class IgniteJmsStreamerTest extends GridCommonAbstractTest {
}
/**
- *
* @throws Exception Iff ailed.
*/
@After
@@ -500,6 +508,67 @@ public class IgniteJmsStreamerTest extends GridCommonAbstractTest {
}
/**
+ * Test for ExceptionListener functionality.
+ *
+ * @throws Exception If fails.
+ */
+ public void testExceptionListener() throws Exception {
+ // restart broker with auth plugin
+ if (broker.isStarted())
+ broker.stop();
+
+ broker.waitUntilStopped();
+
+ broker.setPlugins(new BrokerPlugin[]{new SimpleAuthenticationPlugin(new ArrayList())});
+
+ broker.start(true);
+
+ connFactory = new ActiveMQConnectionFactory(BrokerRegistry.getInstance().findFirst().getVmConnectorURI());
+
+ final List<Throwable> lsnrExceptions = new LinkedList<>();
+
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ Destination dest = new ActiveMQQueue(QUEUE_NAME);
+
+ try (IgniteDataStreamer<String, String> dataStreamer = grid().dataStreamer(DEFAULT_CACHE_NAME)) {
+ JmsStreamer<ObjectMessage, String, String> jmsStreamer = newJmsStreamer(ObjectMessage.class, dataStreamer);
+
+ jmsStreamer.setExceptionListener(new ExceptionListener() {
+ @Override public void onException(JMSException e) {
+ System.out.println("ERROR");
+
+ lsnrExceptions.add(e);
+
+ latch.countDown();
+ }
+ });
+
+ jmsStreamer.setDestination(dest);
+
+ GridTestUtils.assertThrowsWithCause(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ jmsStreamer.start();
+
+ return null;
+ }
+ }, SecurityException.class);
+
+ assertTrue(latch.await(10, TimeUnit.SECONDS));
+
+ assertTrue(lsnrExceptions.size() > 0);
+
+ GridTestUtils.assertThrowsWithCause(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ jmsStreamer.stop();
+
+ return null;
+ }
+ }, IgniteException.class);
+ }
+ }
+
+ /**
*
*/
private void assertAllCacheEntriesLoaded() {