filesender.java

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

    
    private void consoleMessage( String message ) {
        ShellApp.consoleMessage( this.getClass(), console, message );
    }
    
    private void printStackTrace( String annotation, Throwable failure ) {
        ShellApp.printStackTrace( this.getClass(), console, annotation, failure );
    }
    
    // main stuff
    
    private void send( Messenger msngr, OutputPipe pipe ) {
        InputStream is = null;
        DataInputStream fis = null;
        PendingMessageCounter counter = new PendingMessageCounter();
        
        try {
            long fileSize = f.length();
            
            // Try and open the file
            try {
                is = new FileInputStream(f);
                fis = new DataInputStream( is );
            } catch (Throwable all) {
                printStackTrace( "Failure opening file '" + f + "'.", all );
            }
            
            int chunks = (int) (fileSize / blockSize);
            chunks += (0 == (fileSize % blockSize)) ? 0 : 1 ;
            
            String xferidentifier = MessageElement.getUniqueName();
            
            consoleMessage( "Sending file '" + f.getName() + "' (" + fileSize + ") as " + xferidentifier + " in " + chunks + " chunks of " + blockSize + " bytes." );
            
            long fileSent = 0;
            int chunk = 0;
            long stime = System.currentTimeMillis();
            
            while( fileSent < fileSize ) {
                
                byte [] buf = new byte[(int) Math.min( fileSize - fileSent, blockSize)];
                
                fis.readFully(buf);
                
                Message msg = new Message();
                
                // set file info element in first message
                if (0 == fileSent) {
                    // file info
                    ByteArrayOutputStream os = new ByteArrayOutputStream();
                    DataOutputStream dos = new DataOutputStream( os );
                    
                    dos.writeInt( xfer.XFERFILEINFO_VERSION );
                    dos.writeUTF( group.getPeerID().toString() );
                    
                    if( null != src ) {
                        dos.writeUTF( src.getUserName() );
                        
                        PipeAdvertisement srcpipe = src.getPipeAdvertisment();
                        if ( null != srcpipe ) {
                            DiscoveryService disco = group.getDiscoveryService();
                            
                            long exp = disco.getAdvExpirationTime( srcpipe );
                            if( exp > 0 ) {
                                dos.writeUTF( srcpipe.getDocument( MimeMediaType.XMLUTF8 ).toString() );
                                dos.writeLong( exp );
                            } else {
                                dos.writeUTF( "" );
                                dos.writeLong( 0 );
                            }
                        } else {
                            dos.writeUTF( "" );
                            dos.writeLong( 0 );
                        }
                        
                    } else {
                        dos.writeUTF( "Anonymous" );
                        dos.writeUTF( "" );
                        dos.writeLong( 0 );
                    }
                    
                    dos.writeUTF( xferidentifier );
                    dos.writeUTF( f.getName() );
                    dos.writeLong( fileSize );
                    dos.writeLong( blockSize );
                    dos.flush();
                    dos.close();
                    os.flush();
                    os.close();
                    
                    msg.addMessageElement( new ByteArrayMessageElement( xfer.XFERFILEINFO_ELEMENT, null, os.toByteArray(), null ) );
                }
                
                // identifier element
                msg.addMessageElement( new StringMessageElement( xfer.XFERIDENTIFIER_ELEMENT, xferidentifier, null) );
                
                // sequence element
                msg.addMessageElement( new StringMessageElement( xfer.XFERSEQUENCE_ELEMENT, Integer.toString(chunk++), null) );
                
                // data element
                msg.addMessageElement( new ByteArrayMessageElement( xfer.XFERDATA_ELEMENT, null, buf, null) );
                
                
                boolean sent = false;
                long backOffSleep = 0;
                while( !sent ) {
                    
                    if( async && (null == pipe) ) {
                        counter.addPending();
                        
                        msngr.sendMessage( msg, null, null, counter );
                        
                        synchronized( counter ) {
                            while( counter.stillPending() ) {
                                try {
                                    counter.wait();
                                } catch ( InterruptedException woken ) {
                                    Thread.interrupted();
                                }
                            }
                            
                            if( counter.getLastResult() ) {
                                sent = true;
                            } else {
                                OutgoingMessageEvent lastevent = counter.getLastEvent();
                                
                                Throwable failure = lastevent.getFailure();
                                if( null == failure )
                                    sent = false;
                                else {
                                    if( failure instanceof Error ) {
                                        throw (Error) lastevent.getFailure();
                                    } else if ( failure instanceof RuntimeException ) {
                                        throw (RuntimeException) lastevent.getFailure();
                                    } else if ( failure instanceof IOException ) {
                                        throw (IOException) lastevent.getFailure();
                                    } else {
                                        throw new UndeclaredThrowableException( failure );
                                    }
                                }
                            }
                        }
                    } else {
                        if ( null == pipe ) {
                            sent = msngr.sendMessage( msg  );
                        } else {
                            sent = pipe.send( msg );
                        }
                    }
                    
                    if( sent ) {
                        backOffSleep = 0;
                        break;
                    }
                    
                    println( " " );
                    consoleMessage( "Sleeping for " + (1 << backOffSleep) * 2 + " msecs." );
                    
                    try {
                        Thread.sleep( (1 << backOffSleep) * 2 );
                    } catch ( InterruptedException woken ) {
                        Thread.interrupted();
                    }
                    backOffSleep++;
                }
                
                fileSent += buf.length;
                print( "!" );
            }
            
            // K bytes per second calcul
            long elapsedTime = System.currentTimeMillis() - stime;
            
            // protect agains div by 0
            if (elapsedTime == 0) {
                elapsedTime = 1;
            }
            
            long kbpsec = (long)(((long)fileSize)/elapsedTime);
            
            float secs = (float)(((float)elapsedTime)/1000.0);
            
            println( " " );
            consoleMessage( "Sent '" + f.getName() + "' ("+ fileSize + ") in " + secs + " secs[" + kbpsec + "KBytes/sec]");
        } catch( IOException failed ) {
            printStackTrace( "Failure during sending of '" + f.getName() + "'.", failed );
        } finally {
            if( null != pipe ) {
                pipe.close();
            }
            
            if( null != msngr ) {
                msngr.close();
            }
            
            if( null != fis ) {
                try {
                    fis.close();
                    fis = null;
                } catch ( IOException ignored ) {}
            }
            
            if( null != is ) {
                try {
                    is.close();
                    is = null;
                } catch ( IOException ignored ) {}
            }
        }
    }
    
    
}