You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by an...@apache.org on 2016/06/24 19:10:45 UTC
camel git commit: CAMEL-10024: Cherry picked from master
Repository: camel
Updated Branches:
refs/heads/camel-2.17.x 38c75c5cc -> 1d3e4a2d6
CAMEL-10024: Cherry picked from master
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/1d3e4a2d
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/1d3e4a2d
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/1d3e4a2d
Branch: refs/heads/camel-2.17.x
Commit: 1d3e4a2d6dff57b9487574972df8098cf0d2adcd
Parents: 38c75c5
Author: Arno Noordover <an...@users.noreply.github.com>
Authored: Sat Jun 18 17:11:31 2016 +0200
Committer: Arno Noordover <an...@users.noreply.github.com>
Committed: Fri Jun 24 21:06:44 2016 +0200
----------------------------------------------------------------------
.../camel/component/mina2/Mina2Consumer.java | 6 +-
.../camel/component/mina2/Mina2Producer.java | 74 +++++++++++---------
.../component/mina2/Mina2TextLineDelimiter.java | 18 ++++-
...Mina2ClientModeTcpTextlineDelimiterTest.java | 2 +-
.../mina2/Mina2DisconnectRaceConditionTest.java | 70 ++++++++++++++++++
.../component/mina2/Mina2EncodingTest.java | 2 +-
.../mina2/Mina2ExchangeTimeOutTest.java | 2 +-
.../mina2/Mina2NoResponseFromServerTest.java | 4 +-
.../mina2/Mina2ProducerShutdownMockTest.java | 7 +-
.../mina2/Mina2ReverseProtocolHandler.java | 2 +-
.../mina2/Mina2TransferExchangeOptionTest.java | 2 +-
11 files changed, 141 insertions(+), 48 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/1d3e4a2d/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2Consumer.java
----------------------------------------------------------------------
diff --git a/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2Consumer.java b/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2Consumer.java
index 2e913e7..ac461da 100644
--- a/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2Consumer.java
+++ b/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2Consumer.java
@@ -115,7 +115,7 @@ public class Mina2Consumer extends DefaultConsumer {
if (configuration.isClientMode() && configuration.getProtocol().equals("tcp")) {
LOG.info("Disconnect from server address: {} using connector: {}", address, connector);
if (session != null) {
- CloseFuture closeFuture = session.close(true);
+ CloseFuture closeFuture = session.closeNow();
closeFuture.awaitUninterruptibly();
}
connector.dispose(true);
@@ -382,7 +382,7 @@ public class Mina2Consumer extends DefaultConsumer {
// close invalid session
if (session != null) {
LOG.warn("Closing session as an exception was thrown from MINA");
- session.close(true);
+ session.closeNow();
}
// must wrap and rethrow since cause can be of Throwable and we must only throw Exception
@@ -456,7 +456,7 @@ public class Mina2Consumer extends DefaultConsumer {
}
if (disconnect) {
LOG.debug("Closing session when complete at address: {}", address);
- session.close(true);
+ session.closeNow();
}
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/1d3e4a2d/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2Producer.java
----------------------------------------------------------------------
diff --git a/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2Producer.java b/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2Producer.java
index 132abe3..8e3eaec 100644
--- a/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2Producer.java
+++ b/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2Producer.java
@@ -65,8 +65,10 @@ import org.slf4j.LoggerFactory;
public class Mina2Producer extends DefaultProducer implements ServicePoolAware {
private static final Logger LOG = LoggerFactory.getLogger(Mina2Producer.class);
+ private final ResponseHandler handler;
private IoSession session;
- private CountDownLatch latch;
+ private CountDownLatch responseLatch;
+ private CountDownLatch closeLatch;
private boolean lazySessionCreation;
private long timeout;
private SocketAddress address;
@@ -93,6 +95,8 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware {
} else if (protocol.equals("vm")) {
setupVmProtocol(protocol);
}
+ handler = new ResponseHandler();
+ connector.setHandler(handler);
}
@Override
@@ -143,10 +147,9 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware {
// if sync is true then we should also wait for a response (synchronous mode)
if (sync) {
- // only initialize latch if we should get a response
- latch = new CountDownLatch(1);
+ // only initialize responseLatch if we should get a response
+ responseLatch = new CountDownLatch(1);
// reset handler if we expect a response
- ResponseHandler handler = (ResponseHandler) session.getHandler();
handler.reset();
}
@@ -165,13 +168,12 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware {
if (sync) {
// wait for response, consider timeout
LOG.debug("Waiting for response using timeout {} millis.", timeout);
- boolean done = latch.await(timeout, TimeUnit.MILLISECONDS);
+ boolean done = responseLatch.await(timeout, TimeUnit.MILLISECONDS);
if (!done) {
throw new ExchangeTimedOutException(exchange, timeout);
}
// did we get a response
- ResponseHandler handler = (ResponseHandler) session.getHandler();
if (handler.getCause() != null) {
throw new CamelExchangeException("Error occurred in ResponseHandler", exchange, handler.getCause());
} else if (!handler.isMessageReceived()) {
@@ -188,7 +190,7 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware {
}
}
- protected void maybeDisconnectOnDone(Exchange exchange) {
+ protected void maybeDisconnectOnDone(Exchange exchange) throws InterruptedException {
if (session == null) {
return;
}
@@ -208,7 +210,16 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware {
}
if (disconnect) {
LOG.debug("Closing session when complete at address: {}", address);
- session.close(true);
+ closeSessionIfNeededAndAwaitCloseInHandler(session);
+ }
+ }
+
+ private void closeSessionIfNeededAndAwaitCloseInHandler(IoSession sessionToBeClosed) throws InterruptedException {
+ closeLatch = new CountDownLatch(1);
+ if (!sessionToBeClosed.isClosing()) {
+ CloseFuture closeFuture = sessionToBeClosed.closeNow();
+ closeFuture.await(timeout, TimeUnit.MILLISECONDS);
+ closeLatch.await(timeout, TimeUnit.MILLISECONDS);
}
}
@@ -241,10 +252,9 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware {
super.doShutdown();
}
- private void closeConnection() {
+ private void closeConnection() throws InterruptedException {
if (session != null) {
- CloseFuture closeFuture = session.close(true);
- closeFuture.awaitUninterruptibly();
+ closeSessionIfNeededAndAwaitCloseInHandler(session);
}
connector.dispose(true);
@@ -255,14 +265,13 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware {
setSocketAddress(this.configuration.getProtocol());
}
if (LOG.isDebugEnabled()) {
- LOG.debug("Creating connector to address: {} using connector: {} timeout: {} millis.", new Object[]{address, connector, timeout});
+ LOG.debug("Creating connector to address: {} using connector: {} timeout: {} millis.", address, connector, timeout);
}
// connect and wait until the connection is established
if (connectorConfig != null) {
connector.getSessionConfig().setAll(connectorConfig);
}
- connector.setHandler(new ResponseHandler());
ConnectFuture future = connector.connect(address);
future.awaitUninterruptibly();
session = future.getSession();
@@ -342,7 +351,7 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware {
}
addCodecFactory(service, codecFactory);
LOG.debug("{}: Using TextLineCodecFactory: {} using encoding: {} line delimiter: {}({})",
- new Object[]{type, codecFactory, charset, configuration.getTextlineDelimiter(), delimiter});
+ type, codecFactory, charset, configuration.getTextlineDelimiter(), delimiter);
LOG.debug("Encoder maximum line length: {}. Decoder maximum line length: {}",
codecFactory.getEncoderMaxLineLength(), codecFactory.getDecoderMaxLineLength());
} else {
@@ -412,21 +421,7 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware {
if (delimiter == null) {
return LineDelimiter.DEFAULT;
}
-
- switch (delimiter) {
- case DEFAULT:
- return LineDelimiter.DEFAULT;
- case AUTO:
- return LineDelimiter.AUTO;
- case UNIX:
- return LineDelimiter.UNIX;
- case WINDOWS:
- return LineDelimiter.WINDOWS;
- case MAC:
- return LineDelimiter.MAC;
- default:
- throw new IllegalArgumentException("Unknown textline delimiter: " + delimiter);
- }
+ return delimiter.getLineDelimiter();
}
private Charset getEncodingParameter(String type, Mina2Configuration configuration) {
@@ -483,11 +478,11 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware {
this.message = message;
messageReceived = true;
cause = null;
- countDown();
+ notifyResultAvailable();
}
- protected void countDown() {
- CountDownLatch downLatch = latch;
+ protected void notifyResultAvailable() {
+ CountDownLatch downLatch = responseLatch;
if (downLatch != null) {
downLatch.countDown();
}
@@ -500,7 +495,14 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware {
LOG.debug("Session closed but no message received from address: {}", address);
// session was closed but no message received. This could be because the remote server had an internal error
// and could not return a response. We should count down to stop waiting for a response
- countDown();
+ notifyResultAvailable();
+ }
+ notifySessionClosed();
+ }
+
+ private void notifySessionClosed() {
+ if (closeLatch != null) {
+ closeLatch.countDown();
}
}
@@ -512,7 +514,11 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware {
this.messageReceived = false;
this.cause = cause;
if (ioSession != null) {
- ioSession.close(true);
+ try {
+ closeSessionIfNeededAndAwaitCloseInHandler(ioSession);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/1d3e4a2d/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2TextLineDelimiter.java
----------------------------------------------------------------------
diff --git a/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2TextLineDelimiter.java b/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2TextLineDelimiter.java
index 8bf87c7..bc83a7e 100644
--- a/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2TextLineDelimiter.java
+++ b/components/camel-mina2/src/main/java/org/apache/camel/component/mina2/Mina2TextLineDelimiter.java
@@ -16,10 +16,26 @@
*/
package org.apache.camel.component.mina2;
+import org.apache.mina.filter.codec.textline.LineDelimiter;
+
/**
* Possible text line delimiters to be used with the textline codec.
*/
public enum Mina2TextLineDelimiter {
- DEFAULT, AUTO, UNIX, WINDOWS, MAC
+ DEFAULT(LineDelimiter.DEFAULT),
+ AUTO(LineDelimiter.AUTO),
+ UNIX(LineDelimiter.UNIX),
+ WINDOWS(LineDelimiter.WINDOWS),
+ MAC(LineDelimiter.MAC);
+
+ private final LineDelimiter lineDelimiter;
+
+ Mina2TextLineDelimiter(LineDelimiter lineDelimiter) {
+ this.lineDelimiter = lineDelimiter;
+ }
+
+ public LineDelimiter getLineDelimiter() {
+ return lineDelimiter;
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/1d3e4a2d/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ClientModeTcpTextlineDelimiterTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ClientModeTcpTextlineDelimiterTest.java b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ClientModeTcpTextlineDelimiterTest.java
index b160add..62229fc 100644
--- a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ClientModeTcpTextlineDelimiterTest.java
+++ b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ClientModeTcpTextlineDelimiterTest.java
@@ -85,7 +85,7 @@ public class Mina2ClientModeTcpTextlineDelimiterTest extends BaseMina2Test {
private class ServerHandler extends IoHandlerAdapter {
public void sessionOpened(IoSession session) throws Exception {
session.write("Hello there!\n");
- session.close(true);
+ session.closeNow();
}
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/1d3e4a2d/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2DisconnectRaceConditionTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2DisconnectRaceConditionTest.java b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2DisconnectRaceConditionTest.java
new file mode 100644
index 0000000..6b4b353
--- /dev/null
+++ b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2DisconnectRaceConditionTest.java
@@ -0,0 +1,70 @@
+/**
+ * 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.mina2;
+
+import java.lang.reflect.Field;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultExchange;
+import org.apache.mina.core.session.IoSession;
+import org.junit.Test;
+
+public class Mina2DisconnectRaceConditionTest extends BaseMina2Test {
+
+ /**
+ * This is a test for issue CAMEL-10024 - the closing must complete before we return from the producer
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testCloseSessionWhenCompleteManyTimes() throws Exception {
+ final String endpointUri = String.format("mina2:tcp://localhost:%1$s?sync=true&textline=true&disconnect=true&minaLogger=true", getPort());
+ Mina2Producer producer = (Mina2Producer) context.getEndpoint(endpointUri).createProducer();
+ // Access session to check that the session is really closed
+ Field field = producer.getClass().getDeclaredField("session");
+ field.setAccessible(true);
+
+ for (int i = 0; i < 100; i++) {
+ Exchange e = new DefaultExchange(context, ExchangePattern.InOut);
+ e.getIn().setBody("Chad");
+ producer.process(e);
+ final IoSession ioSession = (IoSession) field.get(producer);
+ assertTrue(ioSession.getCloseFuture().isDone());
+ Object out = e.getOut().getBody();
+ assertEquals("Bye Chad", out);
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+
+ public void configure() throws Exception {
+ from(String.format("mina2:tcp://localhost:%1$s?sync=true&textline=true", getPort())).process(new Processor() {
+
+ public void process(Exchange exchange) throws Exception {
+ String body = exchange.getIn().getBody(String.class);
+ exchange.getOut().setBody("Bye " + body);
+ }
+ });
+ }
+ };
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/1d3e4a2d/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2EncodingTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2EncodingTest.java b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2EncodingTest.java
index f6130c1..72d4994 100644
--- a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2EncodingTest.java
+++ b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2EncodingTest.java
@@ -170,7 +170,7 @@ public class Mina2EncodingTest extends BaseMina2Test {
Endpoint endpoint = context.getEndpoint(uri);
Producer producer = endpoint.createProducer();
- Exchange exchange = producer.createExchange();
+ Exchange exchange = endpoint.createExchange();
exchange.getIn().setBody(hello);
producer.start();
http://git-wip-us.apache.org/repos/asf/camel/blob/1d3e4a2d/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ExchangeTimeOutTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ExchangeTimeOutTest.java b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ExchangeTimeOutTest.java
index 512c1f6..d24743f 100644
--- a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ExchangeTimeOutTest.java
+++ b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ExchangeTimeOutTest.java
@@ -35,7 +35,7 @@ public class Mina2ExchangeTimeOutTest extends BaseMina2Test {
Endpoint endpoint = context.getEndpoint(String.format("mina2:tcp://localhost:%1$s?textline=true&sync=true&timeout=500", getPort()));
Producer producer = endpoint.createProducer();
producer.start();
- Exchange exchange = producer.createExchange();
+ Exchange exchange = endpoint.createExchange();
exchange.getIn().setBody("Hello World");
try {
producer.process(exchange);
http://git-wip-us.apache.org/repos/asf/camel/blob/1d3e4a2d/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2NoResponseFromServerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2NoResponseFromServerTest.java b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2NoResponseFromServerTest.java
index 25b82dd..7870efc 100644
--- a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2NoResponseFromServerTest.java
+++ b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2NoResponseFromServerTest.java
@@ -73,7 +73,7 @@ public class Mina2NoResponseFromServerTest extends BaseMina2Test {
public void encode(IoSession ioSession, Object message, ProtocolEncoderOutput out)
throws Exception {
// close session instead of returning a reply
- ioSession.close(true);
+ ioSession.closeNow();
}
public void dispose(IoSession ioSession) throws Exception {
@@ -89,7 +89,7 @@ public class Mina2NoResponseFromServerTest extends BaseMina2Test {
public void decode(IoSession ioSession, IoBuffer in,
ProtocolDecoderOutput out) throws Exception {
// close session instead of returning a reply
- ioSession.close(true);
+ ioSession.closeNow();
}
public void finishDecode(IoSession ioSession, ProtocolDecoderOutput protocolDecoderOutput)
http://git-wip-us.apache.org/repos/asf/camel/blob/1d3e4a2d/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ProducerShutdownMockTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ProducerShutdownMockTest.java b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ProducerShutdownMockTest.java
index c36bade..ab110a0 100644
--- a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ProducerShutdownMockTest.java
+++ b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ProducerShutdownMockTest.java
@@ -27,9 +27,10 @@ import org.apache.camel.component.mock.MockEndpoint;
import org.apache.mina.transport.socket.SocketConnector;
import org.junit.Test;
-import static org.easymock.classextension.EasyMock.createMock;
-import static org.easymock.classextension.EasyMock.replay;
-import static org.easymock.classextension.EasyMock.verify;
+import static org.easymock.EasyMock.createMock;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+
/**
* Unit testing for using a MinaProducer that it can shutdown properly (CAMEL-395)
http://git-wip-us.apache.org/repos/asf/camel/blob/1d3e4a2d/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ReverseProtocolHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ReverseProtocolHandler.java b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ReverseProtocolHandler.java
index 04579fa..6267dc0 100644
--- a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ReverseProtocolHandler.java
+++ b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2ReverseProtocolHandler.java
@@ -30,7 +30,7 @@ public class Mina2ReverseProtocolHandler extends IoHandlerAdapter {
public void exceptionCaught(IoSession session, Throwable cause) {
cause.printStackTrace();
// Close connection when unexpected exception is caught.
- session.close(true);
+ session.closeNow();
}
public void messageReceived(IoSession session, Object message) {
http://git-wip-us.apache.org/repos/asf/camel/blob/1d3e4a2d/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2TransferExchangeOptionTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2TransferExchangeOptionTest.java b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2TransferExchangeOptionTest.java
index 282d471..74b5683 100644
--- a/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2TransferExchangeOptionTest.java
+++ b/components/camel-mina2/src/test/java/org/apache/camel/component/mina2/Mina2TransferExchangeOptionTest.java
@@ -53,7 +53,7 @@ public class Mina2TransferExchangeOptionTest extends BaseMina2Test {
private Exchange sendExchange(boolean setException) throws Exception {
Endpoint endpoint = context.getEndpoint(String.format("mina2:tcp://localhost:%1$s?sync=true&encoding=UTF-8&transferExchange=true", getPort()));
Producer producer = endpoint.createProducer();
- Exchange exchange = producer.createExchange();
+ Exchange exchange = endpoint.createExchange();
//Exchange exchange = endpoint.createExchange();
Message message = exchange.getIn();