socketorchannelacceptorimpl.java

java1.6众多例子参考

/*
 * @(#)SocketOrChannelAcceptorImpl.java	1.55 05/11/17
 * 
 * Copyright 2006 Sun Microsystems, Inc. All rights reserved.
 * SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 */

package com.sun.corba.se.impl.transport;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedList;

import org.omg.CORBA.CompletionStatus;
import org.omg.CORBA.INTERNAL;

import com.sun.corba.se.pept.broker.Broker;
import com.sun.corba.se.pept.encoding.InputObject;
import com.sun.corba.se.pept.encoding.OutputObject;
import com.sun.corba.se.pept.protocol.MessageMediator;
import com.sun.corba.se.pept.transport.Acceptor;
import com.sun.corba.se.pept.transport.Connection;
import com.sun.corba.se.pept.transport.ContactInfo;
import com.sun.corba.se.pept.transport.EventHandler;
import com.sun.corba.se.pept.transport.InboundConnectionCache;
import com.sun.corba.se.pept.transport.Selector;

import com.sun.corba.se.spi.extension.RequestPartitioningPolicy;
import com.sun.corba.se.spi.ior.IORTemplate;
import com.sun.corba.se.spi.ior.TaggedProfileTemplate;
import com.sun.corba.se.spi.ior.iiop.IIOPAddress ;
import com.sun.corba.se.spi.ior.iiop.IIOPFactories;
import com.sun.corba.se.spi.ior.iiop.IIOPProfileTemplate ;
import com.sun.corba.se.spi.ior.iiop.GIOPVersion ;
import com.sun.corba.se.spi.ior.iiop.AlternateIIOPAddressComponent;
import com.sun.corba.se.spi.legacy.connection.LegacyServerSocketEndPointInfo;
import com.sun.corba.se.spi.logging.CORBALogDomains;
import com.sun.corba.se.spi.monitoring.LongMonitoredAttributeBase;
import com.sun.corba.se.spi.monitoring.MonitoringConstants;
import com.sun.corba.se.spi.monitoring.MonitoringFactories;
import com.sun.corba.se.spi.monitoring.MonitoredObject;
import com.sun.corba.se.spi.orb.ORB;
import com.sun.corba.se.spi.orbutil.threadpool.Work;
import com.sun.corba.se.spi.protocol.CorbaMessageMediator;
import com.sun.corba.se.spi.transport.CorbaAcceptor;
import com.sun.corba.se.spi.transport.CorbaConnection;
import com.sun.corba.se.spi.transport.CorbaContactInfo;
import com.sun.corba.se.spi.transport.SocketInfo;
import com.sun.corba.se.spi.transport.SocketOrChannelAcceptor;

import com.sun.corba.se.impl.encoding.CDRInputObject;
import com.sun.corba.se.impl.encoding.CDROutputObject;
import com.sun.corba.se.impl.logging.ORBUtilSystemException;
import com.sun.corba.se.impl.oa.poa.Policies; // REVISIT impl/poa specific
import com.sun.corba.se.impl.orbutil.ORBConstants;
import com.sun.corba.se.impl.orbutil.ORBUtility;
import com.sun.corba.se.impl.ior.iiop.JavaSerializationComponent;

// BEGIN Legacy support.
import com.sun.corba.se.spi.legacy.connection.LegacyServerSocketEndPointInfo;
// END Legacy support.

/**
 * @author Harold Carr
 */
