/*
 *   @(#) $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 java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;

import org.apache.mina.common.ByteBuffer;
import org.apache.mina.common.IoFilterAdapter;
import org.apache.mina.common.IoSession;

/**
 * Simple filter that uses Java serialization to convert objects to and from
 * MINA ByteBuffers, to achieve transparent object messaging over the
 * underlyings network protocols.
 * 
 * @author Matteo Merli (matteo.merli@gmail.com)
 * @version $Rev$
 */
public class SerializationFilter extends IoFilterAdapter
{

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.apache.mina.common.IoFilterAdapter#messageReceived(org.apache.mina.common.IoFilter.NextFilter,
	 *      org.apache.mina.common.IoSession, java.lang.Object)
	 */
	public void messageReceived( NextFilter nextFilter, IoSession session, Object message )
			throws Exception
	{
		// DeSerialize the message if it's a MINA ByteBuffer
		if ( !( message instanceof ByteBuffer ) ) 
		{
			nextFilter.messageReceived( session, message );
			return;
		}

		// Deserialize
		InputStream is = newInputStream( (ByteBuffer) message );
		ObjectInputStream ois = new ObjectInputStream( is );
		Object decodedMessage = ois.readObject();
		ois.close();

		// Forward message
		nextFilter.messageReceived( session, decodedMessage );
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.apache.mina.common.IoFilterAdapter#filterWrite(org.apache.mina.common.IoFilter.NextFilter,
	 *      org.apache.mina.common.IoSession,
	 *      org.apache.mina.common.IoFilter.WriteRequest)
	 */
	public void filterWrite( NextFilter nextFilter, IoSession session,
			WriteRequest writeRequest ) throws Exception
	{
		// Serialize the message if it's not a MINA ByteBuffer
		Object message = writeRequest.getMessage();
		if ( message instanceof ByteBuffer ) 
		{
			nextFilter.filterWrite( session, writeRequest );
			return;
		}

		// Allocate a buffer to contain the serialized object
		ByteBuffer buffer = ByteBuffer.allocate( 0 );
		buffer.setAutoExpand( true );

		// Serialize to buffer
		OutputStream os = newOutputStream( buffer );
		ObjectOutputStream oos = new ObjectOutputStream( os );
		oos.writeObject( message );
		oos.flush();
		oos.close();

		buffer.flip();

		// Forward the message
		nextFilter.filterWrite( session, new WriteRequest( buffer,
				writeRequest.getFuture() ) );
	}

	/**
	 * @return an output stream for a ByteBuffer
	 */
	protected static OutputStream newOutputStream( final ByteBuffer buf )
	{
		return new OutputStream()
		{

			public synchronized void write( int b ) throws IOException
			{
				buf.put( (byte) b );
			}

			public synchronized void write( byte[] bytes, int off, int len )
					throws IOException
			{
				buf.put( bytes, off, len );
			}
		};
	}

	/**
	 * @return an input stream for a ByteBuffer
	 */
	protected static InputStream newInputStream( final ByteBuffer buf )
	{
		return new InputStream()
		{

			public synchronized int read() throws IOException
			{
				if ( !buf.hasRemaining() ) 
				{
					return -1;
				}
				return buf.get();
			}

			public synchronized int read( byte[] bytes, int off, int len )
					throws IOException
			{
				// Read only what's left
				len = Math.min( len, buf.remaining() );
				buf.get( bytes, off, len );
				return len;
			}
		};
	}

}