diff --git a/mina-core/src/main/java/org/apache/mina/core/filterchain/DefaultIoFilterChain.java b/mina-core/src/main/java/org/apache/mina/core/filterchain/DefaultIoFilterChain.java index 893079044..da74a2ac2 100644 --- a/mina-core/src/main/java/org/apache/mina/core/filterchain/DefaultIoFilterChain.java +++ b/mina-core/src/main/java/org/apache/mina/core/filterchain/DefaultIoFilterChain.java @@ -36,6 +36,7 @@ import org.apache.mina.core.write.WriteRequest; import org.apache.mina.core.write.WriteRequestQueue; import org.apache.mina.filter.FilterEvent; +import org.apache.mina.filter.ssl.EncryptedWriteRequest; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -904,7 +905,10 @@ public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest w } } - s.increaseScheduledWriteMessages(); + if (!(writeRequest instanceof EncryptedWriteRequest) || writeRequest.getOriginalRequest() != writeRequest) { + // do not increase the counter for encrypted SSL-related messages + s.increaseScheduledWriteMessages(); + } WriteRequestQueue writeRequestQueue = s.getWriteRequestQueue(); diff --git a/mina-core/src/main/java/org/apache/mina/filter/ssl/SslFilter.java b/mina-core/src/main/java/org/apache/mina/filter/ssl/SslFilter.java index 3238798c8..a39050f6c 100644 --- a/mina-core/src/main/java/org/apache/mina/filter/ssl/SslFilter.java +++ b/mina-core/src/main/java/org/apache/mina/filter/ssl/SslFilter.java @@ -454,13 +454,12 @@ public void messageSent(NextFilter next, IoSession session, WriteRequest request LOGGER.debug("CLIENT: Session {} ack {}", session, request); } } - - EncryptedWriteRequest encryptedWriteRequest = EncryptedWriteRequest.class.cast(request); + SslHandler sslHandler = getSslHandler(session); sslHandler.ack(next, request); - - if (encryptedWriteRequest.getOriginalRequest() != encryptedWriteRequest) { - next.messageSent(session, encryptedWriteRequest.getOriginalRequest()); + + if (request.getOriginalRequest() != request) { + next.messageSent(session, request.getOriginalRequest()); } } else { super.messageSent(next, session, request); diff --git a/mina-core/src/test/java/org/apache/mina/filter/ssl/SslFilterScheduledWriteMessagesTest.java b/mina-core/src/test/java/org/apache/mina/filter/ssl/SslFilterScheduledWriteMessagesTest.java new file mode 100644 index 000000000..e31e10c53 --- /dev/null +++ b/mina-core/src/test/java/org/apache/mina/filter/ssl/SslFilterScheduledWriteMessagesTest.java @@ -0,0 +1,330 @@ +/* + * 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.mina.filter.ssl; + +import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder; +import org.apache.mina.core.future.WriteFuture; +import org.apache.mina.core.service.IoHandler; +import org.apache.mina.core.service.IoHandlerAdapter; +import org.apache.mina.core.service.IoService; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.FilterEvent; +import org.apache.mina.filter.codec.ProtocolCodecFilter; +import org.apache.mina.filter.codec.textline.TextLineCodecFactory; +import org.apache.mina.transport.socket.nio.NioSocketAcceptor; +import org.apache.mina.transport.socket.nio.NioSocketConnector; +import org.apache.mina.util.AvailablePortFinder; +import org.junit.Before; +import org.junit.Test; + +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManagerFactory; +import java.net.InetSocketAddress; +import java.security.KeyStore; +import java.security.Security; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class SslFilterScheduledWriteMessagesTest { + + private static final String KEY_STORE_PATH = "keystore.jks"; + private static final String TRUST_STORE_PATH = "truststore.jks"; + private static final String[] ENABLED_PROTOCOLS = new String[] { "TLSv1.2" }; + private static final String KEY_MANAGER_FACTORY_ALGORITHM; + + static { + String algorithm = Security.getProperty("ssl.KeyManagerFactory.algorithm"); + + if (algorithm == null) { + algorithm = KeyManagerFactory.getDefaultAlgorithm(); + } + + KEY_MANAGER_FACTORY_ALGORITHM = algorithm; + } + + private CountDownLatch handshakeDone; + private CountDownLatch sessionsOpened; + private int port; + + @Before + public void setUp() { + handshakeDone = new CountDownLatch(2); + sessionsOpened = new CountDownLatch(2); + port = AvailablePortFinder.getNextAvailable(5555); + } + + @Test + public void shouldDecrementScheduledWriteMessages() throws Exception { + CountDownLatch handshakeDone = new CountDownLatch(0); + AcceptorIoHandler acceptorIoHandler = new AcceptorIoHandler(handshakeDone, sessionsOpened); + ConnectionIoHandler connectionIoHandler = new ConnectionIoHandler(handshakeDone, sessionsOpened); + + IoService acceptorService = startAcceptor(acceptorIoHandler); + + try { + IoService connectorService = startConnector(connectionIoHandler); + + try { + assertTrue(sessionsOpened.await(10L, TimeUnit.SECONDS)); + + IoSession acceptorSession = acceptorIoHandler.session; + IoSession connectorSession = connectionIoHandler.session; + + assertEquals(0, acceptorSession.getWrittenMessages()); + assertEquals(0, connectorSession.getWrittenMessages()); + assertEquals(0, acceptorSession.getScheduledWriteMessages()); + assertEquals(0, connectorSession.getScheduledWriteMessages()); + assertEquals(0, acceptorIoHandler.sentMessageCount); + assertEquals(0, connectionIoHandler.sentMessageCount); + + WriteFuture connectorWriteFuture = connectorSession.write("connector message"); + assertTrue(connectorWriteFuture.await(2L, TimeUnit.SECONDS)); + + Thread.sleep(1000L); + + assertEquals(0, acceptorSession.getWrittenMessages()); + assertEquals(1, connectorSession.getWrittenMessages()); + assertEquals(0, acceptorSession.getScheduledWriteMessages()); + assertEquals(0, connectorSession.getScheduledWriteMessages()); + assertEquals(0, acceptorIoHandler.sentMessageCount); + assertEquals(1, connectionIoHandler.sentMessageCount); + + WriteFuture acceptorWriteFuture = acceptorSession.write("acceptor message"); + assertTrue(acceptorWriteFuture.await(2L, TimeUnit.SECONDS)); + + Thread.sleep(1000L); + + assertEquals(1, acceptorSession.getWrittenMessages()); + assertEquals(1, connectorSession.getWrittenMessages()); + assertEquals(0, acceptorSession.getScheduledWriteMessages()); + assertEquals(0, connectorSession.getScheduledWriteMessages()); + assertEquals(1, acceptorIoHandler.sentMessageCount); + assertEquals(1, connectionIoHandler.sentMessageCount); + } finally { + connectorService.dispose(); + } + } finally { + acceptorService.dispose(); + } + } + + @Test + public void shouldDecrementScheduledWriteMessagesWithSsl() throws Exception { + SSLContext sslContext = createSSLContext(); + + AcceptorIoHandler acceptorIoHandler = new AcceptorIoHandler(handshakeDone, sessionsOpened); + ConnectionIoHandler connectionIoHandler = new ConnectionIoHandler(handshakeDone, sessionsOpened); + + IoService acceptorService = startSslAcceptor(sslContext, acceptorIoHandler); + + try { + IoService connectorService = startSslConnector(sslContext, connectionIoHandler); + + try { + assertTrue(handshakeDone.await(10L, TimeUnit.SECONDS)); + assertTrue(sessionsOpened.await(10L, TimeUnit.SECONDS)); + + IoSession acceptorSession = acceptorIoHandler.session; + IoSession connectorSession = connectionIoHandler.session; + + assertEquals(0, acceptorSession.getWrittenMessages()); + assertEquals(0, connectorSession.getWrittenMessages()); + assertEquals(0, acceptorSession.getScheduledWriteMessages()); + assertEquals(0, connectorSession.getScheduledWriteMessages()); + assertEquals(0, acceptorIoHandler.sentMessageCount); + assertEquals(0, connectionIoHandler.sentMessageCount); + + WriteFuture connectorWriteFuture = connectorSession.write("connector message"); + assertTrue(connectorWriteFuture.await(2L, TimeUnit.SECONDS)); + + Thread.sleep(1000L); + + assertEquals(0, acceptorSession.getWrittenMessages()); + assertEquals(1, connectorSession.getWrittenMessages()); + assertEquals(0, acceptorSession.getScheduledWriteMessages()); + assertEquals(0, connectorSession.getScheduledWriteMessages()); + assertEquals(0, acceptorIoHandler.sentMessageCount); + assertEquals(1, connectionIoHandler.sentMessageCount); + + WriteFuture acceptorWriteFuture = acceptorSession.write("acceptor message"); + assertTrue(acceptorWriteFuture.await(2L, TimeUnit.SECONDS)); + + Thread.sleep(1000L); + + assertEquals(1, acceptorSession.getWrittenMessages()); + assertEquals(1, connectorSession.getWrittenMessages()); + assertEquals(0, acceptorSession.getScheduledWriteMessages()); + assertEquals(0, connectorSession.getScheduledWriteMessages()); + assertEquals(1, acceptorIoHandler.sentMessageCount); + assertEquals(1, connectionIoHandler.sentMessageCount); + } finally { + connectorService.dispose(); + } + } finally { + acceptorService.dispose(); + } + } + + private IoService startAcceptor(IoHandler handler) throws Exception { + NioSocketAcceptor acceptor = new NioSocketAcceptor(); + acceptor.setReuseAddress(true); + + DefaultIoFilterChainBuilder filters = acceptor.getFilterChain(); + filters.addLast("text", new ProtocolCodecFilter(new TextLineCodecFactory())); + + acceptor.setHandler(handler); + acceptor.bind(new InetSocketAddress(port)); + + return acceptor; + } + + private IoService startSslAcceptor(SSLContext sslContext, IoHandler handler) throws Exception { + NioSocketAcceptor acceptor = new NioSocketAcceptor(); + acceptor.setReuseAddress(true); + + SslFilter sslFilter = new SslFilter(sslContext); + sslFilter.setEnabledProtocols(ENABLED_PROTOCOLS); + + DefaultIoFilterChainBuilder filters = acceptor.getFilterChain(); + filters.addLast("ssl", sslFilter); + filters.addLast("text", new ProtocolCodecFilter(new TextLineCodecFactory())); + + acceptor.setHandler(handler); + acceptor.bind(new InetSocketAddress(port)); + + return acceptor; + } + + private IoService startSslConnector(SSLContext sslContext, IoHandler handler) { + NioSocketConnector connector = new NioSocketConnector(); + + SslFilter sslFilter = new SslFilter(sslContext); + sslFilter.setEnabledProtocols(ENABLED_PROTOCOLS); + + DefaultIoFilterChainBuilder filters = connector.getFilterChain(); + filters.addLast("ssl", sslFilter); + filters.addLast("text", new ProtocolCodecFilter(new TextLineCodecFactory())); + + connector.setHandler(handler); + connector.connect(new InetSocketAddress("localhost", port)); + + return connector; + } + + private IoService startConnector(IoHandler handler) { + NioSocketConnector connector = new NioSocketConnector(); + + DefaultIoFilterChainBuilder filters = connector.getFilterChain(); + filters.addLast("text", new ProtocolCodecFilter(new TextLineCodecFactory())); + + connector.setHandler(handler); + connector.connect(new InetSocketAddress("localhost", port)); + + return connector; + } + + + private static SSLContext createSSLContext() throws Exception { + char[] password = "password".toCharArray(); + + KeyStore keyStore = KeyStore.getInstance("JKS"); + keyStore.load(SslIdentificationAlgorithmTest.class.getResourceAsStream(KEY_STORE_PATH), password); + + KeyManagerFactory kmf = KeyManagerFactory.getInstance(KEY_MANAGER_FACTORY_ALGORITHM); + kmf.init(keyStore, password); + + KeyStore trustStore = KeyStore.getInstance("JKS"); + trustStore.load(SslIdentificationAlgorithmTest.class.getResourceAsStream(TRUST_STORE_PATH), password); + + TrustManagerFactory tmf = TrustManagerFactory.getInstance(KEY_MANAGER_FACTORY_ALGORITHM); + tmf.init(trustStore); + + SSLContext sslContext = SSLContext.getInstance("TLSv1.2"); + sslContext.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null); + + return sslContext; + } + + private static final class AcceptorIoHandler extends IoHandlerAdapter { + + private final CountDownLatch handshakeDone; + private final CountDownLatch sessionsOpened; + private IoSession session; + private int sentMessageCount; + + public AcceptorIoHandler(CountDownLatch handshakeDone, CountDownLatch sessionsOpened) { + this.handshakeDone = handshakeDone; + this.sessionsOpened = sessionsOpened; + } + + @Override + public void sessionOpened(IoSession session) { + this.session = session; + sessionsOpened.countDown(); + } + + @Override + public void messageSent(IoSession session, Object message) { + sentMessageCount++; + } + + @Override + public void event(IoSession session, FilterEvent event) { + if (event == SslEvent.SECURED) { + handshakeDone.countDown(); + } + } + } + + private static final class ConnectionIoHandler extends IoHandlerAdapter { + + private final CountDownLatch handshakeDone; + private final CountDownLatch sessionsOpened; + private IoSession session; + private int sentMessageCount; + + public ConnectionIoHandler(CountDownLatch handshakeDone, CountDownLatch sessionsOpened) { + this.handshakeDone = handshakeDone; + this.sessionsOpened = sessionsOpened; + } + + @Override + public void sessionOpened(IoSession session) { + this.session = session; + sessionsOpened.countDown(); + } + + @Override + public void messageSent(IoSession session, Object message) { + sentMessageCount++; + } + + @Override + public void event(IoSession session, FilterEvent event) { + if (event == SslEvent.SECURED) { + handshakeDone.countDown(); + } + } + } +} diff --git a/mina-example/src/test/java/org/apache/mina/example/echoserver/ssl/SslFilterTest.java b/mina-example/src/test/java/org/apache/mina/example/echoserver/ssl/SslFilterTest.java index 60799aa63..3bc05e21d 100644 --- a/mina-example/src/test/java/org/apache/mina/example/echoserver/ssl/SslFilterTest.java +++ b/mina-example/src/test/java/org/apache/mina/example/echoserver/ssl/SslFilterTest.java @@ -130,14 +130,14 @@ private void testMessageSentIsCalled(boolean useSSL) throws Exception { */ socket.close(); + while (acceptor.getManagedSessions().size() != 0) { Thread.sleep(100); } - // System.out.println("handler: " + handler.sentMessages); assertEquals("handler should have sent 1 messages:", 1, handler.sentMessages.size()); + assertEquals("All scheduled write messages should be cleared", 0, acceptor.getScheduledWriteMessages()); assertTrue(handler.sentMessages.contains("test-1")); - //assertTrue(handler.sentMessages.contains("test-2")); } private int writeMessage(Socket socket, String message) throws Exception { @@ -172,10 +172,6 @@ public void messageReceived(IoSession session, Object message) throws Exception @Override public void messageSent(IoSession session, Object message) throws Exception { sentMessages.add(message.toString()); - - if (sentMessages.size() >= 2) { - session.closeNow(); - } } }