socketcontroller.java

移动Agent编程工具Naplet

            return;
        }
        for ( Enumeration enum = table.keys();
              enum.hasMoreElements(); )
        {

            String key = ( String ) enum.nextElement();
            if ( serverSocketTable.containsKey( key ) )
            {
                log( "wrong update svr sk tbl aft land:already has the key" );
                serverSocketTable.remove( key );
            }
            serverSocketTable.put( key, table.get( key ) );
        }
    }

    /**
     * locate server according to its nid. Locate from directory if available.
     * If not, the user must have set it.
     * @param nid
     * @return
     */
    private static String locateNaplet( NapletID nid )
    {
        String host = null;
        Locator loc = property.getLocator();
        URN svr = null;
        try
        {
            svr = loc.lookup( nid );
            host = svr.getHostName();
        }
        catch ( NapletLocateException ex )
        {
            NapletMonitor nm = property.getNapletMonitor();
            host = nm.locate( nid );
        }

        return host;
    }

    /**
     * remove napletsocket according to its socketID
     * @param id
     */
    public static void removeSocket( String id )
    {
        if ( id == null )
        {
            throw new NullPointerException( "id" );
        }
        socketTable.remove( id );
    }

    public static void removeServerSocket( String s )
    {
        if ( s == null )
        {
            throw new NullPointerException( "id" );
        }
        serverSocketTable.remove( s );
    }

    /**
     * This part of function has been moved to NavigotorImpl
     *
     *  better add a parameter as NapletID
     */
    public static void suspendAll()
    {
        for ( Enumeration enum = socketTable.getKeys();
              enum.hasMoreElements(); )
        {
            String s1 = ( String ) enum.nextElement();
            NapletSocket nsocket = ( NapletSocket ) socketTable.get( s1 );

            //if (nsocket.getNapletID().equals()
            // only suspend if is persistent
            try
            {
                //if (nsocket.isPersistent())
                nsocket.suspend();
            }
            catch ( Exception exception )
            {
                exception.printStackTrace();
            }
        } // end of for
    }

    /**
     * This part of function has been moved to NavigotorImpl
     *
     * Resume all connection
     */
    public static void resumeAll()
    {
        for ( Enumeration enum1 = socketTable.getKeys();
              enum1.hasMoreElements(); )
        {
            String s1 = ( String ) enum1.nextElement();
            NapletSocket nsocket = ( NapletSocket ) socketTable.get( s1 );
            try
            {
                // only need to resume if is persistent
                //if(nsocket.isPersistent())
                nsocket.resume();
            }
            catch ( Exception exc )
            {
                exc.printStackTrace();
            }
        }
    }

    /**
     * Test if the user has the right to create socket/serversocket. It
     * demonstrates the use of subject based security check.
     */
    static void authorize()
    {
        SamplePrincipal userPrincipal = new SamplePrincipal( "testUser" );
        Subject subject = new Subject();
        if ( !subject.getPrincipals().contains( userPrincipal ) )
        {
            subject.getPrincipals().add( userPrincipal );
        }

        PrivilegedAction action = new SampleAction();
        Subject.doAsPrivileged( subject, action, null );

    }

    /**
     * Message handler for those coming from TCP related to socket handoff.
     * Has to be TCP in that case. Others are in UDP.
     */
    public void run()
    {
        if ( instance == null )
        {
            System.out.println( "Please provide the SocketControllerPort " );
            System.out.println(
                "and ControlChannelPort in the configuration file." );
            throw new NullPointerException( "SocketController is not installed" );
        }

        int i = ControlPort;

        // find a free port and start listening. Usually the port
        // defined in as ControlPort should be free.
        do
        {
            if ( controlServer != null )
            {
                break;
            }

            try
            {
                controlServer = new ServerSocket( i );
                log( "listening on port:" + i );
                break;
            }
            catch ( Exception ex )
            {
                i++;
            }
        }
        while ( true );

        while ( !bStop )
        {
            try
            {
                Socket socket1;
                try
                {
                    // keep waiting connection, start another
                    // for waiting if one arrives
                    socket1 = controlServer.accept();

                    Thread t = new Thread(
                        new SocketController(), "control thread" );
                    t.start();
                }
                catch ( Exception ex )
                {
                    continue;
                }

                BufferedReader bufferedreader1 = new BufferedReader( new
                    InputStreamReader( socket1.getInputStream() ) );
                BufferedWriter bufferedwriter1 = new BufferedWriter( new
                    OutputStreamWriter( socket1.getOutputStream() ) );

                while ( !bStop )
                {
                    // keep reading request, only part of message
                    // coming from TCP channel
                    String in = bufferedreader1.readLine();
                    log( "from addr:" + socket1.getInetAddress() +
                         ",get a request:" + in );
                    if ( in == null )
                    {
                        // the other side has exited, so close this
                        // half socket
                        return;
                    }

                    StringTokenizer stringtokenizer
                        = new StringTokenizer( in, ":" );
                    String s1 = stringtokenizer.nextToken();

                    if ( s1.equals( "ConnectMsg" ) )
                    {
                        // ATTN: there is a : in NapletID
                        String napletID = stringtokenizer.nextToken()
                            + ":" + stringtokenizer.nextToken();
                        NapletServerSocket nss = ( NapletServerSocket )
                            nidServerTable.get(
                            napletID );
                        if ( nss == null )
                        {
                            // no server is available so exit with
                            // an error message.
                            System.out.println(
                                "atten(server not up:::::wrong info" +
                                nidServerTable +
                                ", for nid:" + napletID );
                            System.exit( -1 );
                        }

                        String pri = stringtokenizer.nextToken();
                        int intpri = Global.LOW_PRIORITY;
                        if ( pri.equalsIgnoreCase( Integer.toString( Global.
                            LOW_PRIORITY ) ) )
                        {
                            //if client is low, server is high
                            intpri = Global.HIGH_PRIORITY;
                        }

                        // see if should be persistent
                        String persistent = stringtokenizer.nextToken();
                        boolean ifpersist = false;

                        if ( ( persistent != null ) &&
                             ( persistent.equalsIgnoreCase( "true" ) ) )
                        {
                            //nss.setSocket(socket1,true);
                            ifpersist = true;
                        }

                        String ack = Global.ACKMSG;

                        // use diffie-hellman key exchange
                        if ( SocketController.GENKEY )
                        {
                            BigInteger seed = new BigInteger( stringtokenizer.
                                nextToken() );
                            BigInteger y = new BigInteger( Global.KEYSIZE,
                                new Random() );
                            BigInteger pubKey = seed.modPow( y, Global.N );
                            log( "@@@pubkey:" + pubKey );

                            BigInteger tosend
                                = Global.G.modPow( y, Global.N );
                            ack += tosend;
                            nss.setSocket( socket1, ifpersist,
                                           intpri, pubKey );
                        }
                        else
                        {
                            nss.setSocket( socket1, ifpersist, intpri );
                        }

                        bufferedwriter1.write( ack );
                        bufferedwriter1.newLine();
                        bufferedwriter1.flush();

                        return;
                    }
                    else if ( s1.equals( "ResumeMsg" ) )
                    {
                        // socket id
                        String sockID = stringtokenizer.nextToken();
                        NapletSocket nsocket
                            = ( NapletSocket ) socketTable.get( sockID );

                        if ( nsocket == null )
                        {
                            log( "///nsocket is null:" + socketTable +
                                 ",and SID:" + sockID );
                        }

                        // should authenticate with key match
                        if ( SocketController.GENKEY )
                        {
                            BigInteger pubKey
                                = new BigInteger( stringtokenizer.nextToken() );
                            if ( !pubKey.equals( nsocket.getPublicKey() ) )
                            {
                                log( "***resum key not match with socket:" +
                                     nsocket.getPublicKey() );
                            }
                        }

                        if ( nsocket.isSuspendAfterACK() )
                        {
                            // if suspend after ack, it means the
                            // peer has been approved to
                            // migrate. so this message is to resume
                            // the connection because
                            // it has finished migration. send back
                            // a stop_resume message
                            // since the connection will be suspended next.
                            bufferedwriter1.write( Global.ACK_STOP_RESUME );
                        }
                        else
                        {
                            bufferedwriter1.write( Global.ACKMSG );
                        }

                        bufferedwriter1.newLine();
                        bufferedwriter1.flush();

                        nsocket.resume( socket1 );

                        return;
                    }
                } // end of while(true) readline

                bufferedreader1.close();
                bufferedreader1 = null;
                bufferedwriter1.close();
                bufferedwriter1 = null;
                socket1.close();
                socket1 = null;
            }
            catch ( Exception exception1 )
            {
                exception1.printStackTrace();
            }
        }
    } // end of run

    /**
     * Stop control socket thread.
     */
    void stopControlSocketThread()
    {
        bStop = true;

        try
        {
            controlServer.close();
        }
        catch ( Exception exception )
        {
            exception.printStackTrace();
        }
    }

    private static void log( String info )
    {
        if ( debug )
        {
            System.out.println( "SocketController:" + info );
        }
    }

    public class ControlChannel
        implements Runnable
    {
        /**
         * Client side of the message channel. Used to send message.
         */
        private DatagramSocket client;
        /**
         * Server side of the message channel. Used to receive message.
         */
        private DatagramSocket server;
        /**
         * The received message
         */