nonblockingchannel.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;
public class NonBlockingChannel {

    SocketChannel channel;
    SocketNotification mysockNotification;
    Selector selector = null;
    private static Log log = LogFactory.getLog(NonBlockingChannel.class.getName());


    public NonBlockingChannel(SocketChannel sc) throws IOException{
	    channel = sc;
	    this.selector = SelectorProvider.provider().openSelector();
    }

    public void finalize() throws IOException {
        this.selector.close();
    }
    
    public void addCallback(SocketNotification s){
	    mysockNotification = s;
    }


    public void initialize() throws IOException, ClosedChannelException{ 
             channel.configureBlocking( false );
             SelectionKey readKey = channel.register( this.selector,
                            SelectionKey.OP_READ|SelectionKey.OP_WRITE  );
             readKey.attach( channel);
//	     mysockNotification.socketConnectedEvent(readKey, channel);
    }

	     
    public void handleEvents() throws IOException, InterruptedException
    {
	     // Wait for events
//	     while (true) {
		if(!channel.isConnected())
			return;

	     	try {
	        // Wait for an event
	     	selector.select();
	        } catch (IOException e) {
	       // Handle error with selector
	        }
	     	// Get list of selection keys with pending events
		Iterator i = selector.selectedKeys().iterator(); 
		if(i == null)
			return;
 
            while (i.hasNext()) {
                SelectionKey key = (SelectionKey)i.next();
                i.remove();

//		log.info(key.attachment());

                if ( key.isReadable() ) {
//                    SelectableChannel nextReady = (SelectableChannel) key.channel();
                    log.debug( "Processing selection key read="
                        + key.isReadable() + " write=" + key.isWritable() +
                        " accept=" + key.isAcceptable() );
                    mysockNotification.readBytesEvent( (SocketChannel) key.attachment() );
                }

                else if ( key.isWritable() ) {
                    //ChannelCallback callback = (ChannelCallback) key.attachment();
                    //String message = "What is your name? ";
                    //ByteBuffer buf = ByteBuffer.wrap( message.getBytes() );
                    log.debug( "Processing selection key read="
                        + key.isReadable() + " write=" + key.isWritable() +
                        " accept=" + key.isAcceptable() );
                    mysockNotification.writeBytesEvent( (SocketChannel) key.attachment());
                }
            }
//        }
//        log.debug( "End handle Events loop..." );
    }
}