You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cxf.apache.org by re...@apache.org on 2019/12/11 01:03:32 UTC
[cxf] branch 3.2.x-fixes updated (e3c9486 -> a1bd091)
This is an automated email from the ASF dual-hosted git repository.
reta pushed a change to branch 3.2.x-fixes
in repository https://gitbox.apache.org/repos/asf/cxf.git.
from e3c9486 Recording .gitmergeinfo Changes
new bd98adf CXF-8161: fix memory leak and thread leak in JMSDestination. Adding more tests for JMSDestination first
new a1bd091 Recording .gitmergeinfo Changes
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.gitmergeinfo | 1 +
.../cxf/transport/jms/AbstractJMSTester.java | 9 ++
.../cxf/transport/jms/JMSDestinationTest.java | 124 +++++++++++++++++++++
3 files changed, 134 insertions(+)
[cxf] 01/02: CXF-8161: fix memory leak and thread leak in
JMSDestination. Adding more tests for JMSDestination first
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
reta pushed a commit to branch 3.2.x-fixes
in repository https://gitbox.apache.org/repos/asf/cxf.git
commit bd98adffcf470709e1e2763326cfa2a07b4f4d54
Author: reta <dr...@gmail.com>
AuthorDate: Sun Dec 8 13:56:23 2019 -0500
CXF-8161: fix memory leak and thread leak in JMSDestination. Adding more tests for JMSDestination first
(cherry picked from commit dcc9c3f74c2fe9b8797d18296dee2e6270442481)
(cherry picked from commit 53e45f515d36a31d99d0eab4e1776e5a17eeba55)
# Conflicts:
# rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/AbstractJMSTester.java
# rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSDestinationTest.java
---
.../cxf/transport/jms/AbstractJMSTester.java | 9 ++
.../cxf/transport/jms/JMSDestinationTest.java | 124 +++++++++++++++++++++
2 files changed, 133 insertions(+)
diff --git a/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/AbstractJMSTester.java b/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/AbstractJMSTester.java
index 807c4ec..5d18f50 100644
--- a/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/AbstractJMSTester.java
+++ b/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/AbstractJMSTester.java
@@ -26,6 +26,7 @@ import java.io.Reader;
import java.io.StringReader;
import java.io.Writer;
import java.net.URL;
+import java.util.function.Function;
import javax.jms.ConnectionFactory;
import javax.xml.namespace.QName;
@@ -163,6 +164,7 @@ public abstract class AbstractJMSTester extends Assert {
return new JMSConduit(target, jmsConfig, bus);
}
+
protected JMSConduit setupJMSConduitWithObserver(EndpointInfo ei) throws IOException {
JMSConduit jmsConduit = setupJMSConduit(ei);
MessageObserver observer = new MessageObserver() {
@@ -174,6 +176,13 @@ public abstract class AbstractJMSTester extends Assert {
return jmsConduit;
}
+ protected JMSDestination setupJMSDestination(EndpointInfo ei,
+ Function<ConnectionFactory, ConnectionFactory> wrapper) throws IOException {
+ JMSConfiguration jmsConfig = JMSConfigFactory.createFromEndpointInfo(bus, ei, null);
+ jmsConfig.setConnectionFactory(wrapper.apply(cf));
+ return new JMSDestination(bus, ei, jmsConfig);
+ }
+
protected JMSDestination setupJMSDestination(EndpointInfo ei) throws IOException {
JMSConfiguration jmsConfig = JMSConfigFactory.createFromEndpointInfo(bus, ei, null);
jmsConfig.setConnectionFactory(cf);
diff --git a/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSDestinationTest.java b/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSDestinationTest.java
index 89e603b..1a401fb 100644
--- a/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSDestinationTest.java
+++ b/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSDestinationTest.java
@@ -24,8 +24,13 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
import java.io.StringReader;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.InvalidClientIDException;
@@ -33,6 +38,7 @@ import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.Topic;
+import org.apache.activemq.util.ServiceStopper;
import org.apache.cxf.helpers.IOUtils;
import org.apache.cxf.message.Exchange;
import org.apache.cxf.message.ExchangeImpl;
@@ -49,6 +55,34 @@ import org.junit.Ignore;
import org.junit.Test;
public class JMSDestinationTest extends AbstractJMSTester {
+ private static final class FaultyConnectionFactory implements ConnectionFactory {
+ private final ConnectionFactory delegate;
+ private final AtomicInteger latch;
+
+ private FaultyConnectionFactory(ConnectionFactory delegate, int faults) {
+ this.delegate = delegate;
+ this.latch = new AtomicInteger(faults);
+ }
+
+ @Override
+ public Connection createConnection() throws JMSException {
+ if (latch.getAndDecrement() == 0) {
+ return delegate.createConnection();
+ } else {
+ throw new JMSException("createConnection() failed (simulated)");
+ }
+ }
+
+ @Override
+ public Connection createConnection(String userName, String password) throws JMSException {
+ if (latch.decrementAndGet() == 0) {
+ return delegate.createConnection(userName, password);
+ } else {
+ throw new JMSException("createConnection(userName, password) failed (simulated)");
+ }
+ }
+
+ }
@Test
public void testGetConfigurationFromWSDL() throws Exception {
@@ -449,6 +483,96 @@ public class JMSDestinationTest extends AbstractJMSTester {
destination.shutdown();
}
+ @Test
+ public void testMessageObserverExceptionHandling() throws Exception {
+ final CountDownLatch latch = new CountDownLatch(1);
+ EndpointInfo ei = setupServiceInfo("HelloWorldPubSubService", "HelloWorldPubSubPort");
+ JMSConduit conduit = setupJMSConduitWithObserver(ei);
+
+ JMSDestination destination = setupJMSDestination(ei);
+ destination.setMessageObserver(new MessageObserver() {
+ @Override
+ public void onMessage(Message message) {
+ try {
+ throw new RuntimeException("Error!");
+ } finally {
+ latch.countDown();
+ }
+ }
+ });
+
+ Message outMessage = new MessageImpl();
+ setupMessageHeader(outMessage);
+ Thread.sleep(500L);
+
+ sendOneWayMessage(conduit, outMessage);
+ latch.await(5, TimeUnit.SECONDS);
+
+ conduit.close();
+ destination.shutdown();
+ }
+
+ @Test
+ public void testConnectionFactoryExceptionHandling() throws Exception {
+ EndpointInfo ei = setupServiceInfo("HelloWorldPubSubService", "HelloWorldPubSubPort");
+ final Function<ConnectionFactory, ConnectionFactory> wrapper =
+ new Function<ConnectionFactory, ConnectionFactory>() {
+ @Override
+ public ConnectionFactory apply(ConnectionFactory cf) {
+ return new FaultyConnectionFactory(cf, 3);
+ }
+ };
+ JMSConduit conduit = setupJMSConduitWithObserver(ei);
+ JMSDestination destination = setupJMSDestination(ei, wrapper);
+ destination.getJmsConfig().setRetryInterval(1000);
+ destination.setMessageObserver(createMessageObserver());
+
+ Message outMessage = new MessageImpl();
+ setupMessageHeader(outMessage);
+ Thread.sleep(4000L);
+
+ sendOneWayMessage(conduit, outMessage);
+
+ // wait for the message to be got from the destination,
+ // create the thread to handler the Destination incoming message
+ waitForReceiveDestMessage();
+
+ verifyReceivedMessage(destMessage);
+ verifyHeaders(destMessage, outMessage);
+
+ conduit.close();
+ destination.shutdown();
+ }
+
+ @Test
+ public void testBrokerExceptionHandling() throws Exception {
+ EndpointInfo ei = setupServiceInfo("HelloWorldPubSubService", "HelloWorldPubSubPort");
+ JMSConduit conduit = setupJMSConduitWithObserver(ei);
+ JMSDestination destination = setupJMSDestination(ei);
+ destination.getJmsConfig().setRetryInterval(1000);
+ destination.setMessageObserver(createMessageObserver());
+
+ Thread.sleep(500L);
+ broker.stopAllConnectors(new ServiceStopper());
+
+ broker.startAllConnectors();
+ Thread.sleep(2000L);
+
+ Message outMessage = new MessageImpl();
+ setupMessageHeader(outMessage);
+ sendOneWayMessage(conduit, outMessage);
+
+ // wait for the message to be got from the destination,
+ // create the thread to handler the Destination incoming message
+ waitForReceiveDestMessage();
+
+ verifyReceivedMessage(destMessage);
+ verifyHeaders(destMessage, outMessage);
+
+ conduit.close();
+ destination.shutdown();
+ }
+
private String getQueueName(String exName) {
if (exName == null) {
return null;
[cxf] 02/02: Recording .gitmergeinfo Changes
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
reta pushed a commit to branch 3.2.x-fixes
in repository https://gitbox.apache.org/repos/asf/cxf.git
commit a1bd091cb8a8673f2bb6d46aa1607339cf374e17
Author: reta <dr...@gmail.com>
AuthorDate: Tue Dec 10 20:03:11 2019 -0500
Recording .gitmergeinfo Changes
---
.gitmergeinfo | 1 +
1 file changed, 1 insertion(+)
diff --git a/.gitmergeinfo b/.gitmergeinfo
index 97637b9..efa8724 100644
--- a/.gitmergeinfo
+++ b/.gitmergeinfo
@@ -774,6 +774,7 @@ M 4fea734b1349ba023c2e560deda4aa0eaede7f3b
M 50360a938a6e6d807560ac89748fb9f931d44fcf
M 504a1b7827bc76f3c5106b901b44e54513db17aa
M 52f2cc152ceccd3981361dc840338c16416786fa
+M 53e45f515d36a31d99d0eab4e1776e5a17eeba55
M 56f74c18d7c709307b174cb211b28e4384574e9b
M 57f8c554b04af4e06b4e97bb349cbb8adeeb7acc
M 5827b0bb9d5a62f420f5277ffa456fd8a6083ab0