This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch 2.2.X
in repository https://gitbox.apache.org/repos/asf/mina.git
The following commit(s) were added to refs/heads/2.2.X by this push:
new 33b97045f o Commented the LOG added in AbstractPollingIoProcessor,
they were interacting with a MDC test (which has tobe fixed) o Don't increase
the number of written messages when the message is a TLS one (DIRMINA-1146) o
Simplified the messageSent method in SslFilter o Cleaned up the SslFilterTest o
Added a test to check that messages sent are correctly counted when using TLS
(DIRMINA-1146)
33b97045f is described below
commit 33b97045f61b3b95c0681aeba0100e42c9b3f1b6
Author: emmanuel lecharny <[email protected]>
AuthorDate: Thu Sep 10 12:57:56 2026 +0200
o Commented the LOG added in AbstractPollingIoProcessor, they were
interacting with a MDC test (which has tobe fixed)
o Don't increase the number of written messages when the message is a
TLS one (DIRMINA-1146)
o Simplified the messageSent method in SslFilter
o Cleaned up the SslFilterTest
o Added a test to check that messages sent are correctly counted when
using TLS (DIRMINA-1146)
---
.../core/filterchain/DefaultIoFilterChain.java | 8 +-
.../core/polling/AbstractPollingIoProcessor.java | 6 +-
.../java/org/apache/mina/filter/ssl/SslFilter.java | 5 +-
.../ssl/SslFilterScheduledWriteMessagesTest.java | 330 +++++++++++++++++++++
.../mina/example/echoserver/ssl/SslFilterTest.java | 10 +-
5 files changed, 344 insertions(+), 15 deletions(-)
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..5b76e52bf 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.session.IoSession;
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,8 +905,11 @@ public class DefaultIoFilterChain implements IoFilterChain
{
}
}
- 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();
if (!s.isWriteSuspended()) {
diff --git
a/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java
b/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java
index 2cc5f2c70..c905daa99 100644
---
a/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java
+++
b/mina-core/src/main/java/org/apache/mina/core/polling/AbstractPollingIoProcessor.java
@@ -413,13 +413,13 @@ public abstract class AbstractPollingIoProcessor<S
extends AbstractIoSession> im
*/
@Override
public final void remove(S session) {
- LOG.debug( "Session {} has to be removed", session );
+ //LOG.debug( "Session {} has to be removed", session );
scheduleRemove(session);
startupProcessor();
}
private void scheduleRemove(S session) {
- LOG.debug( "Session {} scheduled to be removed", session );
+ //LOG.debug( "Session {} scheduled to be removed", session );
if (!removingSessions.contains(session)) {
removingSessions.add(session);
}
@@ -1137,7 +1137,7 @@ public abstract class AbstractPollingIoProcessor<S
extends AbstractIoSession> im
}
private boolean removeNow(S session) {
- LOG.debug( "RemoveNow requested for session {}", session );
+ //LOG.debug( "RemoveNow requested for session {}", session );
clearWriteRequestQueue(session);
try {
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..bcf645139 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
@@ -455,12 +455,11 @@ public class SslFilter extends IoFilterAdapter {
}
}
- 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..8e2365d83 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 @@ public class SslFilterTest {
*/
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 {
@@ -148,7 +148,7 @@ public class SslFilterTest {
private Socket getClientSocket(boolean ssl) throws Exception {
if (ssl) {
- SSLContext ctx = SSLContext.getInstance("TLS");
+ SSLContext ctx = SSLContext.getInstance("TLSv1.3");
ctx.init(null, trustManagers, null);
return ctx.getSocketFactory().createSocket("localhost", port);
}
@@ -172,10 +172,6 @@ public class SslFilterTest {
@Override
public void messageSent(IoSession session, Object message) throws
Exception {
sentMessages.add(message.toString());
-
- if (sentMessages.size() >= 2) {
- session.closeNow();
- }
}
}
