util/src/main/java/org/apache/etch/util/core/io/Packetizer.java - etch - Git at Google

 /* $Id$
  *
  * 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.etch.util.core.io;

 import java.io.IOException;

 import org.apache.etch.util.FlexBuffer;
 import org.apache.etch.util.Resources;
 import org.apache.etch.util.URL;
 import org.apache.etch.util.core.Who;


 /**
  * Packetizes a stream data source. Reads a packet header:
  * a 32-bit flag and a 32-bit length, both big-endian, verifies
  * the flag, and then, using the length from the header,
  * reads the packet data and passes it to the packet handler.
  * As a packet source, accepts a packet and prepends a packet
  * header to it before delivering it to a data source.
  */
 public final class Packetizer implements SessionData, TransportPacket
 {
 	/**
 	 * URI term to specify max packet size.
 	 */
 	public final static String MAX_PKT_SIZE_TERM = "Packetizer.maxPktSize";

 	private final static int SIG = 0xdeadbeef;

 	private final static int HEADER_SIZE = 8;

 	/**
 	 * The default maximum packet size that will be accepted, 16376 bytes.
 	 */
 	public final static int DEFAULT_MAX_PKT_SIZE = 16384 - HEADER_SIZE;

 	/**
 	 * Constructs the Packetizer with null packet handler and uri specified
 	 * or defaulted maximum packet size.
 	 * @param transport used to deliver data to the peer.
 	 * @param uri
 	 * @param resources
 	 */
 	public Packetizer( TransportData transport, String uri, Resources resources )
 	{
 		this( transport, new URL( uri ), resources );
 	}

 	/**
 	 * Constructs the Packetizer with null packet handler and uri specified
 	 * or defaulted maximum packet size.
 	 *
 	 * @param transport used to deliver data to the peer.
 	 * @param uri
 	 * @param resources
 	 *
 	 * @see #MAX_PKT_SIZE_TERM
 	 * @see #DEFAULT_MAX_PKT_SIZE
 	 */
 	public Packetizer( TransportData transport, URL uri, Resources resources )
 	{
 		this( transport, uri.getIntegerTerm( MAX_PKT_SIZE_TERM, DEFAULT_MAX_PKT_SIZE ) );
 	}

 	/**
 	 * Constructs the Packetizer with the specified packet handler and maximum
 	 * packet size.
 	 *
 	 * @param handler the handler to receive the packets.
 	 *
 	 * @param maxPktSize the maximum packet size that will be accepted.
 	 * Must be >= 0. If maxPktSize == 0, there is no limit.
 	 */
 	private Packetizer( TransportData transport, int maxPktSize )
 	{
 		if (maxPktSize < 0)
 			throw new IllegalArgumentException( "maxPktSize < 0" );

 		this.transport = transport;
 		this.maxPktSize = maxPktSize;

 		transport.setSession( this );
 	}

 	private final TransportData transport;

 	private final int maxPktSize;

 	/**
 	 * @return the transport.
 	 */
 	public TransportData getTransport()
 	{
 		return transport;
 	}

 	@Override
 	public String toString()
 	{
 		return String.format( "Packetizer/%s", transport );
 	}

 	public void sessionData( Who sender, FlexBuffer buf ) throws Exception
 	{
 		// there are two options here. one is that we have no buffered data
 		// and the entire packet is contained within the buf. in that case
 		// we could optimize the daylights out of the process and directly
 		// drop the packet on the handler.

 		while (buf.avail() > 0)
 		{
 			if (wantHeader)
 			{
 				// do we have enough to make a header?

 				if (savedBuf.length() + buf.avail() >= HEADER_SIZE)
 				{
 					int pktSize;

 					if (savedBuf.length() == 0)
 					{
 						// savedBuf is empty, entire header in buf.

 						pktSize = processHeader( buf, false );
 					}
 					else // header split across savedBuf and buf
 					{
 						// move just enough data from buf to savedBuf to have a header.

 						int needFromBuf = HEADER_SIZE - savedBuf.length();
 						savedBuf.put( buf, needFromBuf );
 						savedBuf.setIndex( 0 );

 						pktSize = processHeader( savedBuf, true );
 					}

 					if (pktSize == 0)
 						continue;

 					bodyLen = pktSize;
 					wantHeader = false;
 				}
 				else // want header, but there's not enough to make it.
 				{
 					// save buf in savedBuf.

 					savedBuf.setIndex( savedBuf.length() );
 					savedBuf.put( buf );
 				}
 			}
 			else if (savedBuf.length() + buf.avail() >= bodyLen)
 			{
 				// want body, and there's enough to make it.

 				// three possible cases: the body is entirely in savedBuf,
 				// the body is split, or the body is entirely in buf. assert
 				// that the body cannot entirely be in savedBuf, or else
 				// we'd have processed it last time.
 				assert savedBuf.length() < bodyLen;

 				if (savedBuf.length() == 0)
 				{
 					// savedBuf is empty, entire body in buf.

 					int length = buf.length();
 					int index = buf.index();
 					buf.setLength( index+bodyLen );

 					session.sessionPacket( sender, buf );

 					buf.setLength( length );
 					buf.setIndex( index+bodyLen );

 					wantHeader = true;
 				}
 				else // body split across savedBuf and buf
 				{
 					// move just enough data from buf to savedBuf to have a body.

 					int needFromBuf = bodyLen - savedBuf.length();
 					savedBuf.put( buf, needFromBuf );
 					savedBuf.setIndex( 0 );

 					session.sessionPacket( sender, savedBuf );

 					savedBuf.reset();
 					wantHeader = true;
 				}
 			}
 			else // want body, but there's not enough to make it.
 			{
 				// save buf in savedBuf.

 				savedBuf.put( buf );
 			}
 		}
 		// buf is now empty, and there's nothing else to do.
 		assert buf.avail() == 0;
 	}

 	private int processHeader( FlexBuffer buf, boolean reset ) throws IOException
     {
 		int sig = buf.getInt();
 		if (sig != SIG)
 			throw new IOException( "bad SIG" );

 		int pktSize = buf.getInt();

 		if (reset)
 			buf.reset();

 		if (pktSize < 0 || maxPktSize > 0 && pktSize > maxPktSize)
 			throw new IOException( "pktSize < 0 || (maxPktSize > 0 && pktSize > maxPktSize)" );

 		return pktSize;
     }

 	private boolean wantHeader = true;

 	private int bodyLen;

 	private final FlexBuffer savedBuf = new FlexBuffer();

 	public void transportPacket( Who recipient, FlexBuffer buf ) throws Exception
 	{
 		// data-ize the packet.

 		// assert index is at the start of the header.
 		int dataSize = buf.avail();
 		if (dataSize < HEADER_SIZE)
 			throw new IllegalArgumentException( "dataSize < HEADER_SIZE" );

 		int pktSize = dataSize - HEADER_SIZE;
 		if (maxPktSize > 0 && pktSize > maxPktSize)
 			throw new IOException( "maxPktSize > 0 && pktSize > maxPktSize" );

 		int index = buf.index();
 		buf.putInt( SIG );
 		buf.putInt( pktSize );
 		buf.setIndex( index );
 		transport.transportData( recipient, buf );
 	}

 	public int headerSize()
 	{
 		return HEADER_SIZE;
 	}

 	public Object sessionQuery( Object query ) throws Exception
 	{
 		return session.sessionQuery( query );
 	}

 	public void sessionControl( Object control, Object value ) throws Exception
 	{
 		session.sessionControl( control, value );
 	}

 	public void sessionNotify( Object event ) throws Exception
 	{
 		session.sessionNotify( event );
 	}

 	////////////////////
 	// Source methods //
 	////////////////////

 	public Object transportQuery( Object query ) throws Exception
 	{
 		return transport.transportQuery( query );
 	}

 	public void transportControl( Object control, Object value ) throws Exception
 	{
 		transport.transportControl( control, value );
 	}

 	public void transportNotify( Object event ) throws Exception
 	{
 		transport.transportNotify( event );
 	}

 	public SessionPacket getSession()
 	{
 		return session;
 	}

 	public void setSession( SessionPacket session )
 	{
 		this.session = session;
 	}

 	private SessionPacket session;
 }
	/* $Id$
	*
	* 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.etch.util.core.io;

	import java.io.IOException;

	import org.apache.etch.util.FlexBuffer;
	import org.apache.etch.util.Resources;
	import org.apache.etch.util.URL;
	import org.apache.etch.util.core.Who;


	/**
	* Packetizes a stream data source. Reads a packet header:
	* a 32-bit flag and a 32-bit length, both big-endian, verifies
	* the flag, and then, using the length from the header,
	* reads the packet data and passes it to the packet handler.
	* As a packet source, accepts a packet and prepends a packet
	* header to it before delivering it to a data source.
	*/
	public final class Packetizer implements SessionData, TransportPacket
	{
	/**
	* URI term to specify max packet size.
	*/
	public final static String MAX_PKT_SIZE_TERM = "Packetizer.maxPktSize";

	private final static int SIG = 0xdeadbeef;

	private final static int HEADER_SIZE = 8;

	/**
	* The default maximum packet size that will be accepted, 16376 bytes.
	*/
	public final static int DEFAULT_MAX_PKT_SIZE = 16384 - HEADER_SIZE;

	/**
	* Constructs the Packetizer with null packet handler and uri specified
	* or defaulted maximum packet size.
	* @param transport used to deliver data to the peer.
	* @param uri
	* @param resources
	*/
	public Packetizer( TransportData transport, String uri, Resources resources )
	{
	this( transport, new URL( uri ), resources );
	}

	/**
	* Constructs the Packetizer with null packet handler and uri specified
	* or defaulted maximum packet size.
	*
	* @param transport used to deliver data to the peer.
	* @param uri
	* @param resources
	*
	* @see #MAX_PKT_SIZE_TERM
	* @see #DEFAULT_MAX_PKT_SIZE
	*/
	public Packetizer( TransportData transport, URL uri, Resources resources )
	{
	this( transport, uri.getIntegerTerm( MAX_PKT_SIZE_TERM, DEFAULT_MAX_PKT_SIZE ) );
	}

	/**
	* Constructs the Packetizer with the specified packet handler and maximum
	* packet size.
	*
	* @param handler the handler to receive the packets.
	*
	* @param maxPktSize the maximum packet size that will be accepted.
	* Must be >= 0. If maxPktSize == 0, there is no limit.
	*/
	private Packetizer( TransportData transport, int maxPktSize )
	{
	if (maxPktSize < 0)
	throw new IllegalArgumentException( "maxPktSize < 0" );

	this.transport = transport;
	this.maxPktSize = maxPktSize;

	transport.setSession( this );
	}

	private final TransportData transport;

	private final int maxPktSize;

	/**
	* @return the transport.
	*/
	public TransportData getTransport()
	{
	return transport;
	}

	@Override
	public String toString()
	{
	return String.format( "Packetizer/%s", transport );
	}

	public void sessionData( Who sender, FlexBuffer buf ) throws Exception
	{
	// there are two options here. one is that we have no buffered data
	// and the entire packet is contained within the buf. in that case
	// we could optimize the daylights out of the process and directly
	// drop the packet on the handler.

	while (buf.avail() > 0)
	{
	if (wantHeader)
	{
	// do we have enough to make a header?

	if (savedBuf.length() + buf.avail() >= HEADER_SIZE)
	{
	int pktSize;

	if (savedBuf.length() == 0)
	{
	// savedBuf is empty, entire header in buf.

	pktSize = processHeader( buf, false );
	}
	else // header split across savedBuf and buf
	{
	// move just enough data from buf to savedBuf to have a header.

	int needFromBuf = HEADER_SIZE - savedBuf.length();
	savedBuf.put( buf, needFromBuf );
	savedBuf.setIndex( 0 );

	pktSize = processHeader( savedBuf, true );
	}

	if (pktSize == 0)
	continue;

	bodyLen = pktSize;
	wantHeader = false;
	}
	else // want header, but there's not enough to make it.
	{
	// save buf in savedBuf.

	savedBuf.setIndex( savedBuf.length() );
	savedBuf.put( buf );
	}
	}
	else if (savedBuf.length() + buf.avail() >= bodyLen)
	{
	// want body, and there's enough to make it.

	// three possible cases: the body is entirely in savedBuf,
	// the body is split, or the body is entirely in buf. assert
	// that the body cannot entirely be in savedBuf, or else
	// we'd have processed it last time.
	assert savedBuf.length() < bodyLen;

	if (savedBuf.length() == 0)
	{
	// savedBuf is empty, entire body in buf.

	int length = buf.length();
	int index = buf.index();
	buf.setLength( index+bodyLen );

	session.sessionPacket( sender, buf );

	buf.setLength( length );
	buf.setIndex( index+bodyLen );

	wantHeader = true;
	}
	else // body split across savedBuf and buf
	{
	// move just enough data from buf to savedBuf to have a body.

	int needFromBuf = bodyLen - savedBuf.length();
	savedBuf.put( buf, needFromBuf );
	savedBuf.setIndex( 0 );

	session.sessionPacket( sender, savedBuf );

	savedBuf.reset();
	wantHeader = true;
	}
	}
	else // want body, but there's not enough to make it.
	{
	// save buf in savedBuf.

	savedBuf.put( buf );
	}
	}
	// buf is now empty, and there's nothing else to do.
	assert buf.avail() == 0;
	}

	private int processHeader( FlexBuffer buf, boolean reset ) throws IOException
	{
	int sig = buf.getInt();
	if (sig != SIG)
	throw new IOException( "bad SIG" );

	int pktSize = buf.getInt();

	if (reset)
	buf.reset();

	if (pktSize < 0 \|\| maxPktSize > 0 && pktSize > maxPktSize)
	throw new IOException( "pktSize < 0 \|\| (maxPktSize > 0 && pktSize > maxPktSize)" );

	return pktSize;
	}

	private boolean wantHeader = true;

	private int bodyLen;

	private final FlexBuffer savedBuf = new FlexBuffer();

	public void transportPacket( Who recipient, FlexBuffer buf ) throws Exception
	{
	// data-ize the packet.

	// assert index is at the start of the header.
	int dataSize = buf.avail();
	if (dataSize < HEADER_SIZE)
	throw new IllegalArgumentException( "dataSize < HEADER_SIZE" );

	int pktSize = dataSize - HEADER_SIZE;
	if (maxPktSize > 0 && pktSize > maxPktSize)
	throw new IOException( "maxPktSize > 0 && pktSize > maxPktSize" );

	int index = buf.index();
	buf.putInt( SIG );
	buf.putInt( pktSize );
	buf.setIndex( index );
	transport.transportData( recipient, buf );
	}

	public int headerSize()
	{
	return HEADER_SIZE;
	}

	public Object sessionQuery( Object query ) throws Exception
	{
	return session.sessionQuery( query );
	}

	public void sessionControl( Object control, Object value ) throws Exception
	{
	session.sessionControl( control, value );
	}

	public void sessionNotify( Object event ) throws Exception
	{
	session.sessionNotify( event );
	}

	////////////////////
	// Source methods //
	////////////////////

	public Object transportQuery( Object query ) throws Exception
	{
	return transport.transportQuery( query );
	}

	public void transportControl( Object control, Object value ) throws Exception
	{
	transport.transportControl( control, value );
	}

	public void transportNotify( Object event ) throws Exception
	{
	transport.transportNotify( event );
	}

	public SessionPacket getSession()
	{
	return session;
	}

	public void setSession( SessionPacket session )
	{
	this.session = session;
	}

	private SessionPacket session;
	}