xfer.java

有关java 的p2p应用,是一般很好的教程,有兴趣的朋友应该好好阅读一下

                return syntaxError();
            }
            
            filename = options.getNextParameter();
            
            if( null == filename ) {
                consoleMessage( "Missing filename" );
                return syntaxError();
            }
            
            if( null != destAddr ) {
                destAddr = new EndpointAddress( destAddr, XFERSERVICENAME, destUserName );
            }
            
            if( (null != destAddr) && !destPIDs.isEmpty() ) {
                consoleMessage( "Cannot specify both a destination endpoint address and destination peers" );
                return syntaxError();
            }

            return sendFile( filename, srcUserName, destUserName, destAddr, blockSize, destPIDs, async );
        } else if( "search".equals( command ) ) {
            return findUserAdvs();
        }
        
        consoleMessage( "Unrecognized Command" );
        return syntaxError();
    }
    
    /**
     *  Register
     **/
    private int registerNewUser( String userName, String type ) {
        
        // Check if there is already a registered user of the sme name.
        print( "# " + getCmdShortName() + " - Searching for existing advertisement for user '" + userName + "'" );
        
        PipeAdvertisement adv = findUserAdv(userName);
        if (adv != null) {
            consoleMessage( "Sorry, user '" + userName + "' is already registered");
            return ShellApp.appMiscError;
        }
        
        try {
            // Create a pipe advertisement for this pipe.
            adv = (PipeAdvertisement) AdvertisementFactory.newAdvertisement(PipeAdvertisement.getAdvertisementType());
        } catch( Exception all ) {
            printStackTrace( "Advertisement document could not be created", all );
            return ShellApp.appMiscError;
        }
        
        adv.setPipeID( IDFactory.newPipeID( (PeerGroupID) group.getPeerGroupID() ) );
        adv.setName( XFERUSERNAME + "." + userName);
        adv.setType(type);
        
        try {
            // Save the document into the public folder
            discovery.publish(adv, DiscoveryService.DEFAULT_LIFETIME, DiscoveryService.DEFAULT_EXPIRATION );
            discovery.remotePublish( adv, DiscoveryService.DEFAULT_EXPIRATION );
        } catch (Exception e2) {
            printStackTrace( "Advertisement could not be published.", e2 );
            return ShellApp.appMiscError;
        }
        
        consoleMessage( "Created advertisement for user '" + userName + "'." );
        
        return ShellApp.appNoError;
    }
    
    /**
     *  Stop the receiver daemon for the specified user name.
     *
     *  @param userName user to log in.
     *  @param endpoint if false, login as an input pipe listener or if true, login as an endpoint listener
     *  @return result code
     **/
    private int login( String userName, boolean endpoint ) {
        PipeAdvertisement adv = null;
        
        if ( isDaemonRunning(userName) ) {
            consoleMessage( "User '" + userName + "' is already listening" );
            return ShellApp.appMiscError;
        }
        
        if( !endpoint ) {
            print( "# " + getCmdShortName() + " - Searching for Advertisement for user '" + userName + "'" );
            adv = findUserAdv( userName );
            if (adv == null) {
                consoleMessage( "User '" + userName + "' is not a registered user");
                return ShellApp.appMiscError;
            }
        }
        
        return runDaemon( userName, adv ) ? ShellApp.appNoError : ShellApp.appMiscError;
    }
    
    /**
     *  Stop the receiver daemon for the specified user name.
     *
     *  @param userName user to log out.
     *  @return result code
     **/
    private int logout( String userName ) {
        
        if (!isDaemonRunning(userName)) {
            consoleMessage( "User '" + userName + "' is not listening");
            return ShellApp.appMiscError;
        }
        
        consoleMessage( "Stoping listener for '" + userName + "'" );
        
        return stopDaemon( userName ) ? ShellApp.appNoError : ShellApp.appMiscError;
    }
    
    /**
     *  Send a file to a remote peer.
     *
     *  @param filename the full local path of the file to send.
     *  @param srcUserName optionally our user name otherwise null.
     *  @param destUserName the name of the destination user.
     *  @param destAddr destination peer for direct endpoint transfers.
     *  @param blockSize block size in bytes to segment file into.
     *  @param async use async messengers to send file segments.
     *  @return result code
     **/
    private int sendFile( String filename, String srcUserName, String destUserName, EndpointAddress destAddr, int blockSize, Set destPIDs, boolean async ) {
        
        XferDaemon srcDaemon = null;
        
        // get the src daemon if any
        if ( null != srcUserName  ) {
            
            srcDaemon = getDaemon( srcUserName );
            
            if( null == srcDaemon ) {
                consoleMessage( "User '" + srcUserName + "' is not logged in.");
                
                return ShellApp.appMiscError;
            }
        }
        
        File f = new File( filename );
        
        if( !f.exists() ){
            consoleMessage( "file '" + filename + "' not found.");
            return ShellApp.appMiscError;
        }
        
        FileSender sender = new FileSender( group, getOutputConsPipe(), srcDaemon, blockSize, async, f );
        
        PipeAdvertisement adv = null;
        
        if( null == destAddr ) {
            // Locate target name
            print( "# " + getCmdShortName() + " - Searching for advertisement for user '" + destUserName + "'." );
            adv = findUserAdv(destUserName);
            if (adv == null) {
                consoleMessage( "Advertisement for user '" + destUserName + "' not found.");
                return ShellApp.appMiscError;
            }
            
            // Try and connect to target
            try {
                consoleMessage( "Found advertisement for user '" + destUserName + "' attempting to connect to " + adv.getType() );
                getGroup().getPipeService().createOutputPipe( adv, destPIDs, sender );
            } catch ( Throwable all ) {
                printStackTrace( "failure reaching user '" + destUserName + "'.", all );
                return ShellApp.appMiscError;
            }
        } else {
            // asynchronously get a messenger to the destination address.
            consoleMessage( "Getting messenger for '" + destAddr + "'" );
            boolean working = group.getEndpointService().getMessenger( sender, destAddr, null );
            if( !working ) {
                consoleMessage( "Could not get messenger for '" + destAddr + "'." );
                return ShellApp.appMiscError;
            }
        }
        
        synchronized( sender ) {
            long start =  System.currentTimeMillis();
            long finish = start + MAX_SEARCH_TIME;
            
            while( (!sender.resolved) && (System.currentTimeMillis() < finish) ) {
                try {
                    long waitFor = Math.max( 1, finish - System.currentTimeMillis());
                    waitFor = Math.min( waitFor, 30 * 1000 );
                    consoleMessage( "Please be patient ...(" + ((finish - System.currentTimeMillis()) / 1000) +" secs)" );
                    
                    sender.wait( waitFor );
                } catch( InterruptedException woken ) {
                    Thread.interrupted();
                }
            }
            
            if( !sender.resolved ) {
                if( null != adv ) {
                    getGroup().getPipeService().removeOutputPipeListener( adv.getPipeID().toString(), sender );
                }
                
                consoleMessage( "User '" + destUserName + "' is not listening. Try again later." );
                return ShellApp.appMiscError;
            } else {
                consoleMessage( "Located user '" + destUserName + "' after " + (System.currentTimeMillis() - start) + " msec." );
            }
        }
        
        return ShellApp.appNoError;
    }
    
    /**
     *  Search for a specific user by name.
     **/
    private PipeAdvertisement findUserAdv( String name ) {
        
        String attribute = XFERUSERNAME + "." + name;
        Enumeration each = null;
        
        int currentRetry = 0;
        while ( currentRetry <= MAXRETRIES ) {
            try {
                print( "." );
                
                // Look in the local storage
                each = discovery.getLocalAdvertisements(DiscoveryService.ADV, PipeAdvertisement.NameTag, attribute );
                
                while (each.hasMoreElements()) {
                    PipeAdvertisement adv = null;
                    
                    try {
                        adv = (PipeAdvertisement) each.nextElement();
                    } catch( ClassCastException skip ) {
                        continue;
                    }
                    
                    println(" ");
                    return adv;
                }
                
                // Now, search remote
                discovery.getRemoteAdvertisements(null, DiscoveryService.ADV, PipeAdvertisement.NameTag, attribute, 2, null );
                
                // nap for a while to let responses come in.
                try {
                    Thread.sleep( WAITINGTIME );
                } catch( InterruptedException woken ) {
                    Thread.interrupted();
                }
                
                currentRetry++;
            } catch (Exception ex) {
                println(" ");
                printStackTrace( "failure locating advertisement", ex );
            }
        }
        
        println(" ");
        
        // adv not found
        return null;
    }
    
    /**
     * Search for any users.
     **/
    private int findUserAdvs() {
        
        String attribute = XFERUSERNAME + "." + "*";
        
        int currentRetry = 0;
        while ( currentRetry <= MAXRETRIES ) {
            try {
                print( "." );
                
                // Now, search remote
                discovery.getRemoteAdvertisements(null, DiscoveryService.ADV, PipeAdvertisement.NameTag, attribute, 20, null );
                
                // nap for a while to let responses come in.
                try {
                    Thread.sleep( WAITINGTIME );
                } catch( InterruptedException woken ) {
                    Thread.interrupted();
                }
                
                currentRetry++;
            } catch (Throwable ex) {
                println(" ");
                printStackTrace( "Failure locating advertisement", ex );
            }
        }
        
        println(" ");
        
        // Look in the local storage
        Enumeration each = null;
        
        try {
            each = discovery.getLocalAdvertisements(DiscoveryService.ADV, PipeAdvertisement.NameTag, attribute );
        } catch ( IOException failed ) {
            printStackTrace( "Failure locating advertisement", failed );
        }
        
        while (each.hasMoreElements()) {
            PipeAdvertisement adv = null;
            
            try {
                adv = (PipeAdvertisement) each.nextElement();
            } catch( ClassCastException skip ) {
                continue;
            }
            
            println( adv.getName().substring( XFERUSERNAME.length() + 1 ) );
        }
        
        return ShellApp.appNoError;
    }
    
    /**
     * Create a new daemon for the specified user name.
     *
     *  @param userName user to log in.
     *  @param adv optional pipe advertisement. If null then endpoint listener will be created.
     *  @return if true then daemon was registered successfully otherwise false.
     **/
    private boolean runDaemon( String userName, PipeAdvertisement adv ) {
        
        try {
            XferDaemon daemon = new XferDaemon( getOutputConsPipe(), group, userName, adv );
            
            // Store this object
            String dname = ENVNAME + "." + userName + "@" + (group.getPeerGroupAdvertisement()).getName();
            
            ShellObject obj = new ShellObject("Xfer Deamon for " + userName + "@" + (group.getPeerGroupAdvertisement()).getName(), daemon);
            obj = env.add( dname, obj );
            consoleMessage( "Login of user '" + userName + "' successful." );
            return true;
        } catch (Throwable ex) {
            printStackTrace( "Failed to create daemon for user '" + userName + "'.", ex );
        }
        
        return false;
    }
    
    /**
     *  Check if a daemon is running for the specified user name.
     *
     *  @param userName user to check.
     *  @return if true then daemon was found otherwise false.
     **/
    private boolean isDaemonRunning( String userName ) {
        
        return ( null != getDaemon( userName ) );
    }
    
    
    /**
     *  Gets the daemon for the specified user name
     *
     *  @param userName The user name to get the XferDaemon object for
     *  @return the XferDaemon for the specified user name if available, otherwise null.
     **/
    private XferDaemon getDaemon( String userName ) {
        
        String dname = ENVNAME + "." + userName + "@" + (group.getPeerGroupAdvertisement()).getName();
        
        ShellObject obj = env.get(dname);
        
        if( null == obj ) {
            return null;
        }
        
        if( !XferDaemon.class.isInstance( obj.getObject() ) ) {
            throw new RuntimeException( dname + " is not an XferDaemon object" );
        }
        
        return (XferDaemon) obj.getObject();
    }
    
    /**
     *  Stops the daemon for the specified user name
     *
     *  @param userName The user name to close the XferDaemon
     **/
    private boolean stopDaemon( String userName ) {
        
        String dname = ENVNAME + "." + userName + "@" + (group.getPeerGroupAdvertisement()).getName();
        
        XferDaemon d = getDaemon( userName );
        
        if( null == d ) {
            return false;
        }
        
        try {
            d.close();
            env.remove(dname);
            return true;
        } catch ( Throwable ex ) {
            printStackTrace( "Cannot stop daemon for '" + userName + "'.", ex );
            return false;
        }
    }
}