nonblockingclient.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 NonBlockingClient extends NonBlockingSocket{

	private InetSocketAddress serverAddress;
       	private Selector selector;
	private SocketChannel clientChannel;
	private static Log log = LogFactory.getLog(NonBlockingClient.class.getName());

    public NonBlockingClient() {
    }

    public NonBlockingClient(String hostname , int port){
	    super(hostname , port);
    }

    public NonBlockingClient(InetSocketAddress isa)
    {
	    serverAddress = isa;
    }
    // Creates a non-blocking socket channel for the specified host name and port.
    // connect() is called on the new channel before it is returned.
    public SocketChannel createSocketChannel(String hostName, int port) throws IOException {
	    // Create a non-blocking socket channel
	    SocketChannel sChannel = SocketChannel.open();
            sChannel.configureBlocking(true);
            // Send a connection request to the server; this method is non-blocking
	    log.debug("connecting to the server....");
	    sChannel.connect(new InetSocketAddress(hostName, port));
            sChannel.configureBlocking(false);
	    return sChannel;
    }

    public void processSelectionKey(SelectionKey selKey) throws IOException {
        // Since the ready operations are cumulative,
        // need to check readiness for each operation
        if (selKey.isValid() && selKey.isConnectable()) {
            // Get channel with connection request
            SocketChannel sChannel = (SocketChannel)selKey.channel();
    
            boolean success = sChannel.finishConnect();
            if (!success) {
                // An error occurred; handle it
   		log.debug("the socket has been not connected successfully yet"); 
                // Unregister the channel with this selector
                selKey.cancel();
            }
	    else 
		    mysockNotification.socketConnectedEvent(selKey, sChannel);
        }

        SocketChannel sChannel = (SocketChannel)selKey.channel();
	log.info(sChannel);

        if (selKey.isValid() && selKey.isReadable()) {
            // Get channel with bytes to read
	    //We hand the event to the class that delegetes this class
	    mysockNotification.readBytesEvent( sChannel );
    
        }
        if (selKey.isValid() && selKey.isWritable()) {
            // Get channel that's ready for more bytes
	    
	    //We hand the event to the class that delegetes this class
	    mysockNotification.writeBytesEvent( sChannel );
    
        }
    }
    	
    public SocketChannel initialize() throws IOException {
	InetSocketAddress isa = new InetSocketAddress(hostname, port);
	boolean bContinue = false;
    	try {
       		selector = SelectorProvider.provider().openSelector();
	}
	catch (IOException e) 
	{
		e.printStackTrace();
	}

	while(true)
	{
		try{
			if(bContinue)
				Thread.currentThread().sleep((long)(1));
			clientChannel = SocketChannel.open();
			clientChannel.connect(isa);
	    		if(clientChannel.isConnected())
			{
	   	 		log.info("connected to the server....");
				break;
			}
			else
				log.warn("can't connect to the server...");
		}catch(ConnectException e)
		{
			bContinue = true;	
			log.warn(e);
			continue;
		}
		catch(ClosedChannelException e)
		{
			bContinue = true;
			log.warn(e);
			continue;
		}
		catch(IOException e)
		{
			e.printStackTrace();
			return null;
		}
		catch(java.lang.InterruptedException e)
		{
			e.printStackTrace();
			return null;
		}
	}

	try{
            	clientChannel.configureBlocking(false);
		clientChannel.register(selector, clientChannel.validOps());
	}catch (IOException e) {
		e.printStackTrace();
		return null;
	}
	return clientChannel;
    }

    public void finalize() throws IOException {
        this.clientChannel.close();
        this.selector.close();
    }

    public void handleEvents() throws IOException, InterruptedException {

	    	if(!clientChannel.isConnected())
		{
			log.debug("not connected yet");
			return;
		}
//		log.debug( " waiting for even ");
        	try {
            		selector.select();
	        } catch (IOException e) {
        	// Handle error with selector
        	}
    
	        // Get list of selection keys with pending events
       		Iterator it = selector.selectedKeys().iterator();
    
	        // Process each key at a time
		while (it.hasNext()) {
	        // Get the selection key
        		SelectionKey selKey = (SelectionKey)it.next();
    
	        // Remove it from the list to indicate that it is being processed
			it.remove();
    
       			try {
		                processSelectionKey(selKey);
		        } catch (IOException e) {
               		 // Handle error with channel and unregister
		                selKey.cancel();
       			}
        	}

//        log.debug( "End client handling loop..." );
    }
}