You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by nf...@apache.org on 2017/05/15 10:36:56 UTC
[1/2] camel git commit: CAMEL-11140: re-organizing jmx data
Repository: camel
Updated Branches:
refs/heads/master f2cfc3582 -> 4120c1fd1
CAMEL-11140: re-organizing jmx data
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/4120c1fd
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/4120c1fd
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/4120c1fd
Branch: refs/heads/master
Commit: 4120c1fd1f43fee615d27a6a1d5997758b12e935
Parents: 3960dcc
Author: Nicola Ferraro <ni...@gmail.com>
Authored: Mon May 15 12:34:03 2017 +0200
Committer: Nicola Ferraro <ni...@gmail.com>
Committed: Mon May 15 12:34:13 2017 +0200
----------------------------------------------------------------------
.../reactive/streams/engine/CamelPublisher.java | 8 +-
.../streams/engine/CamelSubscriber.java | 16 ---
.../streams/engine/CamelSubscription.java | 16 ++-
.../DefaultCamelReactiveStreamsService.java | 62 ++++++---
.../streams/ReactiveStreamsJMXTest.java | 132 +++++++++++++++++++
5 files changed, 191 insertions(+), 43 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/4120c1fd/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelPublisher.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelPublisher.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelPublisher.java
index bdd712c..0012984 100644
--- a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelPublisher.java
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelPublisher.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.reactive.streams.engine;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
+import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
@@ -63,7 +64,7 @@ public class CamelPublisher implements Publisher<Exchange>, AutoCloseable {
@Override
public void subscribe(Subscriber<? super Exchange> subscriber) {
Objects.requireNonNull(subscriber, "subscriber must not be null");
- CamelSubscription sub = new CamelSubscription(workerPool, this, name, this.backpressureStrategy, subscriber);
+ CamelSubscription sub = new CamelSubscription(UUID.randomUUID().toString(), workerPool, this, name, this.backpressureStrategy, subscriber);
this.subscriptions.add(sub);
subscriber.onSubscribe(sub);
}
@@ -138,7 +139,8 @@ public class CamelPublisher implements Publisher<Exchange>, AutoCloseable {
subscriptions.clear();
}
- public int getSubscriptionSize() {
- return subscriptions.size();
+ public List<CamelSubscription> getSubscriptions() {
+ return subscriptions;
}
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/4120c1fd/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscriber.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscriber.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscriber.java
index 6df386f..94a9c46 100644
--- a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscriber.java
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscriber.java
@@ -20,7 +20,6 @@ import java.io.Closeable;
import java.io.IOException;
import org.apache.camel.Exchange;
-import org.apache.camel.component.reactive.streams.ReactiveStreamsBackpressureStrategy;
import org.apache.camel.component.reactive.streams.ReactiveStreamsConsumer;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
@@ -208,19 +207,4 @@ public class CamelSubscriber implements Subscriber<Exchange>, Closeable {
return inflightCount;
}
- public long getBufferSize() {
- if (subscription != null && subscription instanceof CamelSubscription) {
- return ((CamelSubscription) subscription).getBufferSize();
- } else {
- return 0;
- }
- }
-
- public ReactiveStreamsBackpressureStrategy getBackpressureStrategy() {
- if (subscription != null && subscription instanceof CamelSubscription) {
- return ((CamelSubscription) subscription).getBackpressureStrategy();
- } else {
- return null;
- }
- }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/4120c1fd/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscription.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscription.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscription.java
index efc676c..d7e17c8 100644
--- a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscription.java
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscription.java
@@ -43,9 +43,11 @@ public class CamelSubscription implements Subscription {
private static final Logger LOG = LoggerFactory.getLogger(CamelSubscription.class);
+ private String id;
+
private ExecutorService workerPool;
- private String name;
+ private String streamName;
private CamelPublisher publisher;
@@ -82,10 +84,12 @@ public class CamelSubscription implements Subscription {
private boolean sending;
- public CamelSubscription(ExecutorService workerPool, CamelPublisher publisher, String name, ReactiveStreamsBackpressureStrategy backpressureStrategy, Subscriber<? super Exchange> subscriber) {
+ public CamelSubscription(String id, ExecutorService workerPool, CamelPublisher publisher, String streamName,
+ ReactiveStreamsBackpressureStrategy backpressureStrategy, Subscriber<? super Exchange> subscriber) {
+ this.id = id;
this.workerPool = workerPool;
this.publisher = publisher;
- this.name = name;
+ this.streamName = streamName;
this.backpressureStrategy = backpressureStrategy;
this.subscriber = subscriber;
}
@@ -241,7 +245,7 @@ public class CamelSubscription implements Subscription {
for (Exchange exchange: discardedMessages.keySet()) {
ReactiveStreamsHelper.invokeDispatchCallback(
exchange,
- new ReactiveStreamsDiscardedException("Discarded by backpressure strategy", exchange, name)
+ new ReactiveStreamsDiscardedException("Discarded by backpressure strategy", exchange, streamName)
);
}
}
@@ -262,4 +266,8 @@ public class CamelSubscription implements Subscription {
public ReactiveStreamsBackpressureStrategy getBackpressureStrategy() {
return backpressureStrategy;
}
+
+ public String getId() {
+ return id;
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/4120c1fd/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/DefaultCamelReactiveStreamsService.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/DefaultCamelReactiveStreamsService.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/DefaultCamelReactiveStreamsService.java
index 9b0e82f..06650d8 100644
--- a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/DefaultCamelReactiveStreamsService.java
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/DefaultCamelReactiveStreamsService.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.reactive.streams.engine;
+import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
@@ -322,20 +323,18 @@ public class DefaultCamelReactiveStreamsService extends ServiceSupport implement
@ManagedOperation(description = "Information about Camel Reactive subscribers")
public TabularData camelSubscribers() {
try {
- final TabularData answer = new TabularDataSupport(subscriptionsTabularType());
+ final TabularData answer = new TabularDataSupport(subscribersTabularType());
subscribers.forEach((k, v) -> {
try {
String name = k;
long inflight = v.getInflightCount();
long requested = v.getRequested();
- long bufferSize = v.getBufferSize();
- String backpressure = v.getBackpressureStrategy() != null ? v.getBackpressureStrategy().name() : "";
- CompositeType ct = subscriptionsCompositeType();
+ CompositeType ct = subscribersCompositeType();
CompositeData data = new CompositeDataSupport(ct,
- new String[] {"name", "inflight", "requested", "buffer size", "back pressure"},
- new Object[] {name, inflight, requested, bufferSize, backpressure});
+ new String[] {"name", "inflight", "requested"},
+ new Object[] {name, inflight, requested});
answer.put(data);
} catch (Exception e) {
throw ObjectHelper.wrapRuntimeCamelException(e);
@@ -356,14 +355,25 @@ public class DefaultCamelReactiveStreamsService extends ServiceSupport implement
publishers.forEach((k, v) -> {
try {
String name = k;
- int subscribers = v.getSubscriptionSize();
+ List<CamelSubscription> subscriptions = v.getSubscriptions();
+ int subscribers = subscriptions.size();
+
+ TabularData subscriptionData = new TabularDataSupport(subscriptionsTabularType());
+ CompositeType subCt = subscriptionsCompositeType();
+ for (CamelSubscription sub : subscriptions) {
+ String id = sub.getId();
+ long bufferSize = sub.getBufferSize();
+ String backpressure = sub.getBackpressureStrategy() != null ? sub.getBackpressureStrategy().name() : "";
+ CompositeData subData = new CompositeDataSupport(subCt, new String[]{"name", "buffer size", "back pressure"}, new Object[]{id, bufferSize, backpressure});
+
+ subscriptionData.put(subData);
+ }
- // TODO: include more subscriber information, either as a nested table or flattern
CompositeType ct = publishersCompositeType();
CompositeData data = new CompositeDataSupport(ct,
- new String[] {"name", "subscribers"},
- new Object[] {name, subscribers});
+ new String[] {"name", "subscribers", "subscriptions"},
+ new Object[] {name, subscribers, subscriptionData});
answer.put(data);
} catch (Exception e) {
throw ObjectHelper.wrapRuntimeCamelException(e);
@@ -376,23 +386,35 @@ public class DefaultCamelReactiveStreamsService extends ServiceSupport implement
}
}
- private static CompositeType subscriptionsCompositeType() throws OpenDataException {
+ private static TabularType subscribersTabularType() throws OpenDataException {
+ CompositeType ct = subscribersCompositeType();
+ return new TabularType("subscribers", "Information about Camel Reactive subscribers", ct, new String[]{"name"});
+ }
+
+ private static CompositeType subscribersCompositeType() throws OpenDataException {
return new CompositeType("subscriptions", "Subscriptions",
- new String[] {"name", "inflight", "requested", "buffer size", "back pressure"},
- new String[] {"Name", "Inflight", "Requested", "Buffer Size", "Back Pressure"},
- new OpenType[] {SimpleType.STRING, SimpleType.LONG, SimpleType.LONG, SimpleType.LONG, SimpleType.STRING});
+ new String[] {"name", "inflight", "requested"},
+ new String[] {"Name", "Inflight", "Requested"},
+ new OpenType[] {SimpleType.STRING, SimpleType.LONG, SimpleType.LONG});
+ }
+
+ private static CompositeType publishersCompositeType() throws OpenDataException {
+ return new CompositeType("publishers", "Publishers",
+ new String[] {"name", "subscribers", "subscriptions"},
+ new String[] {"Name", "Subscribers", "Subscriptions"},
+ new OpenType[] {SimpleType.STRING, SimpleType.INTEGER, subscriptionsTabularType()});
}
private static TabularType subscriptionsTabularType() throws OpenDataException {
CompositeType ct = subscriptionsCompositeType();
- return new TabularType("subscriptions", "Information about Camel Reactive subscribers", ct, new String[]{"name"});
+ return new TabularType("subscriptions", "Information about External Reactive subscribers", ct, new String[]{"name"});
}
- private static CompositeType publishersCompositeType() throws OpenDataException {
- return new CompositeType("publishers", "Publishers",
- new String[] {"name", "subscribers"},
- new String[] {"Name", "Subscribers"},
- new OpenType[] {SimpleType.STRING, SimpleType.INTEGER});
+ private static CompositeType subscriptionsCompositeType() throws OpenDataException {
+ return new CompositeType("subscriptions", "Subscriptions",
+ new String[] {"name", "buffer size", "back pressure"},
+ new String[] {"Name", "Buffer Size", "Back Pressure"},
+ new OpenType[] {SimpleType.STRING, SimpleType.LONG, SimpleType.STRING});
}
private static TabularType publishersTabularType() throws OpenDataException {
http://git-wip-us.apache.org/repos/asf/camel/blob/4120c1fd/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ReactiveStreamsJMXTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ReactiveStreamsJMXTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ReactiveStreamsJMXTest.java
new file mode 100644
index 0000000..55789a1
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ReactiveStreamsJMXTest.java
@@ -0,0 +1,132 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.reactive.streams;
+
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Set;
+import javax.management.AttributeValueExp;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+import javax.management.Query;
+import javax.management.QueryExp;
+import javax.management.StringValueExp;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.TabularData;
+
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreamsService;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * Test exposed services on JMX.
+ */
+public class ReactiveStreamsJMXTest extends CamelTestSupport {
+
+ @Test
+ public void testJmxExposedService() throws Exception {
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName rxService = getReactiveStreamsServiceName(mbeanServer);
+
+ Map<String, Object> subdata = getValues(mbeanServer, rxService, "camelSubscribers", 0);
+ assertEquals("unbounded", subdata.get("name"));
+ assertEquals(0L, subdata.get("inflight"));
+ assertEquals(0L, subdata.get("requested"));
+
+ Map<String, Object> pubdata = getValues(mbeanServer, rxService, "camelPublishers", 0);
+ assertEquals("strings", pubdata.get("name"));
+ assertEquals(1, pubdata.get("subscribers"));
+
+
+ TabularData subTd0 = (TabularData) pubdata.get("subscriptions");
+ assertEquals(1, subTd0.values().size());
+
+ // Create another subscriber
+ CamelReactiveStreamsService rxCamel = CamelReactiveStreams.get(context());
+ Flux.from(rxCamel.fromStream("strings"))
+ .subscribe();
+
+ pubdata = getValues(mbeanServer, rxService, "camelPublishers", 0);
+ TabularData subTd = (TabularData) pubdata.get("subscriptions");
+ assertEquals(2, subTd.values().size());
+ Map<String, Object> subscriptions = getValues(subTd, 1);
+ assertEquals("BUFFER", subscriptions.get("back pressure"));
+ }
+
+ private Map<String, Object> getValues(MBeanServer mbeanServer, ObjectName rxService, String name, int index) throws Exception {
+ TabularData td = (TabularData) mbeanServer.invoke(rxService, name, null, null);
+ return getValues(td, index);
+ }
+
+ private Map<String, Object> getValues(TabularData td, int index) {
+ assertNotNull(td);
+ assertTrue(td.values().size() > index);
+ CompositeData data = (CompositeData) new LinkedList<>(td.values()).get(index);
+ assertNotNull(data);
+ Map<String, Object> res = new HashMap<>();
+ for (String key : data.getCompositeType().keySet()) {
+ res.put(key, data.get(key));
+ }
+ return res;
+ }
+
+ private MBeanServer getMBeanServer() {
+ return context.getManagementStrategy().getManagementAgent().getMBeanServer();
+ }
+
+ private ObjectName getReactiveStreamsServiceName(MBeanServer mbeanServer) throws Exception {
+ ObjectName on = ObjectName.getInstance("org.apache.camel:type=services,*");
+ QueryExp queryExp = Query.match(new AttributeValueExp("ServiceType"), new StringValueExp("DefaultCamelReactiveStreamsService"));
+ Set<ObjectName> names = mbeanServer.queryNames(on, queryExp);
+ assertEquals(1, names.size());
+ return names.iterator().next();
+ }
+
+ @Override
+ protected boolean useJmx() {
+ return true;
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("reactive-streams:unbounded?maxInflightExchanges=-1")
+ .delayer(1)
+ .to("mock:unbounded-endpoint");
+
+ from("timer:tick")
+ .setBody().simple("Hello world ${header.CamelTimerCounter}")
+ .to("reactive-streams:strings");
+
+
+ CamelReactiveStreamsService rxCamel = CamelReactiveStreams.get(getContext());
+
+ Flux.from(rxCamel.fromStream("strings", String.class))
+ .doOnNext(System.out::println)
+ .subscribe();
+ }
+ };
+ }
+}
\ No newline at end of file
[2/2] camel git commit: CAMEL-11140: closing Camel context in tck
tests
Posted by nf...@apache.org.
CAMEL-11140: closing Camel context in tck tests
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/3960dccf
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/3960dccf
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/3960dccf
Branch: refs/heads/master
Commit: 3960dccf137c31fd06474ad47dc6bdcd13df30a9
Parents: f2cfc35
Author: Nicola Ferraro <ni...@gmail.com>
Authored: Mon May 15 12:21:12 2017 +0200
Committer: Nicola Ferraro <ni...@gmail.com>
Committed: Mon May 15 12:34:13 2017 +0200
----------------------------------------------------------------------
...amelPublisherConversionVerificationTest.java | 30 ++++++++++++++++++--
.../tck/CamelPublisherVerificationTest.java | 26 ++++++++++++++++-
...melSubscriberConversionVerificationTest.java | 26 ++++++++++++++++-
.../tck/CamelSubscriberVerificationTest.java | 25 +++++++++++++++-
4 files changed, 101 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/3960dccf/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelPublisherConversionVerificationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelPublisherConversionVerificationTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelPublisherConversionVerificationTest.java
index 1817fdf..429e08c 100644
--- a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelPublisherConversionVerificationTest.java
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelPublisherConversionVerificationTest.java
@@ -20,20 +20,25 @@ import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.impl.DefaultShutdownStrategy;
import org.reactivestreams.Publisher;
import org.reactivestreams.tck.PublisherVerification;
import org.reactivestreams.tck.TestEnvironment;
+import org.testng.annotations.AfterTest;
public class CamelPublisherConversionVerificationTest extends PublisherVerification<Long> {
+ private CamelContext context;
+
public CamelPublisherConversionVerificationTest() {
super(new TestEnvironment(2000L));
}
+
@Override
public Publisher<Long> createPublisher(long l) {
+ init();
- CamelContext context = new DefaultCamelContext();
RouteBuilder builder = new RouteBuilder() {
@Override
public void configure() throws Exception {
@@ -43,8 +48,6 @@ public class CamelPublisherConversionVerificationTest extends PublisherVerificat
}
};
- Publisher<Long> pub = CamelReactiveStreams.get(context).fromStream("prod", Long.class);
-
try {
builder.addRoutesToCamelContext(context);
context.start();
@@ -52,6 +55,7 @@ public class CamelPublisherConversionVerificationTest extends PublisherVerificat
throw new RuntimeException(e);
}
+ Publisher<Long> pub = CamelReactiveStreams.get(context).fromStream("prod", Long.class);
return pub;
}
@@ -65,4 +69,24 @@ public class CamelPublisherConversionVerificationTest extends PublisherVerificat
public Publisher<Long> createFailedPublisher() {
return null;
}
+
+ protected void init() {
+ tearDown();
+ this.context = new DefaultCamelContext();
+ DefaultShutdownStrategy shutdownStrategy = new DefaultShutdownStrategy();
+ shutdownStrategy.setShutdownNowOnTimeout(true);
+ shutdownStrategy.setTimeout(1);
+ this.context.setShutdownStrategy(shutdownStrategy);
+ }
+
+ @AfterTest
+ protected void tearDown() {
+ try {
+ if (this.context != null) {
+ this.context.stop();
+ }
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/3960dccf/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelPublisherVerificationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelPublisherVerificationTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelPublisherVerificationTest.java
index f3a705e..8e0cd7d 100644
--- a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelPublisherVerificationTest.java
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelPublisherVerificationTest.java
@@ -21,20 +21,24 @@ import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.impl.DefaultShutdownStrategy;
import org.reactivestreams.Publisher;
import org.reactivestreams.tck.PublisherVerification;
import org.reactivestreams.tck.TestEnvironment;
+import org.testng.annotations.AfterTest;
public class CamelPublisherVerificationTest extends PublisherVerification<Exchange> {
+ private CamelContext context;
+
public CamelPublisherVerificationTest() {
super(new TestEnvironment(2000L));
}
@Override
public Publisher<Exchange> createPublisher(long l) {
+ init();
- CamelContext context = new DefaultCamelContext();
RouteBuilder builder = new RouteBuilder() {
@Override
public void configure() throws Exception {
@@ -65,4 +69,24 @@ public class CamelPublisherVerificationTest extends PublisherVerification<Exchan
public Publisher<Exchange> createFailedPublisher() {
return null;
}
+
+ protected void init() {
+ tearDown();
+ this.context = new DefaultCamelContext();
+ DefaultShutdownStrategy shutdownStrategy = new DefaultShutdownStrategy();
+ shutdownStrategy.setShutdownNowOnTimeout(true);
+ shutdownStrategy.setTimeout(1);
+ this.context.setShutdownStrategy(shutdownStrategy);
+ }
+
+ @AfterTest
+ protected void tearDown() {
+ try {
+ if (this.context != null) {
+ this.context.stop();
+ }
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/3960dccf/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelSubscriberConversionVerificationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelSubscriberConversionVerificationTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelSubscriberConversionVerificationTest.java
index 33c44a1..6e6cca8 100644
--- a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelSubscriberConversionVerificationTest.java
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelSubscriberConversionVerificationTest.java
@@ -20,9 +20,11 @@ import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.impl.DefaultShutdownStrategy;
import org.reactivestreams.Subscriber;
import org.reactivestreams.tck.SubscriberBlackboxVerification;
import org.reactivestreams.tck.TestEnvironment;
+import org.testng.annotations.AfterTest;
public class CamelSubscriberConversionVerificationTest extends SubscriberBlackboxVerification<Integer> {
@@ -34,7 +36,8 @@ public class CamelSubscriberConversionVerificationTest extends SubscriberBlackbo
@Override
public Subscriber<Integer> createSubscriber() {
- this.context = new DefaultCamelContext();
+ init();
+
RouteBuilder builder = new RouteBuilder() {
@Override
public void configure() throws Exception {
@@ -59,4 +62,25 @@ public class CamelSubscriberConversionVerificationTest extends SubscriberBlackbo
public Integer createElement(int element) {
return element;
}
+
+ protected void init() {
+ tearDown();
+ this.context = new DefaultCamelContext();
+ DefaultShutdownStrategy shutdownStrategy = new DefaultShutdownStrategy();
+ shutdownStrategy.setShutdownNowOnTimeout(true);
+ shutdownStrategy.setTimeout(1);
+ this.context.setShutdownStrategy(shutdownStrategy);
+ }
+
+ @AfterTest
+ protected void tearDown() {
+ try {
+ if (this.context != null) {
+ this.context.stop();
+ }
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/3960dccf/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelSubscriberVerificationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelSubscriberVerificationTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelSubscriberVerificationTest.java
index 75bb064..fb730c2 100644
--- a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelSubscriberVerificationTest.java
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/tck/CamelSubscriberVerificationTest.java
@@ -22,9 +22,11 @@ import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.impl.DefaultExchange;
+import org.apache.camel.impl.DefaultShutdownStrategy;
import org.reactivestreams.Subscriber;
import org.reactivestreams.tck.SubscriberBlackboxVerification;
import org.reactivestreams.tck.TestEnvironment;
+import org.testng.annotations.AfterTest;
public class CamelSubscriberVerificationTest extends SubscriberBlackboxVerification<Exchange> {
@@ -36,7 +38,8 @@ public class CamelSubscriberVerificationTest extends SubscriberBlackboxVerificat
@Override
public Subscriber<Exchange> createSubscriber() {
- this.context = new DefaultCamelContext();
+ init();
+
RouteBuilder builder = new RouteBuilder() {
@Override
public void configure() throws Exception {
@@ -63,4 +66,24 @@ public class CamelSubscriberVerificationTest extends SubscriberBlackboxVerificat
exchange.getIn().setBody(element);
return exchange;
}
+
+ protected void init() {
+ tearDown();
+ this.context = new DefaultCamelContext();
+ DefaultShutdownStrategy shutdownStrategy = new DefaultShutdownStrategy();
+ shutdownStrategy.setShutdownNowOnTimeout(true);
+ shutdownStrategy.setTimeout(1);
+ this.context.setShutdownStrategy(shutdownStrategy);
+ }
+
+ @AfterTest
+ protected void tearDown() {
+ try {
+ if (this.context != null) {
+ this.context.stop();
+ }
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
}