clientcontroller.java

MilGra0.8b for java media server

/*

   Milenia Grafter Server
  
   Copyright (c) 2007-2008 by Milan Toth. All rights reserved.
  
   This program is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   of the License, or (at your option) any later version.
  
   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.
  
   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA. 
   
*/ 

package com.milgra.server;

/**

	ClientController class
	
	@mail [e-mail]
	@author Milan Toth
	@version 20080315

	Tasks of Clientcontroller
	
		- Client initialization
		- Invoke and RTMP message flow control

**/

import com.milgra.server.api.Client;
import com.milgra.server.api.Stream;
import com.milgra.server.api.Wrapper;
import com.milgra.server.api.WrapperMap;
import com.milgra.server.api.WrapperList;
import com.milgra.server.api.InvokeEvent;
import com.milgra.server.api.StatusEvent;
import com.milgra.server.api.IApplication;
import com.milgra.server.api.EventListener;

import com.milgra.server.encoder.Encoder;
import com.milgra.server.encoder.RtmpFactory;

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

import java.util.HashMap;
import java.util.ArrayList;


public class ClientController extends OProcess
{
	
	// static id to make client id creation atomic and global
	
	public static long globalId = 0;			

	// client - related client api object
	// application - related custom application instance
	
	public Client client;
	public IApplication application;

	// closed - controller is closed
	// accepted - controller is accepted
	
	public boolean closed;
	public boolean accepted;
	
	// counter - rtmo update counter
	// id - unique id
	// mode - client mode, active or passive
	
	public int counter;
	public int stepRound;
	public long id;
	public long currentTime;
	public String mode;
	
	public String ip;
	public String agent;
	public String referrer;

	public long ping;
	public long bytesIn = 0;
	public long bytesOut = 0;
	
	public double bandIn = 0;
	public double bandOut = 0;

	// readStepClient - how much bytes to receive before next read message
	// readStepServer - how much bytes to send before next read message from client

	public int readStepClient;
	public int readStepServer;						
	
	// lastRead - last read bytes
	// lastBand - last bandwidth
	// lastPing - last ping send
	// lastUpdate - last update message
	// lastPingRead - last ping received
	
	public long lastRead;
	public long lastBand;
	public long lastPing;						
	public long lastPingRead;

	// lastBytesIn - last received amount for bw calculus
	// lastBytesOut - last sent amout for bw calculus

	public long lastBytesIn;
	public long lastBytesOut;					
	
	// event listeners
	
	public EventListener streamListener;
	public EventListener invokeListener;
	public EventListener statusListener;
	
	// controller
	
	public SocketController socketController;
	public StreamController streamController;
	
	// packet containers
	
	public ArrayList < RtmpPacket > incomingList;
	public ArrayList < RtmpPacket > outgoingList;
	
	// container for incoming invoke channels and ids

	public HashMap < Double , String > incomingInvokes;
	public HashMap < Double , String > outgoingInvokes;
	
	/**
	 * ClientController constructor
	 */
	
	public ClientController (  )
	{
		
		// System.out.println( System.currentTimeMillis() + " ClientController.setApplication" );
		
		// create
		
		outgoingList = new ArrayList < RtmpPacket > ( );
		incomingList = new ArrayList < RtmpPacket > ( );

		incomingInvokes = new HashMap < Double , String > ( );
		outgoingInvokes = new HashMap < Double , String > ( );

		socketController = new SocketController( this );
		streamController = new StreamController( this , socketController );

		// set

		// id - generate unique identifier for client
		// mode - default mode is passive
		// counter - update time counter
		// closed - controller not closed by default
		// accepted - controller is not accepted yet
		
		id = globalId++;							
		mode = "passive";				
		counter = 0;
		stepRound = Math.round( 1000 / Library.STEPTIME );
		closed = false;					
		accepted = false;				
		
		// read stepping is 250000 by default
		
		readStepClient = 250000;					
		readStepServer = 250000;				
		
		// init byte counters
		
		lastRead = 0;
		lastBytesIn = 0;
		lastBytesOut = 0;
		
		// start
				
		// invoke channels must be over 0
			
		incomingInvokes.put( ( double ) 0 , null );	

	}
	
	/**
	 * Sets host application for an active client
	 * @param customX
	 */
	
	public void setApplication ( IApplication customX )
	{
		
		// System.out.println( System.currentTimeMillis() + " " + id + " ClientController.setApplication" );
		
		// mode swithces to active
		
		mode = "active";							
		
		// store host application
		
		application = customX;						
		
		// pair client with application
		
		Server.registerClient( customX , this );	
			
	}
	
	/**
	 * Connects controller to an opened socket passed by ServerSocketConnector
	 * @param socketX
	 */
	
