You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mina.apache.org by el...@apache.org on 2019/03/25 05:42:00 UTC
[mina] 01/01: Pushed my current changes related to the filterWrite
refactoring
This is an automated email from the ASF dual-hosted git repository.
elecharny pushed a commit to branch mina-write-request
in repository https://gitbox.apache.org/repos/asf/mina.git
commit 2d08d530961597b9f21dff861725f08e73fb9291
Author: emmanuel lecharny <el...@apache.org>
AuthorDate: Mon Mar 25 06:41:26 2019 +0100
Pushed my current changes related to the filterWrite refactoring
---
.../apache/mina/core/file/FilenameFileRegion.java | 1 -
.../core/filterchain/DefaultIoFilterChain.java | 29 +---
.../mina/core/filterchain/IoFilterEvent.java | 45 +++---
.../core/polling/AbstractPollingIoProcessor.java | 22 +--
.../java/org/apache/mina/core/session/IoEvent.java | 37 +++--
.../org/apache/mina/core/session/IoEventType.java | 6 +
.../mina/core/write/DefaultWriteRequest.java | 48 +++++-
.../org/apache/mina/core/write/WriteRequest.java | 13 ++
.../mina/core/write/WriteRequestWrapper.java | 101 ------------
.../mina/filter/codec/ProtocolCodecFilter.java | 32 +---
.../mina/filter/keepalive/KeepAliveFilter.java | 10 +-
.../apache/mina/filter/logging/LoggingFilter.java | 2 +-
.../java/org/apache/mina/filter/ssl/SslFilter.java | 27 +---
.../mina/filter/statistic/ProfilerTimerFilter.java | 178 ++++++++++-----------
.../filter/stream/AbstractStreamWriteFilter.java | 1 -
.../mina/filter/stream/FileRegionWriteFilter.java | 2 +-
.../apache/mina/filter/util/CommonEventFilter.java | 47 ++++--
.../mina/filter/util/WriteRequestFilter.java | 92 -----------
.../transport/socket/nio/NioDatagramAcceptor.java | 2 -
.../mina/transport/vmpipe/VmPipeFilterChain.java | 105 ++++++++----
.../executor/ExecutorFilterRegressionTest.java | 2 +-
.../mina/filter/keepalive/KeepAliveFilterTest.java | 62 +++----
.../apache/mina/transport/AbstractBindTest.java | 4 +-
mina-core/src/test/resources/log4j.properties | 2 +-
mina-example/pom.xml | 7 +
.../mina/example/chat/ChatProtocolHandler.java | 5 +
.../java/org/apache/mina/example/chat/Main.java | 6 +-
.../example/chat/client/ChatClientSupport.java | 8 +-
.../org/apache/mina/example/echoserver/Main.java | 4 +
.../mina/filter/compression/CompressionFilter.java | 33 +++-
.../org/apache/mina/filter/compression/Zlib.java | 2 +-
.../filter/compression/CompressionFilterTest.java | 1 +
.../apache/mina/filter/compression/ZlibTest.java | 3 +
33 files changed, 416 insertions(+), 523 deletions(-)
diff --git a/mina-core/src/main/java/org/apache/mina/core/file/FilenameFileRegion.java b/mina-core/src/main/java/org/apache/mina/core/file/FilenameFileRegion.java
index 31c72cb..ce8404b 100644
--- a/mina-core/src/main/java/org/apache/mina/core/file/FilenameFileRegion.java
+++ b/mina-core/src/main/java/org/apache/mina/core/file/FilenameFileRegion.java
@@ -20,7 +20,6 @@
package org.apache.mina.core.file;
import java.io.File;
-import java.io.IOException;
import java.nio.channels.FileChannel;
/**
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 51d1ee6..ddaf36c 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
@@ -742,7 +742,7 @@ public class DefaultIoFilterChain implements IoFilterChain {
* {@inheritDoc}
*/
@Override
-public void fireFilterWrite(WriteRequest writeRequest) {
+ public void fireFilterWrite(WriteRequest writeRequest) {
callPreviousFilterWrite(tail, session, writeRequest);
}
@@ -888,7 +888,7 @@ public void fireFilterWrite(WriteRequest writeRequest) {
private class HeadFilter extends IoFilterAdapter {
@SuppressWarnings("unchecked")
@Override
- public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
+ public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
AbstractIoSession s = (AbstractIoSession) session;
// Maintain counters.
@@ -897,7 +897,6 @@ public void fireFilterWrite(WriteRequest writeRequest) {
// I/O processor implementation will call buffer.reset()
// it after the write operation is finished, because
// the buffer will be specified with messageSent event.
- buffer.mark();
int remaining = buffer.remaining();
if (remaining > 0) {
@@ -905,9 +904,7 @@ public void fireFilterWrite(WriteRequest writeRequest) {
}
}
- if (!writeRequest.isEncoded()) {
- s.increaseScheduledWriteMessages();
- }
+ s.increaseScheduledWriteMessages();
WriteRequestQueue writeRequestQueue = s.getWriteRequestQueue();
@@ -924,7 +921,6 @@ public void fireFilterWrite(WriteRequest writeRequest) {
}
}
-
@SuppressWarnings("unchecked")
@Override
public void filterClose(NextFilter nextFilter, IoSession session) throws Exception {
@@ -1005,7 +1001,7 @@ public void fireFilterWrite(WriteRequest writeRequest) {
public void messageReceived(NextFilter nextFilter, IoSession session, Object message) throws Exception {
AbstractIoSession s = (AbstractIoSession) session;
- if (!(message instanceof IoBuffer) || !((IoBuffer) message).hasRemaining()) {
+ if (message instanceof IoBuffer && !((IoBuffer) message).hasRemaining()) {
s.increaseReadMessages(System.currentTimeMillis());
}
@@ -1026,25 +1022,16 @@ public void fireFilterWrite(WriteRequest writeRequest) {
@Override
public void messageSent(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
- ((AbstractIoSession) session).increaseWrittenMessages(writeRequest, System.currentTimeMillis());
+ long now = System.currentTimeMillis();
+ ((AbstractIoSession) session).increaseWrittenMessages(writeRequest, now);
// Update the statistics
if (session.getService() instanceof AbstractIoService) {
- ((AbstractIoService) session.getService()).getStatistics().updateThroughput(System.currentTimeMillis());
+ ((AbstractIoService) session.getService()).getStatistics().updateThroughput(now);
}
// Propagate the message
- session.getHandler().messageSent(session, writeRequest.getMessage());
- }
-
- @Override
- public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
- nextFilter.filterWrite(session, writeRequest);
- }
-
- @Override
- public void filterClose(NextFilter nextFilter, IoSession session) throws Exception {
- nextFilter.filterClose(session);
+ session.getHandler().messageSent(session, writeRequest.getOriginalMessage());
}
@Override
diff --git a/mina-core/src/main/java/org/apache/mina/core/filterchain/IoFilterEvent.java b/mina-core/src/main/java/org/apache/mina/core/filterchain/IoFilterEvent.java
index 7725619..52b0878 100644
--- a/mina-core/src/main/java/org/apache/mina/core/filterchain/IoFilterEvent.java
+++ b/mina-core/src/main/java/org/apache/mina/core/filterchain/IoFilterEvent.java
@@ -25,6 +25,7 @@ import org.apache.mina.core.session.IoEvent;
import org.apache.mina.core.session.IoEventType;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.core.write.WriteRequest;
+import org.apache.mina.filter.FilterEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -82,6 +83,23 @@ public class IoFilterEvent extends IoEvent {
}
switch (type) {
+ case CLOSE:
+ nextFilter.filterClose(session);
+ break;
+
+ case EVENT:
+ nextFilter.event(session, (FilterEvent)getParameter());
+ break;
+
+ case EXCEPTION_CAUGHT:
+ Throwable throwable = (Throwable) getParameter();
+ nextFilter.exceptionCaught(session, throwable);
+ break;
+
+ case INPUT_CLOSED:
+ nextFilter.inputClosed(session);
+ break;
+
case MESSAGE_RECEIVED:
Object parameter = getParameter();
nextFilter.messageReceived(session, parameter);
@@ -92,18 +110,12 @@ public class IoFilterEvent extends IoEvent {
nextFilter.messageSent(session, writeRequest);
break;
- case WRITE:
- writeRequest = (WriteRequest) getParameter();
- nextFilter.filterWrite(session, writeRequest);
- break;
-
- case CLOSE:
- nextFilter.filterClose(session);
+ case SESSION_CLOSED:
+ nextFilter.sessionClosed(session);
break;
-
- case EXCEPTION_CAUGHT:
- Throwable throwable = (Throwable) getParameter();
- nextFilter.exceptionCaught(session, throwable);
+
+ case SESSION_CREATED:
+ nextFilter.sessionCreated(session);
break;
case SESSION_IDLE:
@@ -114,14 +126,11 @@ public class IoFilterEvent extends IoEvent {
nextFilter.sessionOpened(session);
break;
- case SESSION_CREATED:
- nextFilter.sessionCreated(session);
- break;
-
- case SESSION_CLOSED:
- nextFilter.sessionClosed(session);
+ case WRITE:
+ writeRequest = (WriteRequest) getParameter();
+ nextFilter.filterWrite(session, writeRequest);
break;
-
+
default:
throw new IllegalArgumentException("Unknown event type: " + type);
}
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 5fe4ca1..4a75f41 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
@@ -1117,27 +1117,9 @@ public abstract class AbstractPollingIoProcessor<S extends AbstractIoSession> im
session.increaseWrittenBytes(localWrittenBytes, currentTime);
- // Now, forward the original message
+ // Now, forward the original message if ity has been fully sent
if (!buf.hasRemaining() || (!hasFragmentation && (localWrittenBytes != 0))) {
- WriteRequest originalRequest = req.getOriginalRequest();
-
- if (originalRequest != null) {
- Object originalMessage = originalRequest.getMessage();
-
- if (originalMessage instanceof IoBuffer) {
- buf = (IoBuffer) originalMessage;
-
- int pos = buf.position();
- buf.reset();
- this.fireMessageSent(session, req);
- // And set it back to its position
- buf.position(pos);
- } else {
- this.fireMessageSent(session, req);
- }
- } else {
- this.fireMessageSent(session, req);
- }
+ this.fireMessageSent(session, req);
}
} else {
this.fireMessageSent(session, req);
diff --git a/mina-core/src/main/java/org/apache/mina/core/session/IoEvent.java b/mina-core/src/main/java/org/apache/mina/core/session/IoEvent.java
index b3da2fc..3c923f7 100644
--- a/mina-core/src/main/java/org/apache/mina/core/session/IoEvent.java
+++ b/mina-core/src/main/java/org/apache/mina/core/session/IoEvent.java
@@ -20,6 +20,7 @@
package org.apache.mina.core.session;
import org.apache.mina.core.write.WriteRequest;
+import org.apache.mina.filter.FilterEvent;
/**
* An I/O event or an I/O request that MINA provides.
@@ -93,6 +94,22 @@ public class IoEvent implements Runnable {
*/
public void fire() {
switch ( type ) {
+ case CLOSE:
+ session.getFilterChain().fireFilterClose();
+ break;
+
+ case EVENT:
+ session.getFilterChain().fireEvent((FilterEvent)getParameter());
+ break;
+
+ case EXCEPTION_CAUGHT:
+ session.getFilterChain().fireExceptionCaught((Throwable) getParameter());
+ break;
+
+ case INPUT_CLOSED:
+ session.getFilterChain().fireInputClosed();
+ break;
+
case MESSAGE_RECEIVED:
session.getFilterChain().fireMessageReceived(getParameter());
break;
@@ -101,16 +118,12 @@ public class IoEvent implements Runnable {
session.getFilterChain().fireMessageSent((WriteRequest) getParameter());
break;
- case WRITE:
- session.getFilterChain().fireFilterWrite((WriteRequest) getParameter());
- break;
-
- case CLOSE:
- session.getFilterChain().fireFilterClose();
+ case SESSION_CLOSED:
+ session.getFilterChain().fireSessionClosed();
break;
- case EXCEPTION_CAUGHT:
- session.getFilterChain().fireExceptionCaught((Throwable) getParameter());
+ case SESSION_CREATED:
+ session.getFilterChain().fireSessionCreated();
break;
case SESSION_IDLE:
@@ -121,12 +134,8 @@ public class IoEvent implements Runnable {
session.getFilterChain().fireSessionOpened();
break;
- case SESSION_CREATED:
- session.getFilterChain().fireSessionCreated();
- break;
-
- case SESSION_CLOSED:
- session.getFilterChain().fireSessionClosed();
+ case WRITE:
+ session.getFilterChain().fireFilterWrite((WriteRequest) getParameter());
break;
default:
diff --git a/mina-core/src/main/java/org/apache/mina/core/session/IoEventType.java b/mina-core/src/main/java/org/apache/mina/core/session/IoEventType.java
index 44865fa..6b45582 100644
--- a/mina-core/src/main/java/org/apache/mina/core/session/IoEventType.java
+++ b/mina-core/src/main/java/org/apache/mina/core/session/IoEventType.java
@@ -53,4 +53,10 @@ public enum IoEventType {
/** A close has occured */
CLOSE,
+
+ /** The Input part of the socket has been closed */
+ INPUT_CLOSED,
+
+ /** A generic event has been generated */
+ EVENT
}
diff --git a/mina-core/src/main/java/org/apache/mina/core/write/DefaultWriteRequest.java b/mina-core/src/main/java/org/apache/mina/core/write/DefaultWriteRequest.java
index 1d1e5fb..91c1330 100644
--- a/mina-core/src/main/java/org/apache/mina/core/write/DefaultWriteRequest.java
+++ b/mina-core/src/main/java/org/apache/mina/core/write/DefaultWriteRequest.java
@@ -22,6 +22,7 @@ package org.apache.mina.core.write;
import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;
+import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.future.IoFutureListener;
import org.apache.mina.core.future.WriteFuture;
import org.apache.mina.core.session.IoSession;
@@ -166,10 +167,19 @@ public class DefaultWriteRequest implements WriteRequest {
}
};
- private final Object message;
+ /**
+ * The original message as it was written by the IoHandler. It will be sent back
+ * in the messageSent event
+ */
+ private final Object originalMessage;
+ /** The message that will ultimately be written to the remote peer */
+ private Object message;
+
+ /** The associated Future */
private final WriteFuture future;
+ /** The peer destination (useless ???) */
private final SocketAddress destination;
/**
@@ -211,6 +221,15 @@ public class DefaultWriteRequest implements WriteRequest {
}
this.message = message;
+
+ if (message instanceof IoBuffer) {
+ // duplicate it, so that any modification made on it
+ // won't change the original message
+ this.originalMessage = ((IoBuffer)message).duplicate();
+ } else {
+ originalMessage = message;
+ }
+
this.future = future;
this.destination = destination;
}
@@ -235,6 +254,26 @@ public class DefaultWriteRequest implements WriteRequest {
* {@inheritDoc}
*/
@Override
+ public void setMessage(Object modifiedMessage) {
+ message = modifiedMessage;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Object getOriginalMessage() {
+ if (originalMessage != null) {
+ return originalMessage;
+ } else {
+ return message;
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
public WriteRequest getOriginalRequest() {
return this;
}
@@ -258,10 +297,9 @@ public class DefaultWriteRequest implements WriteRequest {
if (message.getClass().getName().equals(Object.class.getName())) {
sb.append("CLOSE_REQUEST");
} else {
- if (getDestination() == null) {
- sb.append(message);
- } else {
- sb.append(message);
+ sb.append(originalMessage);
+
+ if (getDestination() != null) {
sb.append(" => ");
sb.append(getDestination());
}
diff --git a/mina-core/src/main/java/org/apache/mina/core/write/WriteRequest.java b/mina-core/src/main/java/org/apache/mina/core/write/WriteRequest.java
index 9ee5c55..861cea6 100644
--- a/mina-core/src/main/java/org/apache/mina/core/write/WriteRequest.java
+++ b/mina-core/src/main/java/org/apache/mina/core/write/WriteRequest.java
@@ -21,6 +21,7 @@ package org.apache.mina.core.write;
import java.net.SocketAddress;
+import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.filterchain.IoFilter;
import org.apache.mina.core.future.WriteFuture;
import org.apache.mina.core.session.IoSession;
@@ -48,6 +49,12 @@ public interface WriteRequest {
Object getMessage();
/**
+ * Set the modified message after it has been processed by a filter.
+ * @param modifiedMessage The modified message
+ */
+ void setMessage(Object modifiedMessage);
+
+ /**
* Returns the destination of this write request.
*
* @return <tt>null</tt> for the default destination
@@ -60,4 +67,10 @@ public interface WriteRequest {
* @return true if the message has already been encoded
*/
boolean isEncoded();
+
+ /**
+ * @return the original message which was sent to the session, before
+ * any filter transformation.
+ */
+ Object getOriginalMessage();
}
\ No newline at end of file
diff --git a/mina-core/src/main/java/org/apache/mina/core/write/WriteRequestWrapper.java b/mina-core/src/main/java/org/apache/mina/core/write/WriteRequestWrapper.java
deleted file mode 100644
index 0abea69..0000000
--- a/mina-core/src/main/java/org/apache/mina/core/write/WriteRequestWrapper.java
+++ /dev/null
@@ -1,101 +0,0 @@
-/*
- * 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.core.write;
-
-import java.net.SocketAddress;
-
-import org.apache.mina.core.future.WriteFuture;
-
-/**
- * A wrapper for an existing {@link WriteRequest}.
- *
- * @author <a href="http://mina.apache.org">Apache MINA Project</a>
- */
-public class WriteRequestWrapper implements WriteRequest {
-
- private final WriteRequest parentRequest;
-
- /**
- * Creates a new instance that wraps the specified request.
- *
- * @param parentRequest The parent's request
- */
- public WriteRequestWrapper(WriteRequest parentRequest) {
- if (parentRequest == null) {
- throw new IllegalArgumentException("parentRequest");
- }
- this.parentRequest = parentRequest;
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public SocketAddress getDestination() {
- return parentRequest.getDestination();
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public WriteFuture getFuture() {
- return parentRequest.getFuture();
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public Object getMessage() {
- return parentRequest.getMessage();
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public WriteRequest getOriginalRequest() {
- return parentRequest.getOriginalRequest();
- }
-
- /**
- * @return the wrapped request object.
- */
- public WriteRequest getParentRequest() {
- return parentRequest;
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public String toString() {
- return "WR Wrapper" + parentRequest.toString();
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public boolean isEncoded() {
- return false;
- }
-}
diff --git a/mina-core/src/main/java/org/apache/mina/filter/codec/ProtocolCodecFilter.java b/mina-core/src/main/java/org/apache/mina/filter/codec/ProtocolCodecFilter.java
index 844c39d..628e4f2 100644
--- a/mina-core/src/main/java/org/apache/mina/filter/codec/ProtocolCodecFilter.java
+++ b/mina-core/src/main/java/org/apache/mina/filter/codec/ProtocolCodecFilter.java
@@ -35,7 +35,6 @@ import org.apache.mina.core.session.IoSession;
import org.apache.mina.core.write.DefaultWriteRequest;
import org.apache.mina.core.write.NothingWrittenException;
import org.apache.mina.core.write.WriteRequest;
-import org.apache.mina.core.write.WriteRequestWrapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -293,12 +292,7 @@ public class ProtocolCodecFilter extends IoFilterAdapter {
return;
}
- if (writeRequest instanceof MessageWriteRequest) {
- MessageWriteRequest wrappedRequest = (MessageWriteRequest) writeRequest;
- nextFilter.messageSent(session, wrappedRequest.getParentRequest());
- } else {
- nextFilter.messageSent(session, writeRequest);
- }
+ nextFilter.messageSent(session, writeRequest);
}
/**
@@ -341,15 +335,11 @@ public class ProtocolCodecFilter extends IoFilterAdapter {
// Flush only when the buffer has remaining.
if (!(encodedMessage instanceof IoBuffer) || ((IoBuffer) encodedMessage).hasRemaining()) {
- SocketAddress destination = writeRequest.getDestination();
- WriteRequest encodedWriteRequest = new EncodedWriteRequest(encodedMessage, null, destination);
+ writeRequest.setMessage(encodedMessage);
- nextFilter.filterWrite(session, encodedWriteRequest);
+ nextFilter.filterWrite(session, writeRequest);
}
}
-
- // Call the next filter
- nextFilter.filterWrite(session, new MessageWriteRequest(writeRequest));
} catch (Exception e) {
ProtocolEncoderException pee;
@@ -407,22 +397,6 @@ public class ProtocolCodecFilter extends IoFilterAdapter {
}
}
- private static class MessageWriteRequest extends WriteRequestWrapper {
- public MessageWriteRequest(WriteRequest writeRequest) {
- super(writeRequest);
- }
-
- @Override
- public Object getMessage() {
- return EMPTY_BUFFER;
- }
-
- @Override
- public String toString() {
- return "MessageWriteRequest, parent : " + super.toString();
- }
- }
-
private static class ProtocolDecoderOutputImpl extends AbstractProtocolDecoderOutput {
public ProtocolDecoderOutputImpl() {
// Do nothing
diff --git a/mina-core/src/main/java/org/apache/mina/filter/keepalive/KeepAliveFilter.java b/mina-core/src/main/java/org/apache/mina/filter/keepalive/KeepAliveFilter.java
index d7fdbbc..80359d0 100644
--- a/mina-core/src/main/java/org/apache/mina/filter/keepalive/KeepAliveFilter.java
+++ b/mina-core/src/main/java/org/apache/mina/filter/keepalive/KeepAliveFilter.java
@@ -19,6 +19,7 @@
*/
package org.apache.mina.filter.keepalive;
+import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.filterchain.IoFilter;
import org.apache.mina.core.filterchain.IoFilterAdapter;
import org.apache.mina.core.filterchain.IoFilterChain;
@@ -420,7 +421,14 @@ public class KeepAliveFilter extends IoFilterAdapter {
*/
@Override
public void messageSent(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
- Object message = writeRequest.getMessage();
+ Object message = writeRequest.getOriginalMessage();
+
+ if (message == null)
+ {
+ if (writeRequest.getMessage() instanceof IoBuffer) {
+ message = ((IoBuffer)writeRequest.getMessage()).duplicate().flip();
+ }
+ }
if (!isKeepAliveMessage(session, message)) {
nextFilter.messageSent(session, writeRequest);
diff --git a/mina-core/src/main/java/org/apache/mina/filter/logging/LoggingFilter.java b/mina-core/src/main/java/org/apache/mina/filter/logging/LoggingFilter.java
index 5cc7aa1..99e43c9 100644
--- a/mina-core/src/main/java/org/apache/mina/filter/logging/LoggingFilter.java
+++ b/mina-core/src/main/java/org/apache/mina/filter/logging/LoggingFilter.java
@@ -210,7 +210,7 @@ public class LoggingFilter extends IoFilterAdapter {
@Override
public void messageSent(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
- log(messageSentLevel, "SENT: {}", writeRequest.getOriginalRequest().getMessage());
+ log(messageSentLevel, "SENT: {}", writeRequest.getOriginalMessage());
nextFilter.messageSent(session, writeRequest);
}
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 bfb960b..7e8fa22 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
@@ -42,7 +42,6 @@ import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.session.AttributeKey;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.core.write.WriteRequest;
-import org.apache.mina.core.write.WriteRequestWrapper;
import org.apache.mina.core.write.WriteToClosedSessionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -555,12 +554,7 @@ public class SslFilter extends IoFilterAdapter {
@Override
public void messageSent(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) {
- if (writeRequest instanceof EncryptedWriteRequest) {
- EncryptedWriteRequest wrappedRequest = (EncryptedWriteRequest) writeRequest;
- nextFilter.messageSent(session, wrappedRequest.getParentRequest());
- } else {
- // ignore extra buffers used for handshaking
- }
+ nextFilter.messageSent(session, writeRequest.getOriginalRequest());
}
@Override
@@ -651,11 +645,10 @@ public class SslFilter extends IoFilterAdapter {
sslHandler.scheduleFilterWrite(nextFilter, writeRequest);
} else if (sslHandler.isHandshakeComplete()) {
// SSL encrypt
- buf.mark();
sslHandler.encrypt(buf.buf());
IoBuffer encryptedBuffer = sslHandler.fetchOutNetBuffer();
- sslHandler.scheduleFilterWrite(nextFilter, new EncryptedWriteRequest(writeRequest,
- encryptedBuffer));
+ writeRequest.setMessage( encryptedBuffer );
+ sslHandler.scheduleFilterWrite(nextFilter, writeRequest);
} else {
if (session.isConnected()) {
// Handshake not complete yet.
@@ -850,18 +843,4 @@ public class SslFilter extends IoFilterAdapter {
return name;
}
}
-
- private static class EncryptedWriteRequest extends WriteRequestWrapper {
- private final IoBuffer encryptedMessage;
-
- private EncryptedWriteRequest(WriteRequest writeRequest, IoBuffer encryptedMessage) {
- super(writeRequest);
- this.encryptedMessage = encryptedMessage;
- }
-
- @Override
- public Object getMessage() {
- return encryptedMessage;
- }
- }
}
diff --git a/mina-core/src/main/java/org/apache/mina/filter/statistic/ProfilerTimerFilter.java b/mina-core/src/main/java/org/apache/mina/filter/statistic/ProfilerTimerFilter.java
index c1c2256..dd5815a 100644
--- a/mina-core/src/main/java/org/apache/mina/filter/statistic/ProfilerTimerFilter.java
+++ b/mina-core/src/main/java/org/apache/mina/filter/statistic/ProfilerTimerFilter.java
@@ -157,25 +157,25 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
messageSentTimerWorker = new TimerWorker();
profileMessageSent = true;
break;
+
+ case SESSION_CLOSED:
+ sessionClosedTimerWorker = new TimerWorker();
+ profileSessionClosed = true;
+ break;
case SESSION_CREATED:
sessionCreatedTimerWorker = new TimerWorker();
profileSessionCreated = true;
break;
-
- case SESSION_OPENED:
- sessionOpenedTimerWorker = new TimerWorker();
- profileSessionOpened = true;
- break;
-
+
case SESSION_IDLE:
sessionIdleTimerWorker = new TimerWorker();
profileSessionIdle = true;
break;
- case SESSION_CLOSED:
- sessionClosedTimerWorker = new TimerWorker();
- profileSessionClosed = true;
+ case SESSION_OPENED:
+ sessionOpenedTimerWorker = new TimerWorker();
+ profileSessionOpened = true;
break;
default :
@@ -217,7 +217,16 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
}
return;
+
+ case SESSION_CLOSED:
+ profileSessionClosed = true;
+ if (sessionClosedTimerWorker == null) {
+ sessionClosedTimerWorker = new TimerWorker();
+ }
+
+ return;
+
case SESSION_CREATED:
profileSessionCreated = true;
@@ -226,16 +235,7 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
}
return;
-
- case SESSION_OPENED:
- profileSessionOpened = true;
-
- if (sessionOpenedTimerWorker == null) {
- sessionOpenedTimerWorker = new TimerWorker();
- }
-
- return;
-
+
case SESSION_IDLE:
profileSessionIdle = true;
@@ -245,11 +245,11 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
return;
- case SESSION_CLOSED:
- profileSessionClosed = true;
+ case SESSION_OPENED:
+ profileSessionOpened = true;
- if (sessionClosedTimerWorker == null) {
- sessionClosedTimerWorker = new TimerWorker();
+ if (sessionOpenedTimerWorker == null) {
+ sessionOpenedTimerWorker = new TimerWorker();
}
return;
@@ -273,21 +273,21 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
case MESSAGE_SENT:
profileMessageSent = false;
return;
+
+ case SESSION_CLOSED:
+ profileSessionClosed = false;
+ return;
case SESSION_CREATED:
profileSessionCreated = false;
return;
-
- case SESSION_OPENED:
- profileSessionOpened = false;
- return;
-
+
case SESSION_IDLE:
profileSessionIdle = false;
return;
- case SESSION_CLOSED:
- profileSessionClosed = false;
+ case SESSION_OPENED:
+ profileSessionOpened = false;
return;
default:
@@ -509,21 +509,21 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
}
break;
-
- case SESSION_CREATED:
- if (profileSessionCreated) {
- return sessionCreatedTimerWorker.getAverage();
+
+ case SESSION_CLOSED:
+ if (profileSessionClosed) {
+ return sessionClosedTimerWorker.getAverage();
}
break;
- case SESSION_OPENED:
- if (profileSessionOpened) {
- return sessionOpenedTimerWorker.getAverage();
+ case SESSION_CREATED:
+ if (profileSessionCreated) {
+ return sessionCreatedTimerWorker.getAverage();
}
break;
-
+
case SESSION_IDLE:
if (profileSessionIdle) {
return sessionIdleTimerWorker.getAverage();
@@ -531,9 +531,9 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
break;
- case SESSION_CLOSED:
- if (profileSessionClosed) {
- return sessionClosedTimerWorker.getAverage();
+ case SESSION_OPENED:
+ if (profileSessionOpened) {
+ return sessionOpenedTimerWorker.getAverage();
}
break;
@@ -569,6 +569,13 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
}
break;
+
+ case SESSION_CLOSED:
+ if (profileSessionClosed) {
+ return sessionClosedTimerWorker.getCallsNumber();
+ }
+
+ break;
case SESSION_CREATED:
if (profileSessionCreated) {
@@ -576,14 +583,7 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
}
break;
-
- case SESSION_OPENED:
- if (profileSessionOpened) {
- return sessionOpenedTimerWorker.getCallsNumber();
- }
-
- break;
-
+
case SESSION_IDLE:
if (profileSessionIdle) {
return sessionIdleTimerWorker.getCallsNumber();
@@ -591,13 +591,13 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
break;
- case SESSION_CLOSED:
- if (profileSessionClosed) {
- return sessionClosedTimerWorker.getCallsNumber();
+ case SESSION_OPENED:
+ if (profileSessionOpened) {
+ return sessionOpenedTimerWorker.getCallsNumber();
}
break;
-
+
default:
break;
}
@@ -629,31 +629,31 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
}
break;
-
- case SESSION_CREATED:
- if (profileSessionCreated) {
- return sessionCreatedTimerWorker.getTotal();
+
+ case SESSION_CLOSED:
+ if (profileSessionClosed) {
+ return sessionClosedTimerWorker.getTotal();
}
break;
- case SESSION_OPENED:
- if (profileSessionOpened) {
- return sessionOpenedTimerWorker.getTotal();
+ case SESSION_CREATED:
+ if (profileSessionCreated) {
+ return sessionCreatedTimerWorker.getTotal();
}
break;
-
+
case SESSION_IDLE:
if (profileSessionIdle) {
return sessionIdleTimerWorker.getTotal();
}
break;
-
- case SESSION_CLOSED:
- if (profileSessionClosed) {
- return sessionClosedTimerWorker.getTotal();
+
+ case SESSION_OPENED:
+ if (profileSessionOpened) {
+ return sessionOpenedTimerWorker.getTotal();
}
break;
@@ -689,21 +689,21 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
}
break;
-
+
+ case SESSION_CLOSED:
+ if (profileSessionClosed) {
+ return sessionClosedTimerWorker.getMinimum();
+ }
+
+ break;
+
case SESSION_CREATED:
if (profileSessionCreated) {
return sessionCreatedTimerWorker.getMinimum();
}
break;
-
- case SESSION_OPENED:
- if (profileSessionOpened) {
- return sessionOpenedTimerWorker.getMinimum();
- }
-
- break;
-
+
case SESSION_IDLE:
if (profileSessionIdle) {
return sessionIdleTimerWorker.getMinimum();
@@ -711,11 +711,11 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
break;
- case SESSION_CLOSED:
- if (profileSessionClosed) {
- return sessionClosedTimerWorker.getMinimum();
+ case SESSION_OPENED:
+ if (profileSessionOpened) {
+ return sessionOpenedTimerWorker.getMinimum();
}
-
+
break;
default:
@@ -749,21 +749,21 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
}
break;
-
- case SESSION_CREATED:
- if (profileSessionCreated) {
- return sessionCreatedTimerWorker.getMaximum();
+
+ case SESSION_CLOSED:
+ if (profileSessionClosed) {
+ return sessionClosedTimerWorker.getMaximum();
}
break;
- case SESSION_OPENED:
- if (profileSessionOpened) {
- return sessionOpenedTimerWorker.getMaximum();
+ case SESSION_CREATED:
+ if (profileSessionCreated) {
+ return sessionCreatedTimerWorker.getMaximum();
}
break;
-
+
case SESSION_IDLE:
if (profileSessionIdle) {
return sessionIdleTimerWorker.getMaximum();
@@ -771,9 +771,9 @@ public class ProfilerTimerFilter extends IoFilterAdapter {
break;
- case SESSION_CLOSED:
- if (profileSessionClosed) {
- return sessionClosedTimerWorker.getMaximum();
+ case SESSION_OPENED:
+ if (profileSessionOpened) {
+ return sessionOpenedTimerWorker.getMaximum();
}
break;
diff --git a/mina-core/src/main/java/org/apache/mina/filter/stream/AbstractStreamWriteFilter.java b/mina-core/src/main/java/org/apache/mina/filter/stream/AbstractStreamWriteFilter.java
index 80a0887..524ee4b 100644
--- a/mina-core/src/main/java/org/apache/mina/filter/stream/AbstractStreamWriteFilter.java
+++ b/mina-core/src/main/java/org/apache/mina/filter/stream/AbstractStreamWriteFilter.java
@@ -105,7 +105,6 @@ public abstract class AbstractStreamWriteFilter<T> extends IoFilterAdapter {
nextFilter.filterWrite(session, new DefaultWriteRequest(buffer));
}
-
} else {
nextFilter.filterWrite(session, writeRequest);
}
diff --git a/mina-core/src/main/java/org/apache/mina/filter/stream/FileRegionWriteFilter.java b/mina-core/src/main/java/org/apache/mina/filter/stream/FileRegionWriteFilter.java
index b48e021..840c65f 100644
--- a/mina-core/src/main/java/org/apache/mina/filter/stream/FileRegionWriteFilter.java
+++ b/mina-core/src/main/java/org/apache/mina/filter/stream/FileRegionWriteFilter.java
@@ -73,7 +73,7 @@ public class FileRegionWriteFilter extends AbstractStreamWriteFilter<FileRegion>
}
// Allocate the buffer for reading from the file
- final int bufferSize = (int) Math.min(getWriteBufferSize(), fileRegion.getRemainingBytes());
+ int bufferSize = (int) Math.min(getWriteBufferSize(), fileRegion.getRemainingBytes());
IoBuffer buffer = IoBuffer.allocate(bufferSize);
// Read from the file
diff --git a/mina-core/src/main/java/org/apache/mina/filter/util/CommonEventFilter.java b/mina-core/src/main/java/org/apache/mina/filter/util/CommonEventFilter.java
index 24ae5e3..d2f6762 100644
--- a/mina-core/src/main/java/org/apache/mina/filter/util/CommonEventFilter.java
+++ b/mina-core/src/main/java/org/apache/mina/filter/util/CommonEventFilter.java
@@ -25,10 +25,11 @@ import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoEventType;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.core.write.WriteRequest;
+import org.apache.mina.filter.FilterEvent;
/**
* Extend this class when you want to create a filter that
- * wraps the same logic around all 9 IoEvents
+ * wraps the same logic around all 11 IoEvents
*
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
*/
@@ -39,40 +40,40 @@ public abstract class CommonEventFilter extends IoFilterAdapter {
* {@inheritDoc}
*/
@Override
- public final void sessionCreated(NextFilter nextFilter, IoSession session) throws Exception {
- filter(new IoFilterEvent(nextFilter, IoEventType.SESSION_CREATED, session, null));
+ public void event(NextFilter nextFilter, IoSession session, FilterEvent event) throws Exception {
+ filter(new IoFilterEvent(nextFilter, IoEventType.EVENT, session, event));
}
/**
* {@inheritDoc}
*/
@Override
- public final void sessionOpened(NextFilter nextFilter, IoSession session) throws Exception {
- filter(new IoFilterEvent(nextFilter, IoEventType.SESSION_OPENED, session, null));
+ public final void exceptionCaught(NextFilter nextFilter, IoSession session, Throwable cause) throws Exception {
+ filter(new IoFilterEvent(nextFilter, IoEventType.EXCEPTION_CAUGHT, session, cause));
}
/**
* {@inheritDoc}
*/
@Override
- public final void sessionClosed(NextFilter nextFilter, IoSession session) throws Exception {
- filter(new IoFilterEvent(nextFilter, IoEventType.SESSION_CLOSED, session, null));
+ public final void filterClose(NextFilter nextFilter, IoSession session) throws Exception {
+ filter(new IoFilterEvent(nextFilter, IoEventType.CLOSE, session, null));
}
/**
* {@inheritDoc}
*/
@Override
- public final void sessionIdle(NextFilter nextFilter, IoSession session, IdleStatus status) throws Exception {
- filter(new IoFilterEvent(nextFilter, IoEventType.SESSION_IDLE, session, status));
+ public final void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
+ filter(new IoFilterEvent(nextFilter, IoEventType.WRITE, session, writeRequest));
}
/**
* {@inheritDoc}
*/
@Override
- public final void exceptionCaught(NextFilter nextFilter, IoSession session, Throwable cause) throws Exception {
- filter(new IoFilterEvent(nextFilter, IoEventType.EXCEPTION_CAUGHT, session, cause));
+ public void inputClosed(NextFilter nextFilter, IoSession session) throws Exception {
+ filter(new IoFilterEvent(nextFilter, IoEventType.INPUT_CLOSED, session, null));
}
/**
@@ -95,15 +96,31 @@ public abstract class CommonEventFilter extends IoFilterAdapter {
* {@inheritDoc}
*/
@Override
- public final void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
- filter(new IoFilterEvent(nextFilter, IoEventType.WRITE, session, writeRequest));
+ public final void sessionClosed(NextFilter nextFilter, IoSession session) throws Exception {
+ filter(new IoFilterEvent(nextFilter, IoEventType.SESSION_CLOSED, session, null));
}
/**
* {@inheritDoc}
*/
@Override
- public final void filterClose(NextFilter nextFilter, IoSession session) throws Exception {
- filter(new IoFilterEvent(nextFilter, IoEventType.CLOSE, session, null));
+ public final void sessionCreated(NextFilter nextFilter, IoSession session) throws Exception {
+ filter(new IoFilterEvent(nextFilter, IoEventType.SESSION_CREATED, session, null));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public final void sessionIdle(NextFilter nextFilter, IoSession session, IdleStatus status) throws Exception {
+ filter(new IoFilterEvent(nextFilter, IoEventType.SESSION_IDLE, session, status));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public final void sessionOpened(NextFilter nextFilter, IoSession session) throws Exception {
+ filter(new IoFilterEvent(nextFilter, IoEventType.SESSION_OPENED, session, null));
}
}
diff --git a/mina-core/src/main/java/org/apache/mina/filter/util/WriteRequestFilter.java b/mina-core/src/main/java/org/apache/mina/filter/util/WriteRequestFilter.java
deleted file mode 100644
index bbc1021..0000000
--- a/mina-core/src/main/java/org/apache/mina/filter/util/WriteRequestFilter.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * 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.util;
-
-import org.apache.mina.core.filterchain.IoFilter;
-import org.apache.mina.core.filterchain.IoFilterAdapter;
-import org.apache.mina.core.session.IoEventType;
-import org.apache.mina.core.session.IoSession;
-import org.apache.mina.core.write.WriteRequest;
-import org.apache.mina.core.write.WriteRequestWrapper;
-
-/**
- * An abstract {@link IoFilter} that simplifies the implementation of
- * an {@link IoFilter} that filters an {@link IoEventType#WRITE} event.
- *
- * @author <a href="http://mina.apache.org">Apache MINA Project</a>
- *
- */
-public abstract class WriteRequestFilter extends IoFilterAdapter {
- /**
- * {@inheritDoc}
- */
- @Override
- public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
- Object filteredMessage = doFilterWrite(nextFilter, session, writeRequest);
-
- if (filteredMessage != null && filteredMessage != writeRequest.getMessage()) {
- nextFilter.filterWrite(session, new FilteredWriteRequest(filteredMessage, writeRequest));
- } else {
- nextFilter.filterWrite(session, writeRequest);
- }
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void messageSent(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
- if (writeRequest instanceof FilteredWriteRequest) {
- FilteredWriteRequest req = (FilteredWriteRequest) writeRequest;
-
- if (req.getParent() == this) {
- nextFilter.messageSent(session, req.getParentRequest());
- return;
- }
- }
-
- nextFilter.messageSent(session, writeRequest);
- }
-
- protected abstract Object doFilterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest)
- throws Exception;
-
- private class FilteredWriteRequest extends WriteRequestWrapper {
- private final Object filteredMessage;
-
- public FilteredWriteRequest(Object filteredMessage, WriteRequest writeRequest) {
- super(writeRequest);
-
- if (filteredMessage == null) {
- throw new IllegalArgumentException("filteredMessage");
- }
- this.filteredMessage = filteredMessage;
- }
-
- public WriteRequestFilter getParent() {
- return WriteRequestFilter.this;
- }
-
- @Override
- public Object getMessage() {
- return filteredMessage;
- }
- }
-}
diff --git a/mina-core/src/main/java/org/apache/mina/transport/socket/nio/NioDatagramAcceptor.java b/mina-core/src/main/java/org/apache/mina/transport/socket/nio/NioDatagramAcceptor.java
index 09c35e7..420c970 100644
--- a/mina-core/src/main/java/org/apache/mina/transport/socket/nio/NioDatagramAcceptor.java
+++ b/mina-core/src/main/java/org/apache/mina/transport/socket/nio/NioDatagramAcceptor.java
@@ -873,7 +873,6 @@ public final class NioDatagramAcceptor extends AbstractIoAcceptor implements Dat
if (buf.remaining() == 0) {
// Clear and fire event
session.setCurrentWriteRequest(null);
- buf.reset();
session.getFilterChain().fireMessageSent(writeRequest);
continue;
}
@@ -898,7 +897,6 @@ public final class NioDatagramAcceptor extends AbstractIoAcceptor implements Dat
// Clear and fire event
session.setCurrentWriteRequest(null);
writtenBytes += localWrittenBytes;
- buf.reset();
session.getFilterChain().fireMessageSent(writeRequest);
break;
diff --git a/mina-core/src/main/java/org/apache/mina/transport/vmpipe/VmPipeFilterChain.java b/mina-core/src/main/java/org/apache/mina/transport/vmpipe/VmPipeFilterChain.java
index 3872e04..3dad7a0 100644
--- a/mina-core/src/main/java/org/apache/mina/transport/vmpipe/VmPipeFilterChain.java
+++ b/mina-core/src/main/java/org/apache/mina/transport/vmpipe/VmPipeFilterChain.java
@@ -34,6 +34,7 @@ import org.apache.mina.core.session.IoEventType;
import org.apache.mina.core.write.WriteRequest;
import org.apache.mina.core.write.WriteRequestQueue;
import org.apache.mina.core.write.WriteToClosedSessionException;
+import org.apache.mina.filter.FilterEvent;
/**
* TODO Add documentation
@@ -87,43 +88,71 @@ class VmPipeFilterChain extends DefaultIoFilterChain {
IoEventType type = e.getType();
Object data = e.getParameter();
- if (type == IoEventType.MESSAGE_RECEIVED) {
- if (sessionOpened && (!session.isReadSuspended()) && session.getLock().tryLock()) {
- try {
- if (session.isReadSuspended()) {
- session.receivedMessageQueue.add(data);
- } else {
- super.fireMessageReceived(data);
+ switch (type) {
+ case EVENT:
+ super.fireEvent((FilterEvent) data);
+ break;
+
+ case EXCEPTION_CAUGHT:
+ super.fireExceptionCaught((Throwable) data);
+ break;
+
+ case CLOSE:
+ super.fireFilterClose();
+ break;
+
+ case INPUT_CLOSED:
+ super.fireInputClosed();
+ break;
+
+ case MESSAGE_SENT:
+ super.fireMessageSent((WriteRequest) data);
+ break;
+
+ case MESSAGE_RECEIVED:
+ if (sessionOpened && (!session.isReadSuspended()) && session.getLock().tryLock()) {
+ try {
+ if (session.isReadSuspended()) {
+ session.receivedMessageQueue.add(data);
+ } else {
+ super.fireMessageReceived(data);
+ }
+ } finally {
+ session.getLock().unlock();
}
+ } else {
+ session.receivedMessageQueue.add(data);
+ }
+
+ break;
+
+ case SESSION_CLOSED:
+ flushPendingDataQueues(session);
+ super.fireSessionClosed();
+ break;
+
+ case SESSION_CREATED:
+ session.getLock().lock();
+ try {
+ super.fireSessionCreated();
} finally {
session.getLock().unlock();
}
- } else {
- session.receivedMessageQueue.add(data);
- }
- } else if (type == IoEventType.WRITE) {
- super.fireFilterWrite((WriteRequest) data);
- } else if (type == IoEventType.MESSAGE_SENT) {
- super.fireMessageSent((WriteRequest) data);
- } else if (type == IoEventType.EXCEPTION_CAUGHT) {
- super.fireExceptionCaught((Throwable) data);
- } else if (type == IoEventType.SESSION_IDLE) {
- super.fireSessionIdle((IdleStatus) data);
- } else if (type == IoEventType.SESSION_OPENED) {
- super.fireSessionOpened();
- sessionOpened = true;
- } else if (type == IoEventType.SESSION_CREATED) {
- session.getLock().lock();
- try {
- super.fireSessionCreated();
- } finally {
- session.getLock().unlock();
- }
- } else if (type == IoEventType.SESSION_CLOSED) {
- flushPendingDataQueues(session);
- super.fireSessionClosed();
- } else if (type == IoEventType.CLOSE) {
- super.fireFilterClose();
+
+ break;
+
+ case SESSION_IDLE:
+ super.fireSessionIdle((IdleStatus) data);
+ break;
+
+ case SESSION_OPENED:
+ super.fireSessionOpened();
+ sessionOpened = true;
+ break;
+
+ case WRITE:
+ super.fireFilterWrite((WriteRequest) data);
+ break;
}
}
@@ -133,11 +162,21 @@ class VmPipeFilterChain extends DefaultIoFilterChain {
}
@Override
+ public void fireEvent(FilterEvent event) {
+ pushEvent(new IoEvent(IoEventType.EVENT, getSession(), event));
+ }
+
+ @Override
public void fireFilterClose() {
pushEvent(new IoEvent(IoEventType.CLOSE, getSession(), null));
}
@Override
+ public void fireInputClosed() {
+ pushEvent(new IoEvent(IoEventType.INPUT_CLOSED, getSession(), null));
+ }
+
+ @Override
public void fireFilterWrite(WriteRequest writeRequest) {
pushEvent(new IoEvent(IoEventType.WRITE, getSession(), writeRequest));
}
diff --git a/mina-core/src/test/java/org/apache/mina/filter/executor/ExecutorFilterRegressionTest.java b/mina-core/src/test/java/org/apache/mina/filter/executor/ExecutorFilterRegressionTest.java
index 078849b..0e9cc8c 100644
--- a/mina-core/src/test/java/org/apache/mina/filter/executor/ExecutorFilterRegressionTest.java
+++ b/mina-core/src/test/java/org/apache/mina/filter/executor/ExecutorFilterRegressionTest.java
@@ -63,7 +63,7 @@ public class ExecutorFilterRegressionTest {
final EventOrderCounter[] sessions = new EventOrderCounter[] { new EventOrderCounter(),
new EventOrderCounter(), new EventOrderCounter(), new EventOrderCounter(), new EventOrderCounter(),
new EventOrderCounter(), new EventOrderCounter(), new EventOrderCounter(), new EventOrderCounter(),
- new EventOrderCounter(), };
+ new EventOrderCounter()};
final int loop = 1000000;
final int end = sessions.length - 1;
final ExecutorFilter filter = this.filter;
diff --git a/mina-core/src/test/java/org/apache/mina/filter/keepalive/KeepAliveFilterTest.java b/mina-core/src/test/java/org/apache/mina/filter/keepalive/KeepAliveFilterTest.java
index 908b04e..757891f 100644
--- a/mina-core/src/test/java/org/apache/mina/filter/keepalive/KeepAliveFilterTest.java
+++ b/mina-core/src/test/java/org/apache/mina/filter/keepalive/KeepAliveFilterTest.java
@@ -129,29 +129,30 @@ public class KeepAliveFilterTest {
connector.dispose();
}
- static boolean checkRequest(IoBuffer message) {
- IoBuffer buff = message;
- boolean check = buff.get() == 1;
- buff.rewind();
- return check;
+ static boolean checkRequest(Object message) {
+ if (message instanceof IoBuffer ) {
+ IoBuffer buff = (IoBuffer)message;
+ boolean check = buff.get() == 1;
+ buff.rewind();
+ return check;
+ } else {
+ return false;
+ }
}
- static boolean checkResponse(IoBuffer message) {
- IoBuffer buff = message;
- boolean check = buff.get() == 2;
- buff.rewind();
- return check;
+ static boolean checkResponse(Object message) {
+ if (message instanceof IoBuffer ) {
+ IoBuffer buff = (IoBuffer)message;
+ boolean check = buff.get() == 2;
+ buff.rewind();
+ return check;
+ } else {
+ return false;
+ }
}
// Inner classes -------------------------------------------------
private final class ServerFactory implements KeepAliveMessageFactory {
- /**
- * Default constructor
- */
- public ServerFactory() {
- super();
- }
-
public Object getRequest(IoSession session) {
return null;
}
@@ -161,28 +162,15 @@ public class KeepAliveFilterTest {
}
public boolean isRequest(IoSession session, Object message) {
- if (message instanceof IoBuffer) {
- return checkRequest((IoBuffer) message);
- }
- return false;
+ return checkRequest(message);
}
public boolean isResponse(IoSession session, Object message) {
- if (message instanceof IoBuffer) {
- return checkResponse((IoBuffer) message);
- }
- return false;
+ return checkResponse(message);
}
}
private final class ClientFactory implements KeepAliveMessageFactory {
- /**
- * Default constructor
- */
- public ClientFactory() {
- super();
- }
-
public Object getRequest(IoSession session) {
return PING.duplicate();
}
@@ -192,17 +180,11 @@ public class KeepAliveFilterTest {
}
public boolean isRequest(IoSession session, Object message) {
- if (message instanceof IoBuffer) {
- return checkRequest((IoBuffer) message);
- }
- return false;
+ return checkRequest(message);
}
public boolean isResponse(IoSession session, Object message) {
- if (message instanceof IoBuffer) {
- return checkResponse((IoBuffer) message);
- }
- return false;
+ return checkResponse(message);
}
}
}
diff --git a/mina-core/src/test/java/org/apache/mina/transport/AbstractBindTest.java b/mina-core/src/test/java/org/apache/mina/transport/AbstractBindTest.java
index ba980b6..938c134 100644
--- a/mina-core/src/test/java/org/apache/mina/transport/AbstractBindTest.java
+++ b/mina-core/src/test/java/org/apache/mina/transport/AbstractBindTest.java
@@ -296,9 +296,7 @@ public abstract class AbstractBindTest {
IoBuffer rb = (IoBuffer) message;
// Write the received data back to remote peer
- IoBuffer wb = IoBuffer.allocate(rb.remaining());
- wb.put(rb);
- wb.flip();
+ IoBuffer wb = rb.duplicate();
session.write(wb);
}
}
diff --git a/mina-core/src/test/resources/log4j.properties b/mina-core/src/test/resources/log4j.properties
index b4d371d..d79de52 100644
--- a/mina-core/src/test/resources/log4j.properties
+++ b/mina-core/src/test/resources/log4j.properties
@@ -16,7 +16,7 @@
#############################################################################
# Please don't modify the log level until we reach to acceptable test coverage.
# It's very useful when I test examples manually.
-log4j.rootCategory=INFO, stdout
+log4j.rootCategory=ERROR, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
diff --git a/mina-example/pom.xml b/mina-example/pom.xml
index c8589c4..5f0cdfd 100644
--- a/mina-example/pom.xml
+++ b/mina-example/pom.xml
@@ -58,6 +58,13 @@
</dependency>
<dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>mina-filter-compression</artifactId>
+ <version>${project.version}</version>
+ <type>bundle</type>
+ </dependency>
+
+ <dependency>
<groupId>org.springframework</groupId>
<artifactId>spring</artifactId>
</dependency>
diff --git a/mina-example/src/main/java/org/apache/mina/example/chat/ChatProtocolHandler.java b/mina-example/src/main/java/org/apache/mina/example/chat/ChatProtocolHandler.java
index 5182e32..e25e62d 100644
--- a/mina-example/src/main/java/org/apache/mina/example/chat/ChatProtocolHandler.java
+++ b/mina-example/src/main/java/org/apache/mina/example/chat/ChatProtocolHandler.java
@@ -52,6 +52,11 @@ public class ChatProtocolHandler extends IoHandlerAdapter {
}
@Override
+ public void messageSent(IoSession session, Object message) {
+ System.out.println( message );
+ }
+
+ @Override
public void messageReceived(IoSession session, Object message) {
Logger log = LoggerFactory.getLogger(ChatProtocolHandler.class);
log.info("received: " + message);
diff --git a/mina-example/src/main/java/org/apache/mina/example/chat/Main.java b/mina-example/src/main/java/org/apache/mina/example/chat/Main.java
index d0a74a0..1e76ad6 100644
--- a/mina-example/src/main/java/org/apache/mina/example/chat/Main.java
+++ b/mina-example/src/main/java/org/apache/mina/example/chat/Main.java
@@ -25,6 +25,7 @@ import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.example.echoserver.ssl.BogusSslContextFactory;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
+import org.apache.mina.filter.compression.CompressionFilter;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.filter.logging.MdcInjectionFilter;
import org.apache.mina.filter.ssl.SslFilter;
@@ -40,7 +41,7 @@ public class Main {
private static final int PORT = 1234;
/** Set this to true if you want to make the server SSL */
- private static final boolean USE_SSL = false;
+ private static final boolean USE_SSL = true;
public static void main(String[] args) throws Exception {
NioSocketAcceptor acceptor = new NioSocketAcceptor();
@@ -54,6 +55,9 @@ public class Main {
addSSLSupport(chain);
}
+ // Add the compression filter
+ chain.addLast( "Compressor", new CompressionFilter() );
+
chain.addLast("codec", new ProtocolCodecFilter(
new TextLineCodecFactory()));
diff --git a/mina-example/src/main/java/org/apache/mina/example/chat/client/ChatClientSupport.java b/mina-example/src/main/java/org/apache/mina/example/chat/client/ChatClientSupport.java
index 7527c91..f099ff6 100644
--- a/mina-example/src/main/java/org/apache/mina/example/chat/client/ChatClientSupport.java
+++ b/mina-example/src/main/java/org/apache/mina/example/chat/client/ChatClientSupport.java
@@ -31,6 +31,7 @@ import org.apache.mina.example.echoserver.ssl.BogusSslContextFactory;
import org.apache.mina.filter.ssl.SslFilter;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
+import org.apache.mina.filter.compression.CompressionFilter;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.filter.logging.MdcInjectionFilter;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
@@ -69,6 +70,9 @@ public class ChatClientSupport {
new TextLineCodecFactory());
connector.getFilterChain().addLast("mdc", new MdcInjectionFilter());
+
+ // Add the compression filter
+ connector.getFilterChain().addLast( "Compression", new CompressionFilter() );
connector.getFilterChain().addLast("codec", CODEC_FILTER);
connector.getFilterChain().addLast("logger", LOGGING_FILTER);
@@ -101,9 +105,7 @@ public class ChatClientSupport {
public void broadcast(String message) {
try {
- for ( int i = 0; i < 1000000; i++) {
- session.write("BROADCAST " + message + i);
- }
+ session.write("BROADCAST " + message);
} catch ( Exception e ) {
e.printStackTrace();
}
diff --git a/mina-example/src/main/java/org/apache/mina/example/echoserver/Main.java b/mina-example/src/main/java/org/apache/mina/example/echoserver/Main.java
index 3e74064..6501e78 100644
--- a/mina-example/src/main/java/org/apache/mina/example/echoserver/Main.java
+++ b/mina-example/src/main/java/org/apache/mina/example/echoserver/Main.java
@@ -23,6 +23,7 @@ import java.net.InetSocketAddress;
import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.example.echoserver.ssl.BogusSslContextFactory;
+import org.apache.mina.filter.compression.CompressionFilter;
import org.apache.mina.filter.ssl.SslFilter;
import org.apache.mina.transport.socket.SocketAcceptor;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
@@ -48,6 +49,9 @@ public class Main {
if (USE_SSL) {
addSSLSupport(chain);
}
+
+ // Add the compressor filter
+ chain.addLast( "Compressor", new CompressionFilter() );
// Bind
acceptor.setHandler(new EchoProtocolHandler());
diff --git a/mina-filter-compression/src/main/java/org/apache/mina/filter/compression/CompressionFilter.java b/mina-filter-compression/src/main/java/org/apache/mina/filter/compression/CompressionFilter.java
index b57f090..987a6bc 100644
--- a/mina-filter-compression/src/main/java/org/apache/mina/filter/compression/CompressionFilter.java
+++ b/mina-filter-compression/src/main/java/org/apache/mina/filter/compression/CompressionFilter.java
@@ -23,11 +23,11 @@ import java.io.IOException;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.filterchain.IoFilter;
+import org.apache.mina.core.filterchain.IoFilterAdapter;
import org.apache.mina.core.filterchain.IoFilterChain;
import org.apache.mina.core.session.AttributeKey;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.core.write.WriteRequest;
-import org.apache.mina.filter.util.WriteRequestFilter;
/**
* An {@link IoFilter} which compresses all data using
@@ -55,7 +55,7 @@ import org.apache.mina.filter.util.WriteRequestFilter;
*
* @author <a href="http://mina.apache.org">Apache MINA Project</a>
*/
-public class CompressionFilter extends WriteRequestFilter {
+public class CompressionFilter extends IoFilterAdapter {
/**
* Max compression level. Will give the highest compression ratio, but
* will also take more cpu time and is the slowest.
@@ -137,6 +137,20 @@ public class CompressionFilter extends WriteRequestFilter {
this.compressInbound = compressInbound;
this.compressOutbound = compressOutbound;
}
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
+ Object compressedMessage = doFilterWrite(nextFilter, session, writeRequest);
+
+ if (compressedMessage != null && compressedMessage != writeRequest.getMessage()) {
+ writeRequest.setMessage( compressedMessage );
+ }
+
+ nextFilter.filterWrite(session, writeRequest);
+ }
@Override
public void messageReceived(NextFilter nextFilter, IoSession session, Object message) throws Exception {
@@ -146,19 +160,26 @@ public class CompressionFilter extends WriteRequestFilter {
}
Zlib inflater = (Zlib) session.getAttribute(INFLATER);
+
if (inflater == null) {
throw new IllegalStateException();
}
IoBuffer inBuffer = (IoBuffer) message;
- IoBuffer outBuffer = inflater.inflate(inBuffer);
- nextFilter.messageReceived(session, outBuffer);
+ nextFilter.messageReceived(session, inflater.inflate(inBuffer));
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ @Override
+ public void messageSent(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
+ nextFilter.messageSent(session, writeRequest);
}
/*
* @see org.apache.mina.core.IoFilter#filterWrite(org.apache.mina.core.IoFilter.NextFilter, org.apache.mina.core.IoSession, org.apache.mina.core.IoFilter.WriteRequest)
*/
- @Override
protected Object doFilterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest)
throws IOException {
if (!compressOutbound) {
@@ -172,11 +193,13 @@ public class CompressionFilter extends WriteRequestFilter {
}
Zlib deflater = (Zlib) session.getAttribute(DEFLATER);
+
if (deflater == null) {
throw new IllegalStateException();
}
IoBuffer inBuffer = (IoBuffer) writeRequest.getMessage();
+
if (!inBuffer.hasRemaining()) {
// Ignore empty buffers
return null;
diff --git a/mina-filter-compression/src/main/java/org/apache/mina/filter/compression/Zlib.java b/mina-filter-compression/src/main/java/org/apache/mina/filter/compression/Zlib.java
index 403cdb3..dc7a07c 100644
--- a/mina-filter-compression/src/main/java/org/apache/mina/filter/compression/Zlib.java
+++ b/mina-filter-compression/src/main/java/org/apache/mina/filter/compression/Zlib.java
@@ -173,7 +173,7 @@ class Zlib {
}
byte[] inBytes = new byte[inBuffer.remaining()];
- inBuffer.get(inBytes).flip();
+ inBuffer.get(inBytes);
// according to spec, destination buffer should be 0.1% larger
// than source length plus 12 bytes. We add a single byte to safeguard
diff --git a/mina-filter-compression/src/test/java/org/apache/mina/filter/compression/CompressionFilterTest.java b/mina-filter-compression/src/test/java/org/apache/mina/filter/compression/CompressionFilterTest.java
index 52d17f4..5b097c5 100644
--- a/mina-filter-compression/src/test/java/org/apache/mina/filter/compression/CompressionFilterTest.java
+++ b/mina-filter-compression/src/test/java/org/apache/mina/filter/compression/CompressionFilterTest.java
@@ -104,6 +104,7 @@ public class CompressionFilterTest {
// prepare the input data
IoBuffer buf = IoBuffer.wrap(strCompress.getBytes("UTF8"));
IoBuffer actualOutput = actualDeflater.deflate(buf);
+ buf.flip();
WriteRequest writeRequest = new DefaultWriteRequest(buf);
// record all the mock calls
diff --git a/mina-filter-compression/src/test/java/org/apache/mina/filter/compression/ZlibTest.java b/mina-filter-compression/src/test/java/org/apache/mina/filter/compression/ZlibTest.java
index 5056c3b..99a37b7 100644
--- a/mina-filter-compression/src/test/java/org/apache/mina/filter/compression/ZlibTest.java
+++ b/mina-filter-compression/src/test/java/org/apache/mina/filter/compression/ZlibTest.java
@@ -61,6 +61,7 @@ public class ZlibTest {
IoBuffer byteUncompressed = inflater.inflate(byteCompressed);
String strOutput = byteUncompressed.getString(Charset.forName("UTF8").newDecoder());
assertTrue(strOutput.equals(strInput));
+ byteInput.flip();
}
}
@@ -115,6 +116,8 @@ public class ZlibTest {
String strOutput = byteUncompressed.getString(Charset.forName("UTF8").newDecoder());
assertTrue(strOutput.equals(strInput));
}
+
+ byteInput.flip();
}
// check if the last compressed data block can be decompressed
// successfully.