You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@directory.apache.org by ni...@apache.org on 2006/01/15 16:37:30 UTC
svn commit: r369220 - in /directory/trunks/mina/core/src:
main/java/org/apache/mina/filter/StreamWriteFilter.java
test/java/org/apache/mina/filter/StreamWriteFilterTest.java
Author: niklas
Date: Sun Jan 15 07:37:21 2006
New Revision: 369220
URL: http://svn.apache.org/viewcvs?rev=369220&view=rev
Log:
Added StreamWriteFilter
Added:
directory/trunks/mina/core/src/main/java/org/apache/mina/filter/StreamWriteFilter.java (with props)
directory/trunks/mina/core/src/test/java/org/apache/mina/filter/StreamWriteFilterTest.java (with props)
Added: directory/trunks/mina/core/src/main/java/org/apache/mina/filter/StreamWriteFilter.java
URL: http://svn.apache.org/viewcvs/directory/trunks/mina/core/src/main/java/org/apache/mina/filter/StreamWriteFilter.java?rev=369220&view=auto
==============================================================================
--- directory/trunks/mina/core/src/main/java/org/apache/mina/filter/StreamWriteFilter.java (added)
+++ directory/trunks/mina/core/src/main/java/org/apache/mina/filter/StreamWriteFilter.java Sun Jan 15 07:37:21 2006
@@ -0,0 +1,178 @@
+/*
+ * @(#) $Id$
+ *
+ * Copyright 2004 The Apache Software Foundation
+ *
+ * Licensed 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;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.mina.common.ByteBuffer;
+import org.apache.mina.common.IoFilterAdapter;
+import org.apache.mina.common.IoSession;
+import org.apache.mina.common.WriteFuture;
+
+/**
+ * Filter implementation which makes it possible to write {@link InputStream}
+ * objects directly using {@link IoSession#write(Object)}. When an
+ * {@link InputStream} is written to a session this filter will read the bytes
+ * from the stream into {@link ByteBuffer} objects and write those buffers
+ * to the next filter. When end of stream has been reached this filter will
+ * call {@link NextFilter#messageSent(IoSession, Object)} using the original
+ * {@link InputStream} written to the session and call
+ * {@link org.apache.mina.common.WriteFuture#setWritten(boolean)} on the
+ * original {@link org.apache.mina.common.IoFilter.WriteRequest}.
+ * <p>
+ * This filter will ignore written messages which aren't {@link InputStream}
+ * instances. Such messages will be passed to the next filter directly.
+ * </p>
+ * <p>
+ * NOTE: this filter does not close the stream after all data from stream
+ * has been written. The {@link org.apache.mina.common.IoHandler} should take
+ * care of that in its
+ * {@link org.apache.mina.common.IoHandler#messageSent(IoSession, Object)}
+ * callback.
+ * </p>
+ *
+ * @author The Apache Directory Project (dev@directory.apache.org)
+ * @version $Rev$, $Date$
+ */
+public class StreamWriteFilter extends IoFilterAdapter
+{
+ /**
+ * The default buffer size this filter uses for writing.
+ */
+ public static final int DEFAULT_STREAM_BUFFER_SIZE = 4096;
+
+ /**
+ * The attribute name used when binding the {@link InputStream} to the session.
+ */
+ public static final String CURRENT_STREAM = StreamWriteFilter.class.getName() + ".stream";
+
+ protected static final String INITIAL_WRITE_FUTURE = StreamWriteFilter.class.getName() + ".future";
+
+ private int writeBufferSize = DEFAULT_STREAM_BUFFER_SIZE;
+
+ public void filterWrite( NextFilter nextFilter, IoSession session,
+ WriteRequest writeRequest ) throws Exception
+ {
+ // Make sure we're not already processing a stream.
+ if( session.getAttribute( CURRENT_STREAM ) != null )
+ {
+ throw new IllegalStateException( "Already processing a stream" );
+ }
+
+ Object message = writeRequest.getMessage();
+
+ if( message instanceof InputStream )
+ {
+
+ InputStream inputStream = ( InputStream ) message;
+
+ ByteBuffer byteBuffer = getNextByteBuffer( inputStream );
+ if ( byteBuffer == null )
+ {
+ // End of stream reached.
+ writeRequest.getFuture().setWritten( true );
+ nextFilter.messageSent( session, message );
+ }
+ else
+ {
+ session.setAttribute( CURRENT_STREAM, inputStream );
+ session.setAttribute( INITIAL_WRITE_FUTURE, writeRequest.getFuture() );
+
+ nextFilter.filterWrite( session, new WriteRequest( byteBuffer ) );
+ }
+
+ }
+ else
+ {
+ nextFilter.filterWrite( session, writeRequest );
+ }
+ }
+
+ public void messageSent( NextFilter nextFilter, IoSession session, Object message ) throws Exception
+ {
+ InputStream inputStream = ( InputStream ) session.getAttribute( CURRENT_STREAM );
+
+ if( inputStream == null )
+ {
+ nextFilter.messageSent( session, message );
+ }
+ else
+ {
+ ByteBuffer byteBuffer = getNextByteBuffer( inputStream );
+
+ if( byteBuffer == null )
+ {
+ // End of stream reached.
+ session.removeAttribute( CURRENT_STREAM );
+ WriteFuture writeFuture = ( WriteFuture ) session.removeAttribute( INITIAL_WRITE_FUTURE );
+ writeFuture.setWritten( true );
+ nextFilter.messageSent( session, inputStream );
+ }
+ else
+ {
+ nextFilter.filterWrite( session, new WriteRequest( byteBuffer ) );
+ }
+ }
+ }
+
+ private ByteBuffer getNextByteBuffer( InputStream inputStream ) throws IOException
+ {
+ byte[] bytes = new byte[ writeBufferSize ];
+ int packetLength = inputStream.read( bytes );
+ if( packetLength == -1 )
+ {
+ return null;
+ }
+
+ ByteBuffer buffer = ByteBuffer.wrap( bytes, 0, packetLength );
+ buffer.acquire(); // prevent from being pooled.
+
+ return buffer;
+ }
+
+ /**
+ * Returns the size of the write buffer in bytes. Data will be read from the
+ * stream in chunks of this size and then written to the next filter.
+ *
+ * @return the write buffer size.
+ */
+ public int getWriteBufferSize()
+ {
+ return writeBufferSize;
+ }
+
+ /**
+ * Sets the size of the write buffer in bytes. Data will be read from the
+ * stream in chunks of this size and then written to the next filter.
+ *
+ * @return the write buffer size.
+ * @throws IllegalArgumentException if the specified size is < 1.
+ */
+ public void setWriteBufferSize( int writeBufferSize )
+ {
+ if( writeBufferSize < 1 )
+ {
+ throw new IllegalArgumentException( "writeBufferSize must be at least 1" );
+ }
+ this.writeBufferSize = writeBufferSize;
+ }
+
+
+}
Propchange: directory/trunks/mina/core/src/main/java/org/apache/mina/filter/StreamWriteFilter.java
------------------------------------------------------------------------------
svn:keywords = Id
Added: directory/trunks/mina/core/src/test/java/org/apache/mina/filter/StreamWriteFilterTest.java
URL: http://svn.apache.org/viewcvs/directory/trunks/mina/core/src/test/java/org/apache/mina/filter/StreamWriteFilterTest.java?rev=369220&view=auto
==============================================================================
--- directory/trunks/mina/core/src/test/java/org/apache/mina/filter/StreamWriteFilterTest.java (added)
+++ directory/trunks/mina/core/src/test/java/org/apache/mina/filter/StreamWriteFilterTest.java Sun Jan 15 07:37:21 2006
@@ -0,0 +1,525 @@
+/*
+ * @(#) $Id$
+ *
+ * Copyright 2004 The Apache Software Foundation
+ *
+ * Licensed 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;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.security.MessageDigest;
+import java.util.Random;
+
+import junit.framework.TestCase;
+
+import org.apache.mina.common.ByteBuffer;
+import org.apache.mina.common.IdleStatus;
+import org.apache.mina.common.IoHandlerAdapter;
+import org.apache.mina.common.IoSession;
+import org.apache.mina.common.TransportType;
+import org.apache.mina.common.IoFilter.NextFilter;
+import org.apache.mina.common.IoFilter.WriteRequest;
+import org.apache.mina.registry.Service;
+import org.apache.mina.registry.ServiceRegistry;
+import org.apache.mina.registry.SimpleServiceRegistry;
+import org.apache.mina.transport.socket.nio.SocketAcceptor;
+import org.apache.mina.transport.socket.nio.SocketConnector;
+import org.apache.mina.util.AvailablePortFinder;
+import org.easymock.AbstractMatcher;
+import org.easymock.MockControl;
+
+/**
+ * Tests {@link StreamWriteFilter}.
+ *
+ * @author The Apache Directory Project (dev@directory.apache.org)
+ * @version $Rev$, $Date$
+ */
+public class StreamWriteFilterTest extends TestCase {
+ MockControl mockSession;
+ MockControl mockNextFilter;
+ IoSession session;
+ NextFilter nextFilter;
+
+ protected void setUp() throws Exception
+ {
+ super.setUp();
+
+ /*
+ * Create the mocks.
+ */
+ mockSession = MockControl.createControl( IoSession.class );
+ mockNextFilter = MockControl.createControl( NextFilter.class );
+ session = ( IoSession ) mockSession.getMock();
+ nextFilter = ( NextFilter ) mockNextFilter.getMock();
+
+ session.getAttribute( StreamWriteFilter.CURRENT_STREAM );
+ mockSession.setReturnValue( null );
+ }
+
+ public void testWriteEmptyStream() throws Exception
+ {
+ StreamWriteFilter filter = new StreamWriteFilter();
+
+ InputStream stream = new ByteArrayInputStream( new byte[ 0 ] );
+ WriteRequest writeRequest = new WriteRequest( stream );
+
+ /*
+ * Record expectations
+ */
+ nextFilter.messageSent( session, stream );
+
+ /*
+ * Replay.
+ */
+ mockNextFilter.replay();
+ mockSession.replay();
+
+ filter.filterWrite( nextFilter, session, writeRequest );
+
+ /*
+ * Verify.
+ */
+ mockNextFilter.verify();
+ mockSession.verify();
+
+ assertTrue( writeRequest.getFuture().isWritten() );
+ }
+
+ /**
+ * Tests that the filter just passes objects which aren't InputStreams
+ * through to the next filter.
+ */
+ public void testWriteNonStreamMessage() throws Exception
+ {
+ StreamWriteFilter filter = new StreamWriteFilter();
+
+ Object message = new Object();
+ WriteRequest writeRequest = new WriteRequest( message );
+
+ /*
+ * Record expectations
+ */
+ nextFilter.filterWrite( session, writeRequest );
+ session.getAttribute( StreamWriteFilter.CURRENT_STREAM );
+ mockSession.setReturnValue( null );
+ nextFilter.messageSent( session, message );
+
+ /*
+ * Replay.
+ */
+ mockNextFilter.replay();
+ mockSession.replay();
+
+ filter.filterWrite( nextFilter, session, writeRequest );
+ filter.messageSent( nextFilter, session, message );
+
+ /*
+ * Verify.
+ */
+ mockNextFilter.verify();
+ mockSession.verify();
+
+ assertTrue( writeRequest.getFuture().isWritten() );
+ }
+
+ /**
+ * Tests when the contents of the stream fits into one write buffer.
+ */
+ public void testWriteSingleBufferStream() throws Exception
+ {
+ StreamWriteFilter filter = new StreamWriteFilter();
+
+ byte[] data = new byte[] { 1, 2, 3, 4 };
+
+ InputStream stream = new ByteArrayInputStream( data );
+ WriteRequest writeRequest = new WriteRequest( stream );
+
+ /*
+ * Record expectations
+ */
+ session.setAttribute( StreamWriteFilter.CURRENT_STREAM, stream );
+ mockSession.setReturnValue(null);
+ session.setAttribute( StreamWriteFilter.INITIAL_WRITE_FUTURE, writeRequest.getFuture() );
+ mockSession.setReturnValue(null);
+ nextFilter.filterWrite( session, new WriteRequest( ByteBuffer.wrap( data ) ) );
+ mockNextFilter.setMatcher( new WriteRequestMatcher() );
+
+ session.getAttribute( StreamWriteFilter.CURRENT_STREAM );
+ mockSession.setReturnValue( stream );
+ session.removeAttribute( StreamWriteFilter.CURRENT_STREAM );
+ mockSession.setReturnValue( stream );
+ session.removeAttribute( StreamWriteFilter.INITIAL_WRITE_FUTURE );
+ mockSession.setReturnValue( writeRequest.getFuture() );
+ nextFilter.messageSent( session, stream );
+
+ /*
+ * Replay.
+ */
+ mockNextFilter.replay();
+ mockSession.replay();
+
+ filter.filterWrite( nextFilter, session, writeRequest );
+ filter.messageSent( nextFilter, session, data );
+
+ /*
+ * Verify.
+ */
+ mockNextFilter.verify();
+ mockSession.verify();
+
+ assertTrue( writeRequest.getFuture().isWritten() );
+ }
+
+ /**
+ * Tests when the contents of the stream doesn't fit into one write buffer.
+ */
+ public void testWriteSeveralBuffersStream() throws Exception
+ {
+ StreamWriteFilter filter = new StreamWriteFilter();
+ filter.setWriteBufferSize( 4 );
+
+ byte[] data = new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 };
+ byte[] chunk1 = new byte[] { 1, 2, 3, 4 };
+ byte[] chunk2 = new byte[] { 5, 6, 7, 8 };
+ byte[] chunk3 = new byte[] { 9, 10 };
+
+ InputStream stream = new ByteArrayInputStream( data );
+ WriteRequest writeRequest = new WriteRequest( stream );
+
+ /*
+ * Record expectations
+ */
+ session.setAttribute( StreamWriteFilter.CURRENT_STREAM, stream );
+ mockSession.setReturnValue(null);
+ session.setAttribute( StreamWriteFilter.INITIAL_WRITE_FUTURE, writeRequest.getFuture() );
+ mockSession.setReturnValue(null);
+ nextFilter.filterWrite( session, new WriteRequest( ByteBuffer.wrap( chunk1 ) ) );
+ mockNextFilter.setMatcher( new WriteRequestMatcher() );
+
+ session.getAttribute( StreamWriteFilter.CURRENT_STREAM );
+ mockSession.setReturnValue( stream );
+ nextFilter.filterWrite( session, new WriteRequest( ByteBuffer.wrap( chunk2 ) ) );
+
+ session.getAttribute( StreamWriteFilter.CURRENT_STREAM );
+ mockSession.setReturnValue( stream );
+ nextFilter.filterWrite( session, new WriteRequest( ByteBuffer.wrap( chunk3 ) ) );
+
+ session.getAttribute( StreamWriteFilter.CURRENT_STREAM );
+ mockSession.setReturnValue( stream );
+ session.removeAttribute( StreamWriteFilter.CURRENT_STREAM );
+ mockSession.setReturnValue( stream );
+ session.removeAttribute( StreamWriteFilter.INITIAL_WRITE_FUTURE );
+ mockSession.setReturnValue( writeRequest.getFuture() );
+ nextFilter.messageSent( session, stream );
+
+ /*
+ * Replay.
+ */
+ mockNextFilter.replay();
+ mockSession.replay();
+
+ filter.filterWrite( nextFilter, session, writeRequest );
+ filter.messageSent( nextFilter, session, chunk1 );
+ filter.messageSent( nextFilter, session, chunk2 );
+ filter.messageSent( nextFilter, session, chunk3 );
+
+ /*
+ * Verify.
+ */
+ mockNextFilter.verify();
+ mockSession.verify();
+
+ assertTrue( writeRequest.getFuture().isWritten() );
+ }
+
+ public void testWriteWhileWriteInProgress() throws Exception
+ {
+ StreamWriteFilter filter = new StreamWriteFilter();
+
+ InputStream stream = new ByteArrayInputStream( new byte[ 5 ] );
+
+ /*
+ * Record expectations
+ */
+ mockSession.reset();
+ session.getAttribute( StreamWriteFilter.CURRENT_STREAM );
+ mockSession.setReturnValue( stream );
+
+ /*
+ * Replay.
+ */
+ mockNextFilter.replay();
+ mockSession.replay();
+
+ try
+ {
+ filter.filterWrite( nextFilter, session, new WriteRequest( new Object() ) );
+ fail( "Alreday processing a stream. IllegalStateException expected." );
+ }
+ catch ( IllegalStateException ise )
+ {
+ }
+
+ /*
+ * Verify.
+ */
+ mockNextFilter.verify();
+ mockSession.verify();
+ }
+
+ /**
+ * Tests that {@link StreamWriteFilter#setWriteBufferSize(int)} checks the
+ * specified size.
+ */
+ public void testSetWriteBufferSize() throws Exception
+ {
+ StreamWriteFilter filter = new StreamWriteFilter();
+
+ try
+ {
+ filter.setWriteBufferSize( 0 );
+ fail( "0 writeBuferSize specified. IllegalArgumentException expected." );
+ }
+ catch ( IllegalArgumentException iae )
+ {
+ }
+
+ try
+ {
+ filter.setWriteBufferSize( -100 );
+ fail( "Negative writeBuferSize specified. IllegalArgumentException expected." );
+ }
+ catch ( IllegalArgumentException iae )
+ {
+ }
+
+ filter.setWriteBufferSize( 1 );
+ assertEquals( 1, filter.getWriteBufferSize() );
+ filter.setWriteBufferSize( 1024 );
+ assertEquals( 1024, filter.getWriteBufferSize() );
+ }
+
+ public void testWriteUsingSocketTransport() throws Exception
+ {
+ ServiceRegistry reg = new SimpleServiceRegistry();
+ ( ( SocketAcceptor ) reg.getAcceptor( TransportType.SOCKET ) ).setReuseAddress( true );
+ SocketAddress address = new InetSocketAddress( "localhost", AvailablePortFinder.getNextAvailable() );
+ Service service = new Service( "stream", TransportType.SOCKET, address );
+
+ SocketConnector connector = new SocketConnector();
+ connector.getFilterChain().addFirst( "threadPool", new ThreadPoolFilter() );
+
+ FixedRandomInputStream stream = new FixedRandomInputStream( 4 * 1024 * 1024 );
+
+ SenderHandler sender = new SenderHandler( stream );
+ ReceiverHandler receiver = new ReceiverHandler( stream.size );
+
+ reg.bind( service, sender );
+
+ synchronized( sender.lock )
+ {
+ synchronized( receiver.lock )
+ {
+ connector.connect( address, receiver );
+
+ sender.lock.wait();
+ receiver.lock.wait();
+ }
+ }
+
+ reg.unbindAll();
+
+ assertEquals( stream.bytesRead, receiver.bytesRead );
+ assertEquals( stream.size, receiver.bytesRead );
+ byte[] expectedMd5 = stream.digest.digest();
+ byte[] actualMd5 = receiver.digest.digest();
+ assertEquals( expectedMd5.length, actualMd5.length );
+ for( int i = 0; i < expectedMd5.length; i++ )
+ {
+ assertEquals( expectedMd5[ i ], actualMd5[ i ] );
+ }
+ }
+
+ private static class FixedRandomInputStream extends InputStream
+ {
+ long size;
+ long bytesRead = 0;
+ Random random = new Random();
+ MessageDigest digest;
+
+ public FixedRandomInputStream( long size ) throws Exception
+ {
+ this.size = size;
+ digest = MessageDigest.getInstance( "MD5" );
+ }
+
+ public int read() throws IOException
+ {
+ if ( isAllWritten() )
+ return -1;
+ bytesRead++;
+ byte b = ( byte ) random.nextInt( 255 );
+ digest.update( b );
+ return b;
+ }
+
+ public long getBytesRead()
+ {
+ return bytesRead;
+ }
+
+ public long getSize()
+ {
+ return size;
+ }
+
+ public boolean isAllWritten()
+ {
+ return bytesRead >= size;
+ }
+ }
+
+ private static class SenderHandler extends IoHandlerAdapter
+ {
+ Object lock = new Object();
+ InputStream inputStream;
+ StreamWriteFilter streamWriteFilter = new StreamWriteFilter();
+
+ public SenderHandler( InputStream inputStream )
+ {
+ this.inputStream = inputStream;
+ }
+
+ public void sessionCreated(IoSession session) throws Exception {
+ super.sessionCreated( session );
+ session.getFilterChain().addLast( "codec", streamWriteFilter );
+ }
+
+ public void sessionOpened(IoSession session) throws Exception {
+ session.write( inputStream );
+ }
+
+ public void exceptionCaught( IoSession session, Throwable cause ) throws Exception
+ {
+ synchronized( lock )
+ {
+ lock.notifyAll();
+ }
+ }
+
+ public void sessionClosed( IoSession session ) throws Exception
+ {
+ synchronized( lock )
+ {
+ lock.notifyAll();
+ }
+ }
+
+ public void sessionIdle( IoSession session, IdleStatus status ) throws Exception
+ {
+ synchronized( lock )
+ {
+ lock.notifyAll();
+ }
+ }
+
+ public void messageSent( IoSession session, Object message ) throws Exception
+ {
+ if( message == inputStream )
+ {
+ synchronized( lock )
+ {
+ lock.notifyAll();
+ }
+ }
+ }
+ }
+
+ private static class ReceiverHandler extends IoHandlerAdapter
+ {
+ Object lock = new Object();
+ long bytesRead = 0;
+ long size = 0;
+ MessageDigest digest;
+
+ public ReceiverHandler( long size ) throws Exception
+ {
+ this.size = size;
+ digest = MessageDigest.getInstance( "MD5" );
+ }
+
+ public void sessionCreated( IoSession session ) throws Exception
+ {
+ super.sessionCreated(session);
+
+ session.setIdleTime( IdleStatus.READER_IDLE, 5 );
+ }
+
+ public void sessionIdle( IoSession session, IdleStatus status ) throws Exception
+ {
+ session.close();
+ }
+
+ public void exceptionCaught( IoSession session, Throwable cause ) throws Exception
+ {
+ synchronized( lock )
+ {
+ lock.notifyAll();
+ }
+ }
+
+ public void sessionClosed( IoSession session ) throws Exception
+ {
+ synchronized( lock )
+ {
+ lock.notifyAll();
+ }
+ }
+
+ public void messageReceived( IoSession session, Object message ) throws Exception
+ {
+ ByteBuffer buf = ( ByteBuffer ) message;
+ while( buf.hasRemaining() )
+ {
+ digest.update( buf.get() );
+ bytesRead++;
+ }
+ if( bytesRead >= size )
+ {
+ session.close();
+ }
+ }
+ }
+
+ public static class WriteRequestMatcher extends AbstractMatcher
+ {
+ protected boolean argumentMatches( Object expected, Object actual )
+ {
+ if( expected instanceof WriteRequest && expected instanceof WriteRequest )
+ {
+ WriteRequest w1 = ( WriteRequest ) expected;
+ WriteRequest w2 = ( WriteRequest ) actual;
+
+ return w1.getMessage().equals( w2.getMessage() )
+ && w1.getFuture().isWritten() == w2.getFuture().isWritten();
+ }
+ return super.argumentMatches( expected, actual );
+ }
+ }
+}
Propchange: directory/trunks/mina/core/src/test/java/org/apache/mina/filter/StreamWriteFilterTest.java
------------------------------------------------------------------------------
svn:keywords = Id