camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From anoordo...@apache.org
Subject [1/3] camel git commit: CAMEL-10024: sync on close and deprecation
Date Wed, 22 Jun 2016 12:35:03 GMT
Repository: camel
Updated Branches:
  refs/heads/master 8f2bc1514 -> 76544116f


CAMEL-10024: sync on close and deprecation


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/fe41b1bb
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/fe41b1bb
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/fe41b1bb

Branch: refs/heads/master
Commit: fe41b1bb9ac743214a6f6baba393a251e55037fc
Parents: e784761
Author: Arno Noordover <anoordover@users.noreply.github.com>
Authored: Fri Jun 17 16:47:14 2016 +0200
Committer: Arno Noordover <anoordover@users.noreply.github.com>
Committed: Fri Jun 17 16:47:14 2016 +0200

----------------------------------------------------------------------
 .../camel/component/mina2/Mina2Consumer.java    |  6 +-
 .../camel/component/mina2/Mina2Producer.java    | 36 +++++++---
 ...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 +-
 10 files changed, 109 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/fe41b1bb/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 c2a7c50..a916bab 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/fe41b1bb/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 fdd02dc..4337075 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,6 +65,7 @@ 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 boolean lazySessionCreation;
@@ -76,6 +77,7 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware
{
     private Mina2Configuration configuration;
     private IoSessionConfig connectorConfig;
     private ExecutorService workerPool;
+    private CountDownLatch closeLatch;
 
     public Mina2Producer(Mina2Endpoint endpoint) throws Exception {
         super(endpoint);
@@ -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
@@ -146,7 +150,6 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware
{
             // only initialize latch if we should get a response
             latch = new CountDownLatch(1);
             // reset handler if we expect a response
-            ResponseHandler handler = (ResponseHandler) session.getHandler();
             handler.reset();
         }
 
@@ -171,7 +174,6 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware
{
             }
 
             // 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 {
@@ -502,6 +511,7 @@ public class Mina2Producer extends DefaultProducer implements ServicePoolAware
{
                 // and could not return a response. We should count down to stop waiting
for a response
                 countDown();
             }
+            closeLatch.countDown();
         }
 
         @Override
@@ -512,7 +522,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/fe41b1bb/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 ba11d11..3f7bb47 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
@@ -86,7 +86,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/fe41b1bb/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/fe41b1bb/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 63328d5..17a120f 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
@@ -171,7 +171,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/fe41b1bb/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/fe41b1bb/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/fe41b1bb/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/fe41b1bb/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/fe41b1bb/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();


Mime
View raw message