	public void connect ( SocketChannel channelX )
	{
		
		// System.out.println( System.currentTimeMillis( ) + " " + id + " ClientController.connect " +	channelX + " " + closed);
	
		if ( !closed )
		{
			
			// connecting socketcontroller to our socketchannel
		
			socketController.connect( channelX , mode );				
			
			// registering processes
			
			Server.registerProcess( this , "client" );					
			Server.registerProcess( socketController , "socket" );
			Server.registerProcess( streamController , "stream" );
			
		}
		
	}
	
	/**
	 * Connects controller to a remote server with a simple argument
	 * @param urlX remote server's url
	 * @param argumentsX connection arguments
	 */
	
	public void connect ( String urlX , Wrapper argumentX ) 
	{
		
		// System.out.println( System.currentTimeMillis( ) + " " + id + " ClientController.connect " + urlX );
		
		connect( urlX , new WrapperList( argumentX ) ); 
	
	}

	/**
	 * Connects controller to a remote server with wrapperlist argument
	 * @param urlX remote server's url
	 * @param argumentsX connection arguments
	 */
	
	public void connect ( String urlX , WrapperList argumentsX )
	{
	
		// System.out.println( System.currentTimeMillis( ) + " " + id + " ClientController.connect " + urlX );
			
		if ( !closed )
		{
			
			// url -  part after rtmp://
			// host - host address
			// appid - application id
			
			String url = urlX.substring( 7 );
			String host = url.split( "/" )[0 ];
			String appid = url.split( "/" )[1 ];				

			// invokeChannel - new inckeChannel
			// info - invoke info message
			// message - invoke rtmp packet
			
			double invokeChannel = incomingInvokes.size( );		
			WrapperMap info = new WrapperMap( Library.CONNECTKEYS , Library.MACPLAYERARR );
			RtmpPacket message = new RtmpPacket( );
			WrapperList arguments = new WrapperList( );
			
			// store application and tcUrl
			
			info.put( "app" , appid );
			info.put( "tcUrl" , urlX );
			
			// message creation, no null needed after invokechannel at connect

			arguments.add( new Wrapper( "connect" ) );
			arguments.add( new Wrapper( invokeChannel ) );
			arguments.add( new Wrapper( info ) );
			arguments.addAll( argumentsX );
			
			// rtmp message
			
			message.bodyType = 0x14;
			message.rtmpChannel = 0x03;
			message.body = Encoder.encode( arguments );
			
			// put packet on outgoing list 
			
			addOutgoingPacket( message );
			
			// invokeChannel 1 is reserved for connect
			
			incomingInvokes.put( invokeChannel , "connect" );
			
			// start connecting
			
			Server.socketConnector.connect( host , this );
		
		}
		
	}
		
	/**
	 * Channel connection failed
	 */
	
	public void connectFailed ( )
	{
		
		// System.out.println( System.currentTimeMillis() + " " + id + " ClientController.connectFailed " );
		
		// status info
		
		WrapperMap info = new WrapperMap( Library.STATUSKEYS , Library.FAILUREARR );
		
		// dispatch status event
		
		if ( statusListener != null ) statusListener.onEvent( new StatusEvent( info , client ) );
		
		// close controller
		
		close( );
		
	}
	
	/**
	 * Closes clientcontroller, cleanup
	 */
	
	public synchronized void close ( )
	{
		
		// System.out.println( System.currentTimeMillis() + " " + id + " ClientController.close " );
		
		// we are closed, no more invokes allowed
		
		closed = true;														
		
		// if streamcontroller isn't closed
		// if close isn't triggered by socketcontroller
		
		if ( !streamController.closed ) streamController.close( );			
		if ( !socketController.closed ) socketController.close( "detach" );	
		
		// if we had an application
		
		if ( application != null )
		{
			
			// if we were passive, telling application that we left
			// if we were active, dispatch event to status listener
			
			if ( mode.equals( "passive" ) ) dispatchLeave( );						
			if ( mode.equals( "active" ) ) dispatchClosure( );			
			
		}

		// unpair client with application
		
		Server.unregisterClient( application , this );						

		// unregister processes
		
		Server.unregisterProcess( this , "client" );						
		Server.unregisterProcess( socketController , "socket" );
		Server.unregisterProcess( streamController , "stream" );
	
	}
	
	/**
	 * Dispathes leave event to paired application
	 */
	
	public void dispatchLeave ( )
	{
		
		// System.out.println( System.currentTimeMillis() + " " + id + " ClientController.dispatchLeave " );
		
		// Syncing is needed, multiple clients from multiple threads can disconnect at the same time
		
		synchronized ( application ) 
		{