packetchannel.java

Examples to create your Conferencing System in .NET, C# VOIP & Video Conferencing Systems using H.32

/*
 * Copyright 2004 WIT-Software, Lda. 
 * - web: http://www.wit-software.com 
 * - email: [e-mail]
 *
 * All rights reserved. Relased under terms of the 
 * Creative Commons' Attribution-NonCommercial-ShareAlike license.
 */
package handlers;

import io.ProtocolDecoder;
import io.SelectorThread;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

/**
 * Uses non-blocking operations to read and write from a socket. Internally,
 * this class uses a selector to receive read and write events from the
 * underlying socket.  
 *  
 * Methods on this class should be called only by the selector's thread 
 * (including the constructor). If necessary, use Selector.invokeLater() 
 * to dispatch a invocation to the selector's thread.
 * 
 * @author Nuno Santos
 */
final public class PacketChannel implements ChannelListener {
	/** The socket where read and write operations are performed. */
	private final Channel channel;
  /** Used for reading from the socket. */
  private ByteBuffer inBuffer;
  /**
   * The buffer with the packet currently being sent. 
   * This class can only send one packet at a time, there are no
   * queueing mechanisms.
   */
  private ByteBuffer outBuffer = null;
  
  /**
   * Used to convert raw bytes into packets. 
   * (Strategy design pattern)
   */
  private final ProtocolDecoder protocolDecoder;

  /**
   * Object interested in the events generated by this class.
   * It is notified whenever an error occurs or a packet is read. 
   */
  private final PacketChannelListener listener;
  
  /**
   * Creates and initializes a new instance. Read interest is enabled
   * by the constructor, so callers should be ready to star receiving
   * packets.
   * 
   * @param socketChannel Socket to be wrapped.
   * @param selector Selector to be used for managing IO events.
   * @param listener Object to receive the callbacks.
   * @param clientMode Just for secure sockets. True if this socket
   * was established as client, false if it was established as a server.
   * @param protocolDecoder Decoder for reassembling the packets.
   * @throws IOException
   */
  public PacketChannel(SocketChannel sc,
  		ChannelFactory channelFactory,
			SelectorThread st,
			ProtocolDecoder protocolDecoder,
			PacketChannelListener listener) 
  throws Exception {
  	
  	this.protocolDecoder = protocolDecoder;
  	this.channel = channelFactory.createChannel(sc, st, this);
    this.listener = listener;
    
    // Creates the reading buffer
    // The size is the same as the size of the TCP sockets receive buffer.
    // We will never read more than that at a time.
//    inBuffer = ByteBuffer.allocateDirect(sc.socket().getReceiveBufferSize()-10);
    inBuffer = ByteBuffer.allocate(64*1024);
    
    // Quick and dirty hack. When a buffer is created by the first time
    // it is empty, with 
    inBuffer.position(inBuffer.capacity());
    
//    inBuffer.flip();
  }
  
  /**
   * Activates reading from the socket. This method is non-blocking. 
   */
  public void resumeReading() throws IOException {
    processInBuffer();
  }
  
  public void close() {
    try {
      channel.close();
    } catch (IOException e) {
    	System.out.println("PacketChannel.close(): ");
    	e.printStackTrace();
      // Ignore
    }
  }
  
  /**
   * Reads from the socket into the internal buffer. This method should 
   * be called only from the SelectorThread class.
   */
  public void handleRead() {
    try {      
      // Reads from the socket      
      // Returns -1 if it has reached end-of-stream
      int readBytes = channel.read(inBuffer);      
//      System.out.println("Read: " + readBytes);
      // End of stream???
      if (readBytes == -1) {
      	System.out.println("End of stream???");
        // End of stream. Closing channel...
        close();
        listener.socketDisconnected(this);
        return;
      }
      
      // Nothing else to be read?
      if (readBytes == 0) {        
        // There was nothing to read. Shouldn't happen often, but 
        // it is not an error, we can deal with it. Ignore this event
        // and reactivate reading.
        reactivateReading();
        return;
      }
      
      // There is some data in the buffer. Processes it.
      inBuffer.flip();
      processInBuffer();
    } catch (IOException ex) {
      // Serious error. Close socket.
      listener.socketException(this, ex);
      close();
    }
  }
  
  /**
   * Processes the internal buffer, converting it into packets if enough
   * data is available.
   */
  private void processInBuffer() throws IOException {    
    ByteBuffer packet = protocolDecoder.decode(inBuffer);
    // A packet may or may not have been fully assembled, depending
    // on the data available in the buffer
    if (packet == null) {      
      // Partial packet received. Must wait for more data. All the contents
      // of inBuffer were processed by the protocol decoder. We can
      // delete it and prepare for more data.
      inBuffer.clear();
      reactivateReading();
    } else {      
      // A packet was reassembled.
      listener.packetArrived(this, packet);
      // The inBuffer might still have some data left. Perhaps
      // the beginning of another packet. So don't clear it. Next
      // time reading is activated, we start by processing the inBuffer
      // again.
    }
  }

  /**
   * Disable interest in reading.
   * @throws IOException 
   */
  public void disableReading() throws IOException {
  	channel.unregisterForRead();        
  }
  
  /**
   * Enables interest in reading.
   * @throws IOException
   */
  private void reactivateReading() throws IOException {
  	channel.registerForRead();
//    selector.addChannelInterestNow(channel, SelectionKey.OP_READ);
  }
  
  /**
   * Sends a packet using non-blocking writes. One packet cannot be sent 
   * before the previous one has been dispatched. The caller must ensure 
   * this. 
   * 
   * This class keeps a reference to buffer given as argument while sending
   * it. So it is important not to change this buffer after calling this 
   * method.  
   * 
   * @param packet The packet to be sent. 
   */
  public void sendPacket(ByteBuffer packet) {
    // keeps a reference to the packet. In production code this should copy
    // the contents of the buffer.
    outBuffer = packet;
    handleWrite();
  }

  /**
   * Activates interest in writing.
   * 
   * @throws IOException
   */
  private void requestWrite() throws IOException {
  	channel.registerForWrite();
//    selector.addChannelInterestNow(channel, SelectionKey.OP_WRITE);
  }
  
  /**
   * Writes to the underlying channel. Non-blocking. This method is called
   * only from sendPacket() and from the SelectorThread class.
   */
  public void handleWrite() {    
    try {
      // Writes to the socket as much as possible. Since this is a 
      // non-blocking operation, we don't know in advance how many
      // bytes will actually be written.
      int written = channel.write(outBuffer);
//      System.out.println("Wrote: " + written);
      // Check if there are more to be written. 
      if (outBuffer.hasRemaining()) {
        // There is. Reactivate interest in writing. We will try again 
        // when the socket is ready.
        requestWrite();
      } else {
        // outBuffer was completly written. Notifies listener
        ByteBuffer sentPacket = outBuffer;
        outBuffer = null;
        listener.packetSent(this, sentPacket);
      }
    } catch (IOException ioe) {
      close();
      listener.socketException(this, ioe);
    }
  }
  
  public SocketChannel getSocketChannel() {
    return channel.getSocketChannel();
  }
  
  public String toString() {  
    return Integer.toString(channel.getSocketChannel().socket().getLocalPort());
  }
}