public class SocketOrChannelAcceptorImpl
    extends
	EventHandlerBase
    implements
	CorbaAcceptor,
	SocketOrChannelAcceptor,
	Work,
	// BEGIN Legacy
	SocketInfo,
	LegacyServerSocketEndPointInfo
	// END Legacy
{
    protected ServerSocketChannel serverSocketChannel;
    protected ServerSocket serverSocket;
    protected int port;
    protected long enqueueTime;
    protected boolean initialized;
    protected ORBUtilSystemException wrapper ;
    protected InboundConnectionCache connectionCache;

    // BEGIN Legacy
    protected String type = "";
    protected String name = "";
    protected String hostname;
    protected int locatorPort;
    // END Legacy

    public SocketOrChannelAcceptorImpl(ORB orb)
    {
	this.orb = orb;
	wrapper = ORBUtilSystemException.get( orb,
	    CORBALogDomains.RPC_TRANSPORT ) ;

	setWork(this);
	initialized = false;

	// BEGIN Legacy support.
	this.hostname = orb.getORBData().getORBServerHost();
	this.name = LegacyServerSocketEndPointInfo.NO_NAME;
	this.locatorPort = -1;
	// END Legacy support.
    }

    public SocketOrChannelAcceptorImpl(ORB orb, int port)
    {
	this(orb);
	this.port = port;
    }

    // BEGIN Legacy support.
    public SocketOrChannelAcceptorImpl(ORB orb, int port, 
				       String name, String type)
    {
	this(orb, port);
	this.name = name;
	this.type = type;
    }
    // END Legacy support.

    ////////////////////////////////////////////////////
    //
    // pept.transport.Acceptor
    //

    public boolean initialize()
    {
	if (initialized) {
	    return false;
	}
	if (orb.transportDebugFlag) {
	    dprint(".initialize: " + this);
	}
	InetSocketAddress inetSocketAddress = null;
	try {
	    if (orb.getORBData().getListenOnAllInterfaces().equals(ORBConstants.LISTEN_ON_ALL_INTERFACES)) {
		inetSocketAddress = new InetSocketAddress(port);
	    } else {
		String host = orb.getORBData().getORBServerHost();
		inetSocketAddress = new InetSocketAddress(host, port);
	    }
	    serverSocket = orb.getORBData().getSocketFactory()
		.createServerSocket(type, inetSocketAddress);
	    internalInitialize();
	} catch (Throwable t) {
	    throw wrapper.createListenerFailed( t, Integer.toString(port) ) ;
	}
	initialized = true;
	return true;
    }

    protected void internalInitialize()
	throws Exception
    {
	// Determine the listening port (for the IOR).
	// This is important when using emphemeral ports (i.e.,
	// when the port value to the constructor is 0).

	port = serverSocket.getLocalPort();

	// Register with transport (also sets up monitoring).

	orb.getCorbaTransportManager().getInboundConnectionCache(this);

	// Finish configuation.

	serverSocketChannel = serverSocket.getChannel();

	if (serverSocketChannel != null) {
	    setUseSelectThreadToWait(
	        orb.getORBData().acceptorSocketUseSelectThreadToWait());
	    serverSocketChannel.configureBlocking(
	        ! orb.getORBData().acceptorSocketUseSelectThreadToWait());
	} else {
	    // Configure to use listener and reader threads.
	    setUseSelectThreadToWait(false);
	}
	setUseWorkerThreadForEvent(
            orb.getORBData().acceptorSocketUseWorkerThreadForEvent());

    }

    public boolean initialized()
    {
	return initialized;
    }

    public String getConnectionCacheType()
    {
	return this.getClass().toString();
    }

    public void setConnectionCache(InboundConnectionCache connectionCache)
    {
	this.connectionCache = connectionCache;
    }

    public InboundConnectionCache getConnectionCache()
    {
	return connectionCache;
    }

    public boolean shouldRegisterAcceptEvent()
    {
	return true;
    }

    public void accept()
    {
	try {
	    SocketChannel socketChannel = null;
	    Socket socket = null;
	    if (serverSocketChannel == null) {
		socket = serverSocket.accept();
	    } else {
		socketChannel = serverSocketChannel.accept();
		socket = socketChannel.socket();
	    }
	    orb.getORBData().getSocketFactory()
		.setAcceptedSocketOptions(this, serverSocket, socket);
	    if (orb.transportDebugFlag) {
		dprint(".accept: " + 
		       (serverSocketChannel == null 
			? serverSocket.toString()
			: serverSocketChannel.toString()));
	    }

	    CorbaConnection connection = 
		new SocketOrChannelConnectionImpl(orb, this, socket);
	    if (orb.transportDebugFlag) {
		dprint(".accept: new: " + connection);
	    }

	    // NOTE: The connection MUST be put in the cache BEFORE being
	    // registered with the selector.  Otherwise if the bytes
	    // are read on the connection it will attempt a time stamp
	    // but the cache will be null, resulting in NPE.
	    getConnectionCache().put(this, connection);

	    if (connection.shouldRegisterServerReadEvent()) {
		Selector selector = orb.getTransportManager().getSelector(0);
		selector.registerForEvent(connection.getEventHandler());
	    }

	    getConnectionCache().reclaim();

	} catch (IOException e) {
	    if (orb.transportDebugFlag) {
		dprint(".accept:", e);
	    }
	    orb.getTransportManager().getSelector(0).unregisterForEvent(this);
	    // REVISIT - need to close - recreate - then register new one.
	    orb.getTransportManager().getSelector(0).registerForEvent(this);
	    // NOTE: if register cycling we do not want to shut down ORB
	    // since local beans will still work.  Instead one will see
	    // a growing log file to alert admin of problem.
	}
    }

    public void close ()
    {
	try {
	    if (orb.transportDebugFlag) {
		dprint(".close->:");
	    }
	    Selector selector = orb.getTransportManager().getSelector(0);
	    selector.unregisterForEvent(this);
	    if (serverSocketChannel != null) {
		serverSocketChannel.close();
	    }
	    if (serverSocket != null) {
		serverSocket.close();
	    }
	} catch (IOException e) {
	    if (orb.transportDebugFlag) {
		dprint(".close:", e);
	    }
	} finally {
	    if (orb.transportDebugFlag) {
		dprint(".close<-:");
	    }
	}
    }

    public EventHandler getEventHandler()
    {
	return this;
    }

    ////////////////////////////////////////////////////
    //
    // CorbaAcceptor
    //

    public String getObjectAdapterId()
    {
	return null;
    }

    public String getObjectAdapterManagerId()
    {
	return null;
    }

    public void addToIORTemplate(IORTemplate iorTemplate,
				 Policies policies,
				 String codebase)
    {
	Iterator iterator = iorTemplate.iteratorById(
            org.omg.IOP.TAG_INTERNET_IOP.value);