navigatorimpl.java~1~

移动Agent编程工具Naplet

/*
 * @<#>NavigatorImpl.java version 0.0.1, 1/1/2000
 *
 * THIS PROGRAM IS FREE SOFTWARE; YOU CAN DISTRIBUTE IT AND/OR
 * MODIFY IT UNDER THE TERMS OF THE GNU GENERAL PUBLIC LICENSE
 * AS PUBLISHED BY THE FREE SOFTWARE FOUNDATION.
 *
 * 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.
 *
 * Copyright (c) 2000 Wayne State University. All Rights Reserved.
 */
package naplet.server;

import java.net.*;
import java.rmi.*;
import java.util.*;
import naplet.*;
import naplet.directory.*;
import naplet.message.*;
import naplet.nsocket.*;

/**
 * The <code>NavigatorImpl</code> class implements the methods
 * of <code>NapletServer</code>.
 *
 * @version 0.0.1, 1/1/2000
 * @author C. Xu
 */
public class NavigatorImpl
    implements Navigator, Runnable
{
    private static NavigatorImpl instance = null;
    private ThreadGroup dispatchThreadGroup;
    private ServerProperty property;
    protected NavigatorImpl( ServerProperty property )
    {
        this.property = property;
    }

    public static NavigatorImpl getInstance( ServerProperty property )
    {
        if ( instance == null )
        {
            instance = new NavigatorImpl( property );
        }
        return instance;
    }

    public void run()
    {
        dispatchThreadGroup = new ThreadGroup( "Dispatch Handler Main" );
    }

    /**
     * Receive a naplet from remote server
     *
     * @param nap Naplet to land
     * @param src The naplet comes from
     */
    void land( Naplet nap, InetAddress src )
    {
        Hashtable socketTable = nap.getSocket();
        Hashtable serverSocketTable = nap.getServerSocket();

        //update socket table
        SocketController.setSocketTable( socketTable );
        SocketController.setServerSocketTable( serverSocketTable );

        for ( Enumeration enum = serverSocketTable.keys();
              enum.hasMoreElements(); )
        {
            String s = ( String ) enum.nextElement();
            NapletServerSocket serversocket =
                ( NapletServerSocket ) serverSocketTable.get( s );

            // must be persistent if in this table
            serversocket.resume();
        }

        for ( Enumeration enum1 = socketTable.keys();
              enum1.hasMoreElements(); )
        {
            String s1 = ( String ) enum1.nextElement();
            NapletSocket nsocket = ( NapletSocket ) socketTable.get( s1 );
            try
            {
                // must be persistent if in this table
                if ( nsocket.isDualMigration() )
                {
                    nsocket.halfResume();
                }
                else
                {
                    nsocket.resume();
                }
            }
            catch ( Exception exc )
            {
                exc.printStackTrace();
            }
            socketTable.remove( s1 );
        }

        NapletDirectory directory = property.getNapletDirectory();
        if ( directory != null )
        {
            try
            {
                directory.register( nap.getNapletID(), 
				    property.getServerURN(),
                                    new Date(), NapletEvent.ARRIVE );
            }
            catch ( RemoteException dae )
            {
                throw new NapletInternalError(
                    "Unable to register the naplet arrival in the directory" );
            }
        }

        if ( nap.isLongLived() )
        {
            Messenger messenger = property.getMessenger();
            messenger.openMailBox( nap.getNapletID() );
        }

        NapletMonitor monitor = property.getNapletMonitor();
        monitor.join( nap );

    }

    /**
     * Called to dispatch a naplet, using the default port and server name,
     * to a naplet server.
     *
     * @param dest hostname of a remote naplet server
     * @param naplet naplet to be dispatched
     */
    public void toDispatch( String dest, Naplet naplet )
        throws RemoteException
    {
        try
        {
            int port = property.getServerURN().getPort();
            String name = property.getServerURN().getServerName();
            URN ser = new URN( dest, port, name );
            toDispatch( ser, naplet );
        }
        catch ( InvalidURNException iue )
        {
            System.out.println( "Invalid URN" );
            System.exit( 1 );
        }
    }

    /**
     * Called to dispatch a naplet to a naplet server
     *
     * @param dest remote naplet server address
     * @param naplet naplet to be dispatched
     */
    public void toDispatch( URN dest, Naplet naplet )
        throws RemoteException
    {

        NapletID nid = naplet.getNapletID();

        //socketController.stopControlThread();

        // begin to add
        NapletServerSocket serversocket;
        Hashtable serverSocketTable = SocketController.getServerSocketTable();
        Hashtable newServerSockTable = new Hashtable();

        for ( Enumeration enum = serverSocketTable.keys();
              enum.hasMoreElements(); )
        {
            String s = ( String ) enum.nextElement();
            serversocket = ( NapletServerSocket ) serverSocketTable.get( s );
            String owner = serversocket.getNapletID();

            // go to the next one if not owner for this naplet
            if ( ( owner != null ) 
		 && ( !owner.equals( nid.toString() ) ) )
            {
                continue;
            }
            newServerSockTable.put( s, serversocket );
            // only suspend if is persistent
            try
            {
                if ( serversocket.isPersistent() )
                {
                    serversocket.suspend();
                }
            }
            catch ( Exception ex )
            {
                ex.printStackTrace();
            }
        }
        // record serversocket table
        naplet.setServerSocket( newServerSockTable );

        Hashtable socketTable = SocketController.getSocketTable();
        Hashtable newSockTable = new Hashtable();
        for ( Enumeration enum = socketTable.keys();
              enum.hasMoreElements(); )
        {
            String s1 = ( String ) enum.nextElement();
            NapletSocket nsocket = ( NapletSocket ) socketTable.get( s1 );

            String owner = nsocket.getSrcNapletID(); // client side
            if ( owner == null )
            {
                owner = nsocket.getDestNapletID(); // server side
            }
            if ( ( owner != null ) && ( !owner.equals( nid.toString() ) ) )
            {
                //System.out.println("!!@@not match:"+nid+",with id:"+owner);
                continue;
            }
            // only suspend the connection which is from the naplet id
            newSockTable.put( s1, nsocket );
            try
            {
                if ( nsocket.isPersistent() )
                {
                    nsocket.suspend();

                }
            }
            catch ( Exception exception )
            {
                exception.printStackTrace();
            }
        }

        // record socket table
        naplet.setSocket( newSockTable );

        // end of add

        System.out.println( nid.toString() + " is to leave" );

        //
        // An option to dispatch naplets in parallel
        //
        new Thread( dispatchThreadGroup,
                    new DispatchProxy( dest, naplet ) ).start();

        /*
         * An option to dispatch naplets in sequence
         *
           try {
         NapletServer ns = (NapletServer)
           Naming.lookup("rmi://"+dest.toString());
         ns.dispatch( naplet );
           } catch (Exception e) { e.printStackTrace(); }
         */

        NapletDirectory dir = property.getNapletDirectory();
        if ( dir != null )
        { // Directory service is enabled
            dir.register( nid, property.getServerURN(), new Date(),
                          NapletEvent.DEPART );
        }

        property.getNapletServerLog().depart( naplet, dest );

        property.getMessenger().removeMailBox( nid );

    }

    /**
     * <The code>DispatchProxy</code> class implements a dispatching thread
     * that help dispatch naplets to remote servers
     */
    private class DispatchProxy
        implements Runnable
    {
        private URN dest;
        private Naplet naplet;
        DispatchProxy( URN dest, Naplet naplet )
        {
            this.dest = dest;
            this.naplet = naplet;
        }

        public void run()
        {

            try
            {
                NapletServer ns = ( NapletServer )
                    Naming.lookup( "rmi://" + dest.toString() );
                ns.dispatch( naplet );
            }
            catch ( Exception e )
            {
                // throw new UnableDispatchException();
                System.out.println( "Unable to  dispatch naplets to "
                                    + dest.toString() );
                System.out.println( e.getMessage() );
                e.printStackTrace();
            }
        }
    }
}