/*
* @(#)SocketOrChannelAcceptorImpl.java 1.54 04/06/21
*
* Copyright 2004 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);
String hostname = orb.getORBData().getORBServerHost();
if (iterator.hasNext()) {
// REVISIT - how does this play with legacy ORBD port exchange?
IIOPAddress iiopAddress =
IIOPFactories.makeIIOPAddress(orb, hostname, port);
AlternateIIOPAddressComponent iiopAddressComponent =
IIOPFactories.makeAlternateIIOPAddressComponent(iiopAddress);
while (iterator.hasNext()) {
TaggedProfileTemplate taggedProfileTemplate =
(TaggedProfileTemplate) iterator.next();
taggedProfileTemplate.add(iiopAddressComponent);
}
} else {
GIOPVersion version = orb.getORBData().getGIOPVersion();
int templatePort;
if (policies.forceZeroPort()) {
templatePort = 0;
} else if (policies.isTransient()) {
templatePort = port;
} else {
templatePort = orb.getLegacyServerSocketManager()
.legacyGetPersistentServerPort(SocketInfo.IIOP_CLEAR_TEXT);
}
IIOPAddress addr =
IIOPFactories.makeIIOPAddress(orb, hostname, templatePort);
IIOPProfileTemplate iiopProfile =
IIOPFactories.makeIIOPProfileTemplate(orb, version, addr);
if (version.supportsIORIIOPProfileComponents()) {
iiopProfile.add(IIOPFactories.makeCodeSetsComponent(orb));
iiopProfile.add(IIOPFactories.makeMaxStreamFormatVersionComponent());
RequestPartitioningPolicy rpPolicy = (RequestPartitioningPolicy)
policies.get_effective_policy(
ORBConstants.REQUEST_PARTITIONING_POLICY);
if (rpPolicy != null) {
iiopProfile.add(
IIOPFactories.makeRequestPartitioningComponent(
rpPolicy.getValue()));
}
if (codebase != null && codebase != "") {
iiopProfile.add(IIOPFactories. makeJavaCodebaseComponent(codebase));
}
if (orb.getORBData().isJavaSerializationEnabled()) {
iiopProfile.add(
IIOPFactories.makeJavaSerializationComponent());
}
}
iorTemplate.add(iiopProfile);
}
}
public String getMonitoringName()
{
return "AcceptedConnections";
}
////////////////////////////////////////////////////
//
// EventHandler methods
//
public SelectableChannel getChannel()
{
return serverSocketChannel;
}
public int getInterestOps()
{
return SelectionKey.OP_ACCEPT;
}
public Acceptor getAcceptor()
{
return this;
}
public Connection getConnection()
{
throw new RuntimeException("Should not happen.");
}
////////////////////////////////////////////////////
//
// Work methods.
//
/* CONFLICT: with legacy below.
public String getName()
{
return this.toString();
}
*/
public void doWork()
{
try {
if (orb.transportDebugFlag) {
dprint(".doWork->: " + this);
}
if (selectionKey.isAcceptable()) {
AccessController.doPrivileged(new PrivilegedAction() {
public java.lang.Object run() {
accept();
return null;
}
});
} else {
if (orb.transportDebugFlag) {
dprint(".doWork: ! selectionKey.isAcceptable: " + this);
}
}
} catch (SecurityException se) {
if (orb.transportDebugFlag) {
dprint(".doWork: ignoring SecurityException: "
+ se
+ " " + this);
}
String permissionStr = ORBUtility.getClassSecurityInfo(getClass());
wrapper.securityExceptionInAccept(se, permissionStr);
} catch (Exception ex) {
if (orb.transportDebugFlag) {
dprint(".doWork: ignoring Exception: "
+ ex
+ " " + this);
}
wrapper.exceptionInAccept(ex);
} catch (Throwable t) {
if (orb.transportDebugFlag) {
dprint(".doWork: ignoring Throwable: "
+ t
+ " " + this);
}
} finally {
// IMPORTANT: To avoid bug (4953599), we force the
// Thread that does the NIO select to also do the
// enable/disable of Ops using SelectionKey.interestOps().
// Otherwise, the SelectionKey.interestOps() may block
// indefinitely.
// NOTE: If "acceptorSocketUseWorkerThreadForEvent" is
// set to to false in ParserTable.java, then this method,
// doWork(), will get executed by the same thread
// (SelectorThread) that does the NIO select.
// If "acceptorSocketUseWorkerThreadForEvent" is set
// to true, a WorkerThread will execute this method,
// doWork(). Hence, the registering of the enabling of
// the SelectionKey's interestOps is done here instead
// of calling SelectionKey.interestOps(<interest op>).
Selector selector = orb.getTransportManager().getSelector(0);
selector.registerInterestOps(this);
if (orb.transportDebugFlag) {
dprint(".doWork<-:" + this);
}
}
}
public void setEnqueueTime(long timeInMillis)
{
enqueueTime = timeInMillis;
}
public long getEnqueueTime()
{
return enqueueTime;
}
//
// Factory methods.
//
// REVISIT: refactor into common base or delegate.
public MessageMediator createMessageMediator(Broker broker,
Connection connection)
{
// REVISIT - no factoring so cheat to avoid code dup right now.
// REVISIT **** COUPLING !!!!
ContactInfo contactInfo = new SocketOrChannelContactInfoImpl();
return contactInfo.createMessageMediator(broker, connection);
}
// REVISIT: refactor into common base or delegate.
public MessageMediator finishCreatingMessageMediator(Broker broker,
Connection connection,
MessageMediator messageMediator)
{
// REVISIT - no factoring so cheat to avoid code dup right now.
// REVISIT **** COUPLING !!!!
ContactInfo contactInfo = new SocketOrChannelContactInfoImpl();
return contactInfo.finishCreatingMessageMediator(broker,
connection, messageMediator);
}
public InputObject createInputObject(Broker broker,
MessageMediator messageMediator)
{
CorbaMessageMediator corbaMessageMediator = (CorbaMessageMediator)
messageMediator;
return new CDRInputObject((ORB)broker,
(CorbaConnection)messageMediator.getConnection(),
corbaMessageMediator.getDispatchBuffer(),
corbaMessageMediator.getDispatchHeader());
}
public OutputObject createOutputObject(Broker broker,
MessageMediator messageMediator)
{
CorbaMessageMediator corbaMessageMediator = (CorbaMessageMediator)
messageMediator;
return new CDROutputObject((ORB) broker, corbaMessageMediator,
corbaMessageMediator.getReplyHeader(),
corbaMessageMediator.getStreamFormatVersion());
}
////////////////////////////////////////////////////
//
// SocketOrChannelAcceptor
//
public ServerSocket getServerSocket()
{
return serverSocket;
}
////////////////////////////////////////////////////
//
// Implementation.
//
public String toString()
{
String sock;
if (serverSocketChannel == null) {
if (serverSocket == null) {
sock = "(not initialized)";
} else {
sock = serverSocket.toString();
}
} else {
sock = serverSocketChannel.toString();
}
return
toStringName() +
"["
+ sock + " "
+ type + " "
+ shouldUseSelectThreadToWait() + " "
+ shouldUseWorkerThreadForEvent()
+ "]" ;
}
protected String toStringName()
{
return "SocketOrChannelAcceptorImpl";
}
protected void dprint(String msg)
{
ORBUtility.dprint(toStringName(), msg);
}
protected void dprint(String msg, Throwable t)
{
dprint(msg);
t.printStackTrace(System.out);
}
// BEGIN Legacy support
////////////////////////////////////////////////////
//
// LegacyServerSocketEndPointInfo and EndPointInfo
//
public String getType()
{
return type;
}
public String getHostName()
{
return hostname;
}
public String getHost()
{
return hostname;
}
public int getPort()
{
return port;
}
public int getLocatorPort()
{
return locatorPort;
}
public void setLocatorPort (int port)
{
locatorPort = port;
}
public String getName()
{
// Kluge alert:
// Work and Legacy both define getName.
// Try to make this behave best for most cases.
String result =
name.equals(LegacyServerSocketEndPointInfo.NO_NAME) ?
this.toString() : name;
return result;
}
// END Legacy support
}
// End of file.
|
