1 /*

     2  * Copyright (c) 2004, Oracle and/or its affiliates. All rights reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.  Oracle designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Oracle in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    22  * or visit www.oracle.com if you need additional information or have any

    23  * questions.

    24  */

    25 

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

    27 

    28 import java.io.IOException;

    29 import java.net.InetSocketAddress;

    30 import java.net.Socket;

    31 import java.nio.ByteBuffer;

    32 import java.nio.channels.SelectableChannel;

    33 import java.nio.channels.SelectionKey;

    34 import java.nio.channels.SocketChannel;

    35 import java.security.AccessController;

    36 import java.security.PrivilegedAction;

    37 import java.util.Collections;

    38 import java.util.Hashtable;

    39 import java.util.HashMap;

    40 import java.util.Map;

    41 

    42 import org.omg.CORBA.COMM_FAILURE;

    43 import org.omg.CORBA.CompletionStatus;

    44 import org.omg.CORBA.DATA_CONVERSION;

    45 import org.omg.CORBA.INTERNAL;

    46 import org.omg.CORBA.MARSHAL;

    47 import org.omg.CORBA.OBJECT_NOT_EXIST;

    48 import org.omg.CORBA.SystemException;

    49 

    50 import com.sun.org.omg.SendingContext.CodeBase;

    51 

    52 import com.sun.corba.se.pept.broker.Broker;

    53 import com.sun.corba.se.pept.encoding.InputObject;

    54 import com.sun.corba.se.pept.encoding.OutputObject;

    55 import com.sun.corba.se.pept.protocol.MessageMediator;

    56 import com.sun.corba.se.pept.transport.Acceptor;

    57 import com.sun.corba.se.pept.transport.Connection;

    58 import com.sun.corba.se.pept.transport.ConnectionCache;

    59 import com.sun.corba.se.pept.transport.ContactInfo;

    60 import com.sun.corba.se.pept.transport.EventHandler;

    61 import com.sun.corba.se.pept.transport.InboundConnectionCache;

    62 import com.sun.corba.se.pept.transport.OutboundConnectionCache;

    63 import com.sun.corba.se.pept.transport.ResponseWaitingRoom;

    64 import com.sun.corba.se.pept.transport.Selector;

    65 

    66 import com.sun.corba.se.spi.ior.IOR;

    67 import com.sun.corba.se.spi.ior.iiop.GIOPVersion;

    68 import com.sun.corba.se.spi.logging.CORBALogDomains;

    69 import com.sun.corba.se.spi.orb.ORB ;

    70 import com.sun.corba.se.spi.orbutil.threadpool.Work;

    71 import com.sun.corba.se.spi.protocol.CorbaMessageMediator;

    72 import com.sun.corba.se.spi.transport.CorbaContactInfo;

    73 import com.sun.corba.se.spi.transport.CorbaConnection;

    74 import com.sun.corba.se.spi.transport.CorbaResponseWaitingRoom;

    75 

    76 import com.sun.corba.se.impl.encoding.CachedCodeBase;

    77 import com.sun.corba.se.impl.encoding.CDRInputStream_1_0;

    78 import com.sun.corba.se.impl.encoding.CDROutputObject;

    79 import com.sun.corba.se.impl.encoding.CDROutputStream_1_0;

    80 import com.sun.corba.se.impl.encoding.CodeSetComponentInfo;

    81 import com.sun.corba.se.impl.encoding.OSFCodeSetRegistry;

    82 import com.sun.corba.se.impl.logging.ORBUtilSystemException;

    83 import com.sun.corba.se.impl.orbutil.ORBConstants;

    84 import com.sun.corba.se.impl.orbutil.ORBUtility;

    85 import com.sun.corba.se.impl.protocol.giopmsgheaders.Message;

    86 import com.sun.corba.se.impl.protocol.giopmsgheaders.MessageBase;

    87 import com.sun.corba.se.impl.transport.CorbaResponseWaitingRoomImpl;

    88 

    89 /**

    90  * @author Ken Cavanaugh

    91  */

    92 public class BufferConnectionImpl

    93     extends

    94 	EventHandlerBase

    95     implements

    96         CorbaConnection,

    97 	Work

    98 {

    99     //

   100     // New transport.

   101     //

   102 

   103     protected long enqueueTime;

   104 

   105     public SocketChannel getSocketChannel()

   106     {

   107 	return null;

   108     }

   109 

   110     // REVISIT:

   111     // protected for test: genericRPCMSGFramework.IIOPConnection constructor.

   112 

   113     //

   114     // From iiop.Connection.java

   115     //

   116 

   117     protected long timeStamp = 0;

   118     protected boolean isServer = false;

   119 

   120     // Start at some value other than zero since this is a magic

   121     // value in some protocols.

   122     protected int requestId = 5;

   123     protected CorbaResponseWaitingRoom responseWaitingRoom;

   124     protected int state;

   125     protected java.lang.Object stateEvent = new java.lang.Object();

   126     protected java.lang.Object writeEvent = new java.lang.Object();

   127     protected boolean writeLocked;

   128     protected int serverRequestCount = 0;

   129     

   130     // Server request map: used on the server side of Connection

   131     // Maps request ID to IIOPInputStream.

   132     Map serverRequestMap = new HashMap() ;

   133 

   134     // This is a flag associated per connection telling us if the

   135     // initial set of sending contexts were sent to the receiver

   136     // already...

   137     protected boolean postInitialContexts = false;

   138  

   139     // Remote reference to CodeBase server (supplies

   140     // FullValueDescription, among other things)

   141     protected IOR codeBaseServerIOR;

   142 

   143     // CodeBase cache for this connection.  This will cache remote operations,

   144     // handle connecting, and ensure we don't do any remote operations until

   145     // necessary.

   146     protected CachedCodeBase cachedCodeBase = new CachedCodeBase(this);

   147 

   148     protected ORBUtilSystemException wrapper ;

   149 

   150     List buffers ;

   151 

   152     public BufferConnectionImpl(ORB orb, byte[][] data )

   153     {

   154 	this.orb = orb;

   155 	wrapper = ORBUtilSystemException.get( orb,

   156 	    CORBALogDomains.RPC_TRANSPORT ) ;

   157 	buffers = new ArrayList() ;

   158     }

   159 

   160     ////////////////////////////////////////////////////

   161     //

   162     // framework.transport.Connection

   163     //

   164 

   165     public boolean shouldRegisterReadEvent()

   166     {

   167 	return false;

   168     }

   169 

   170     public boolean shouldRegisterServerReadEvent()

   171     {

   172 	return false;

   173     }

   174 

   175     public boolean read()

   176     {

   177 	return true ;

   178     }

   179 

   180     protected CorbaMessageMediator readBits()

   181     {

   182 	return null ;

   183     }

   184 

   185     protected boolean dispatch(CorbaMessageMediator messageMediator)

   186     {

   187     }

   188 

   189     public boolean shouldUseDirectByteBuffers()

   190     {

   191 	return false ;

   192     }

   193 

   194     // Only called from readGIOPMessage with (12, 0, 12) as arguments

   195     // size is size of buffer to create

   196     // offset is offset from start of message in buffer

   197     // length is length to read

   198     public ByteBuffer read(int size, int offset, int length)

   199 	throws IOException

   200     {

   201 	byte[] buf = new byte[size];

   202 	readFully( buf, offset, length);

   203 	ByteBuffer byteBuffer = ByteBuffer.wrap(buf);

   204 	byteBuffer.limit(size);

   205 	return byteBuffer;

   206     }

   207 

   208     // Only called as read( buf, 12, msgsize-12 ) in readGIOPMessage

   209     // We can ignore the byteBuffer parameter

   210     // offset is the starting position to place data in the result

   211     // length is the length of the data to read

   212     public ByteBuffer read(ByteBuffer byteBuffer, int offset, int length)

   213 	throws IOException

   214     {

   215 	int size = offset + length;

   216 	byte[] buf = new byte[size];

   217 	readFully(buf, offset, length);

   218 	return ByteBuffer.wrap(buf);

   219     }

   220 

   221     // Read size bytes from buffer list and place the data

   222     // starting at offset in buf.

   223     public void readFully(byte[] buf, int offset, int size) 

   224 	throws IOException

   225     {

   226 	int remaining = size ;

   227 	int position = offset ;

   228 	while (remaining > 0) {

   229 	    ByteBuffer buff = (ByteBuffer)buffers.get(0) ; 

   230 	    int dataSize = buff.remaining() ;

   231 	    int xferSize = dataSize ;

   232 	    if (dataSize >= remaining) :

   233 		xferSize = remaining ;

   234 		buffers.remove(0) ;

   235 	    }

   236 	    

   237 	    buff.get( buf, offset, xferSize ) ;

   238 

   239 	    offset += xferSize ;

   240 	    remaining -= xferSize ;

   241 	}

   242     }    

   243 

   244     public void write(ByteBuffer byteBuffer)

   245 	throws IOException

   246     {

   247 	buffers.add( byteBuffer ) ;

   248     }

   249 

   250     /**

   251      * Note:it is possible for this to be called more than once

   252      */

   253     public synchronized void close() 

   254     {

   255     }

   256 

   257     public Acceptor getAcceptor()

   258     {

   259 	return null;

   260     }

   261 

   262     public ContactInfo getContactInfo()

   263     {

   264 	return null;

   265     }

   266 

   267     public EventHandler getEventHandler()

   268     {

   269 	return this;

   270     }

   271 

   272     public OutputObject createOutputObject(MessageMediator messageMediator)

   273     {

   274 	// REVISIT - remove this method from Connection and all it subclasses.

   275 	throw new RuntimeException("*****SocketOrChannelConnectionImpl.createOutputObject - should not be called.");

   276     }

   277 

   278     // This is used by the GIOPOutputObject in order to

   279     // throw the correct error when handling code sets.

   280     // Can we determine if we are on the server side by

   281     // other means?  XREVISIT

   282     public boolean isServer()

   283     {

   284         return isServer;

   285     }

   286 

   287     public boolean isBusy()

   288     {

   289 	return false ;

   290     }

   291 

   292     public long getTimeStamp()

   293     {

   294 	return timeStamp;

   295     }

   296 

   297     public void setTimeStamp(long time)

   298     {

   299 	timeStamp = time;

   300     }

   301 

   302     public void setState(String stateString)

   303     {

   304 	synchronized (stateEvent) {

   305 	    if (stateString.equals("ESTABLISHED")) {

   306 		state =  ESTABLISHED;

   307 		stateEvent.notifyAll();

   308 	    } else {

   309 		// REVISIT: ASSERT

   310 	    }

   311 	}

   312     }

   313 

   314     public void writeLock()

   315     {

   316     }

   317 

   318     public void writeUnlock()

   319     {

   320     }

   321 

   322     public void sendWithoutLock(OutputObject outputObject)

   323     {

   324     }

   325 

   326     public void registerWaiter(MessageMediator messageMediator)

   327     {

   328     }

   329 

   330     public void unregisterWaiter(MessageMediator messageMediator)

   331     {

   332     }

   333 

   334     public InputObject waitForResponse(MessageMediator messageMediator)

   335     {

   336 	return null ;

   337     }

   338 

   339     public void setConnectionCache(ConnectionCache connectionCache)

   340     {

   341     }

   342 

   343     public ConnectionCache getConnectionCache()

   344     {

   345 	return null;	

   346     }

   347 

   348     ////////////////////////////////////////////////////

   349     //

   350     // EventHandler methods

   351     //

   352 

   353     public SelectableChannel getChannel()

   354     {

   355 	return null;

   356     }

   357 

   358     public int getInterestOps()

   359     {

   360 	return null;

   361     }

   362 

   363     //    public Acceptor getAcceptor() - already defined above.

   364 

   365     public Connection getConnection()

   366     {

   367 	return this;

   368     }

   369 

   370     ////////////////////////////////////////////////////

   371     //

   372     // Work methods.

   373     //

   374 

   375     public String getName()

   376     {

   377 	return this.toString();

   378     }

   379 

   380     public void doWork()

   381     {

   382     }

   383 

   384     public void setEnqueueTime(long timeInMillis)

   385     {

   386 	enqueueTime = timeInMillis;

   387     }

   388 

   389     public long getEnqueueTime()

   390     {

   391 	return enqueueTime;

   392     }

   393 

   394     ////////////////////////////////////////////////////

   395     //

   396     // spi.transport.CorbaConnection.

   397     //

   398 

   399     public ResponseWaitingRoom getResponseWaitingRoom()

   400     {

   401 	return null ;

   402     }

   403 

   404     // REVISIT - inteface defines isServer but already defined in 

   405     // higher interface.

   406 

   407 

   408     public void serverRequestMapPut(int requestId, 

   409 				    CorbaMessageMediator messageMediator)

   410     {

   411 	serverRequestMap.put(new Integer(requestId), messageMediator);

   412     }

   413 

   414     public CorbaMessageMediator serverRequestMapGet(int requestId)

   415     {

   416 	return (CorbaMessageMediator)

   417 	    serverRequestMap.get(new Integer(requestId));

   418     }

   419 

   420     public void serverRequestMapRemove(int requestId)

   421     {

   422 	serverRequestMap.remove(new Integer(requestId));

   423     }

   424 

   425 

   426     // REVISIT: this is also defined in:

   427     // com.sun.corba.se.spi.legacy.connection.Connection

   428     public java.net.Socket getSocket()

   429     {

   430 	return null;

   431     }

   432 

   433     /** It is possible for a Close Connection to have been

   434      ** sent here, but we will not check for this. A "lazy"

   435      ** Exception will be thrown in the Worker thread after the

   436      ** incoming request has been processed even though the connection

   437      ** is closed before the request is processed. This is o.k because

   438      ** it is a boundary condition. To prevent it we would have to add

   439      ** more locks which would reduce performance in the normal case.

   440      **/

   441     public synchronized void serverRequestProcessingBegins()

   442     {

   443         serverRequestCount++;

   444     }

   445 

   446     public synchronized void serverRequestProcessingEnds()

   447     {

   448         serverRequestCount--;

   449     }

   450 

   451     //

   452     //

   453     //

   454 

   455     public synchronized int getNextRequestId() 

   456     {

   457 	return requestId++;

   458     }

   459 

   460     // Negotiated code sets for char and wchar data

   461     protected CodeSetComponentInfo.CodeSetContext codeSetContext = null;

   462 

   463     public ORB getBroker() 

   464     {

   465         return orb;

   466     }

   467 

   468     public CodeSetComponentInfo.CodeSetContext getCodeSetContext() 

   469     {

   470         // Needs to be synchronized for the following case when the client

   471         // doesn't send the code set context twice, and we have two threads

   472         // in ServerRequestDispatcher processCodeSetContext.

   473         //

   474         // Thread A checks to see if there is a context, there is none, so

   475         //     it calls setCodeSetContext, getting the synch lock.

   476         // Thread B checks to see if there is a context.  If we didn't synch,

   477         //     it might decide to outlaw wchar/wstring.

   478         if (codeSetContext == null) {

   479             synchronized(this) {

   480                 return codeSetContext;

   481             }

   482         }

   483 

   484         return codeSetContext;

   485     }

   486 

   487     public synchronized void setCodeSetContext(CodeSetComponentInfo.CodeSetContext csc) {

   488         // Double check whether or not we need to do this

   489         if (codeSetContext == null) {

   490             

   491             if (OSFCodeSetRegistry.lookupEntry(csc.getCharCodeSet()) == null ||

   492                 OSFCodeSetRegistry.lookupEntry(csc.getWCharCodeSet()) == null) {

   493                 // If the client says it's negotiated a code set that

   494                 // isn't a fallback and we never said we support, then

   495                 // it has a bug.

   496 		throw wrapper.badCodesetsFromClient() ;

   497             }

   498 

   499             codeSetContext = csc;

   500         }

   501     }

   502 

   503     //

   504     // from iiop.IIOPConnection.java

   505     //

   506 

   507     // Map request ID to an InputObject.

   508     // This is so the client thread can start unmarshaling

   509     // the reply and remove it from the out_calls map while the

   510     // ReaderThread can still obtain the input stream to give

   511     // new fragments.  Only the ReaderThread touches the clientReplyMap,

   512     // so it doesn't incur synchronization overhead.

   513 

   514     public MessageMediator clientRequestMapGet(int requestId)

   515     {

   516 	return null ;

   517     }

   518 

   519     protected MessageMediator clientReply_1_1;

   520 

   521     public void clientReply_1_1_Put(MessageMediator x)

   522     {

   523 	clientReply_1_1 = x;

   524     }

   525 

   526     public MessageMediator clientReply_1_1_Get()

   527     {

   528 	return 	clientReply_1_1;

   529     }

   530 

   531     public void clientReply_1_1_Remove()

   532     {

   533 	clientReply_1_1 = null;

   534     }

   535 

   536     protected MessageMediator serverRequest_1_1;

   537 

   538     public void serverRequest_1_1_Put(MessageMediator x)

   539     {

   540 	serverRequest_1_1 = x;

   541     }

   542 

   543     public MessageMediator serverRequest_1_1_Get()

   544     {

   545 	return 	serverRequest_1_1;

   546     }

   547 

   548     public void serverRequest_1_1_Remove()

   549     {

   550 	serverRequest_1_1 = null;

   551     }

   552 

   553     protected String getStateString( int state ) 

   554     {

   555         synchronized ( stateEvent ){

   556             switch (state) {

   557             case OPENING : return "OPENING" ;

   558             case ESTABLISHED : return "ESTABLISHED" ;

   559             case CLOSE_SENT : return "CLOSE_SENT" ;

   560             case CLOSE_RECVD : return "CLOSE_RECVD" ;

   561             case ABORT : return "ABORT" ;

   562             default : return "???" ;

   563             }

   564         }

   565     }

   566     

   567     public synchronized boolean isPostInitialContexts() {

   568         return postInitialContexts;

   569     }

   570 

   571     // Can never be unset...

   572     public synchronized void setPostInitialContexts(){

   573         postInitialContexts = true;

   574     }

   575     

   576     /**

   577      * Wake up the outstanding requests on the connection, and hand them

   578      * COMM_FAILURE exception with a given minor code.

   579      *

   580      * Also, delete connection from connection table and

   581      * stop the reader thread.

   582 

   583      * Note that this should only ever be called by the Reader thread for

   584      * this connection.

   585      * 

   586      * @param minor_code The minor code for the COMM_FAILURE major code.

   587      * @param die Kill the reader thread (this thread) before exiting.

   588      */

   589     public void purgeCalls(SystemException systemException,

   590 			   boolean die, boolean lockHeld)

   591     {

   592     }

   593 

   594     /*************************************************************************

   595     * The following methods are for dealing with Connection cleaning for

   596     * better scalability of servers in high network load conditions.

   597     **************************************************************************/

   598 

   599     public void sendCloseConnection(GIOPVersion giopVersion)

   600 	throws IOException 

   601     {

   602         Message msg = MessageBase.createCloseConnection(giopVersion);

   603 	sendHelper(giopVersion, msg);

   604     }

   605 

   606     public void sendMessageError(GIOPVersion giopVersion)

   607 	throws IOException 

   608     {

   609         Message msg = MessageBase.createMessageError(giopVersion);

   610 	sendHelper(giopVersion, msg);

   611     }

   612 

   613     /**

   614      * Send a CancelRequest message. This does not lock the connection, so the

   615      * caller needs to ensure this method is called appropriately.

   616      * @exception IOException - could be due to abortive connection closure.

   617      */

   618     public void sendCancelRequest(GIOPVersion giopVersion, int requestId)

   619 	throws IOException 

   620     {

   621 

   622         Message msg = MessageBase.createCancelRequest(giopVersion, requestId);

   623 	sendHelper(giopVersion, msg);

   624     }

   625 

   626     protected void sendHelper(GIOPVersion giopVersion, Message msg)

   627 	throws IOException

   628     {

   629 	// REVISIT: See comments in CDROutputObject constructor.

   630         CDROutputObject outputObject = 

   631 	    new CDROutputObject((ORB)orb, null, giopVersion, this, msg,

   632 				ORBConstants.STREAM_FORMAT_VERSION_1);

   633         msg.write(outputObject);

   634 

   635 	outputObject.writeTo(this);

   636     }

   637 

   638     public void sendCancelRequestWithLock(GIOPVersion giopVersion,

   639 					  int requestId)

   640 	throws IOException 

   641     {

   642 	writeLock();

   643 	try {

   644 	    sendCancelRequest(giopVersion, requestId);

   645 	} finally {

   646 	    writeUnlock();

   647 	}

   648     }

   649 

   650     // Begin Code Base methods ---------------------------------------

   651     //

   652     // Set this connection's code base IOR.  The IOR comes from the

   653     // SendingContext.  This is an optional service context, but all

   654     // JavaSoft ORBs send it.

   655     //

   656     // The set and get methods don't need to be synchronized since the

   657     // first possible get would occur during reading a valuetype, and

   658     // that would be after the set.

   659 

   660     // Sets this connection's code base IOR.  This is done after

   661     // getting the IOR out of the SendingContext service context.

   662     // Our ORBs always send this, but it's optional in CORBA.

   663 

   664     public final void setCodeBaseIOR(IOR ior) {

   665         codeBaseServerIOR = ior;

   666     }

   667 

   668     public final IOR getCodeBaseIOR() {

   669         return codeBaseServerIOR;

   670     }

   671 

   672     // Get a CodeBase stub to use in unmarshaling.  The CachedCodeBase

   673     // won't connect to the remote codebase unless it's necessary.

   674     public final CodeBase getCodeBase() {

   675         return cachedCodeBase;

   676     }

   677 

   678     // End Code Base methods -----------------------------------------

   679 

   680     // Can be overridden in subclass for different options.

   681     protected void setSocketOptions(Socket socket)

   682     {

   683     }

   684 

   685     public String toString()

   686     {

   687         synchronized ( stateEvent ){

   688             return 

   689 		"BufferConnectionImpl[" + " "

   690 		+ getStateString( state ) + " "

   691 		+ shouldUseSelectThreadToWait() + " "

   692 		+ shouldUseWorkerThreadForEvent()

   693 		+ "]" ;

   694         }

   695     }

   696     

   697     // Must be public - used in encoding.

   698     public void dprint(String msg) 

   699     {

   700 	ORBUtility.dprint("SocketOrChannelConnectionImpl", msg);

   701     }

   702 

   703     protected void dprint(String msg, Throwable t)

   704     {

   705 	dprint(msg);

   706 	t.printStackTrace(System.out);

   707     }

   708 }

   709 

   710 // End of file.