nonblockingsocket.java

本人历尽千辛万苦找的clustream中的jar包

package org.osu.ogsa.stream.util;


import java.io.*;
import java.nio.*;
import java.nio.channels.*;
import java.nio.channels.spi.*;
import java.nio.charset.*;
import java.net.*;
import java.util.*;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

abstract public class NonBlockingSocket {

    SocketNotification mysockNotification;
    int port = DefConstants.SOCKET_PORT;
    String hostname;

    static Log log =
        LogFactory.getLog(NonBlockingSocket.class.getName());

    static final String QUIT_SERVER = "quit";
    static final String SHUTDOWN = "shutdown";

    public NonBlockingSocket() {
    }

    public NonBlockingSocket( int port ) {
        this.port = port;
    }

    public NonBlockingSocket( String hostname, int port ) {
	this.hostname = hostname;
        this.port = port;
    }

    public void addCallback(SocketNotification mysockNotification)
    {
 	this.mysockNotification = mysockNotification;
    }


    abstract public SocketChannel initialize() throws IOException; 

    abstract public void finalize() throws IOException; 

    public void writeMessage( SocketChannel channel, String message )
    throws IOException {
        ByteBuffer buf = ByteBuffer.wrap( message.getBytes()  );
        int nbytes = channel.write( buf );
        log.debug( "Wrote " + nbytes + " to channel." );
    }

    public void writeBytes( SocketChannel channel, byte[] buf)
    throws IOException {
        ByteBuffer src = ByteBuffer.wrap( buf );
        int nbytes = channel.write( src );
        log.debug( "Wrote " + nbytes + " to channel." );
    }
 
    public void writeBytes( SocketChannel channel, byte[] buf, 
		    			int offset, int length)
    throws IOException {
        ByteBuffer src = ByteBuffer.wrap( buf, offset, length );
        int nbytes = channel.write( src );
        log.debug( "Wrote " + nbytes + " to channel." );
    }

//    static final int BUFSIZE = 8;

    public String decode( ByteBuffer byteBuffer )
    throws CharacterCodingException {

        Charset charset = Charset.forName( "us-ascii" );
        CharsetDecoder decoder = charset.newDecoder();
        CharBuffer charBuffer = decoder.decode( byteBuffer );
        String result = charBuffer.toString();
        return result;
    }

/*    public void readBytes( SocketChannel channel)
    throws IOException, InterruptedException {
        ByteBuffer byteBuffer = ByteBuffer.allocate( BUFSIZE );
        int nbytes = callback.getChannel().read( byteBuffer );
        byteBuffer.flip();
        String result = this.decode( byteBuffer );
        log.debug( result );
        if ( result.indexOf( "quit" ) >= 0 ) callback.getChannel().close();
        else if ( result.indexOf( "shutdown" ) >= 0 ) {
            callback.getChannel().close();
            throw new InterruptedException();
        }
        else {
            callback.append( result.toString() );
            //If we are done with the line then we execute the callback.
            if ( result.indexOf( "\n" ) >= 0 )
                callback.execute();
        }
    }*/

/*    public class ChannelCallback {
        private SocketChannel channel;
        private StringBuffer buffer;

        public ChannelCallback( SocketChannel channel ) {
            this.channel = channel;
            this.buffer = new StringBuffer();
        }

        public void execute() throws IOException {
            log.debug( this.buffer.toString() );
            writeMessage( this.channel, this.buffer.toString() );
            buffer = new StringBuffer();
        }

        public SocketChannel getChannel() {
            return this.channel;
        }

        public void append( String values ) {
            buffer.append( values );
        }

    }
 

    public static void main( String[] args ) {
//        BasicConfigurator.configure();

        NonBlockingSocket nbServer = new NonBlockingSocket();

        try {
            nbServer.initialize();
        } catch ( IOException e ) {
            e.printStackTrace();
            System.exit( -1 );
        }

        try {
            nbServer.acceptConnections();
        }

        catch ( IOException e ) {
            e.printStackTrace();
            log.error( e );
        }

        catch ( InterruptedException e ) {
            log.info( "Exiting normally..." );
        }

    }
*/
}