/*
 * Copyright (c) 2001, 2004, Oracle and/or its affiliates. All rights reserved.
 * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 */

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

import java.util.Hashtable;

import org.omg.CORBA.CompletionStatus;
import org.omg.CORBA.SystemException;

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.spi.logging.CORBALogDomains;
import com.sun.corba.se.spi.orb.ORB;
import com.sun.corba.se.spi.protocol.CorbaMessageMediator;
import com.sun.corba.se.spi.transport.CorbaConnection;
import com.sun.corba.se.spi.transport.CorbaResponseWaitingRoom;

import com.sun.corba.se.impl.encoding.BufferManagerReadStream;
import com.sun.corba.se.impl.encoding.CDRInputObject;
import com.sun.corba.se.impl.logging.ORBUtilSystemException;
import com.sun.corba.se.impl.orbutil.ORBUtility;
import com.sun.corba.se.impl.protocol.giopmsgheaders.LocateReplyOrReplyMessage;
import com.sun.corba.se.impl.protocol.giopmsgheaders.ReplyMessage;

/**
 * @author Harold Carr
 */
public class CorbaResponseWaitingRoomImpl
    implements
        CorbaResponseWaitingRoom
{
    final static class OutCallDesc
    {
        java.lang.Object done = new java.lang.Object();
        Thread thread;
        MessageMediator messageMediator;
        SystemException exception;
        InputObject inputObject;
    }

    private ORB orb;
    private ORBUtilSystemException wrapper ;

    private CorbaConnection connection;
    // Maps requestId to an OutCallDesc.
    private Hashtable out_calls = null; // REVISIT - use int hastable/map

    public CorbaResponseWaitingRoomImpl(ORB orb, CorbaConnection connection)
    {
        this.orb = orb;
        wrapper = ORBUtilSystemException.get( orb,
            CORBALogDomains.RPC_TRANSPORT ) ;
        this.connection = connection;
        out_calls = new Hashtable();
    }

    ////////////////////////////////////////////////////
    //
    // pept.transport.ResponseWaitingRoom
    //

    public void registerWaiter(MessageMediator mediator)
    {
        CorbaMessageMediator messageMediator = (CorbaMessageMediator) mediator;

        if (orb.transportDebugFlag) {
            dprint(".registerWaiter: " + opAndId(messageMediator));
        }

        Integer requestId = messageMediator.getRequestIdInteger();

        OutCallDesc call = new OutCallDesc();
        call.thread = Thread.currentThread();
        call.messageMediator = messageMediator;
        out_calls.put(requestId, call);
    }

    public void unregisterWaiter(MessageMediator mediator)
    {
        CorbaMessageMediator messageMediator = (CorbaMessageMediator) mediator;

        if (orb.transportDebugFlag) {
            dprint(".unregisterWaiter: " + opAndId(messageMediator));
        }

        Integer requestId = messageMediator.getRequestIdInteger();

        out_calls.remove(requestId);
    }

    public InputObject waitForResponse(MessageMediator mediator)
    {
      CorbaMessageMediator messageMediator = (CorbaMessageMediator) mediator;

      try {

        InputObject returnStream = null;

        if (orb.transportDebugFlag) {
            dprint(".waitForResponse->: " + opAndId(messageMediator));
        }

        Integer requestId = messageMediator.getRequestIdInteger();

        if (messageMediator.isOneWay()) {
            // The waiter is removed in releaseReply in the same
            // way as a normal request.

            if (orb.transportDebugFlag) {
                dprint(".waitForResponse: one way - not waiting: "
                       + opAndId(messageMediator));
            }

            return null;
        }

        OutCallDesc call = (OutCallDesc)out_calls.get(requestId);
        if (call == null) {
            throw wrapper.nullOutCall(CompletionStatus.COMPLETED_MAYBE);
        }

        synchronized(call.done) {

            while (call.inputObject == null && call.exception == null) {
                // Wait for the reply from the server.
                // The ReaderThread reads in the reply IIOP message
                // and signals us.
                try {
                    if (orb.transportDebugFlag) {
                        dprint(".waitForResponse: waiting: "
                               + opAndId(messageMediator));
                    }
                    call.done.wait();
                } catch (InterruptedException ie) {};
            }

            if (call.exception != null) {
                if (orb.transportDebugFlag) {
                    dprint(".waitForResponse: exception: "
                           + opAndId(messageMediator));
                }
                throw call.exception;
            }

            returnStream = call.inputObject;
        }

        // REVISIT -- exceptions from unmarshaling code will
        // go up through this client thread!

        if (returnStream != null) {
            // On fragmented streams the header MUST be unmarshaled here
            // (in the client thread) in case it blocks.
            // If the header was already unmarshaled, this won't
            // do anything
            // REVISIT: cast - need interface method.
            ((CDRInputObject)returnStream).unmarshalHeader();
        }

        return returnStream;

      } finally {
        if (orb.transportDebugFlag) {
            dprint(".waitForResponse<-: " + opAndId(messageMediator));
        }
      }
    }

    public void responseReceived(InputObject is)
    {
        CDRInputObject inputObject = (CDRInputObject) is;
        LocateReplyOrReplyMessage header = (LocateReplyOrReplyMessage)
            inputObject.getMessageHeader();
        Integer requestId = new Integer(header.getRequestId());
        OutCallDesc call = (OutCallDesc) out_calls.get(requestId);

        if (orb.transportDebugFlag) {
            dprint(".responseReceived: id/"
                   + requestId  + ": "
                   + header);
        }

        // This is an interesting case.  It could mean that someone sent us a
        // reply message, but we don't know what request it was for.  That
        // would probably call for an error.  However, there's another case
        // that's normal and we should think about --
        //
        // If the unmarshaling thread does all of its work inbetween the time
        // the ReaderThread gives it the last fragment and gets to the
        // out_calls.get line, then it will also be null, so just return;
        if (call == null) {
            if (orb.transportDebugFlag) {
                dprint(".responseReceived: id/"
                       + requestId
                       + ": no waiter: "
                       + header);
            }
            return;
        }

        // Set the reply InputObject and signal the client thread
        // that the reply has been received.
        // The thread signalled will remove outcall descriptor if appropriate.
        // Otherwise, it'll be removed when last fragment for it has been put on
        // BufferManagerRead's queue.
        synchronized (call.done) {
            CorbaMessageMediator messageMediator = (CorbaMessageMediator)
                call.messageMediator;

            if (orb.transportDebugFlag) {
                dprint(".responseReceived: "
                       + opAndId(messageMediator)
                       + ": notifying waiters");
            }

            messageMediator.setReplyHeader(header);
            messageMediator.setInputObject(is);
            inputObject.setMessageMediator(messageMediator);
            call.inputObject = is;
            call.done.notify();
        }
    }

    public int numberRegistered()
    {
        // Note: Hashtable.size() is not synchronized
        return out_calls.size();
    }

    //////////////////////////////////////////////////
    //
    // CorbaResponseWaitingRoom
    //

    public void signalExceptionToAllWaiters(SystemException systemException)
    {

        if (orb.transportDebugFlag) {
            dprint(".signalExceptionToAllWaiters: " + systemException);
        }

        OutCallDesc call;
        java.util.Enumeration e = out_calls.elements();
        while(e.hasMoreElements()) {
            call = (OutCallDesc) e.nextElement();

            synchronized(call.done){
                // anything waiting for BufferManagerRead's fragment queue
                // needs to be cancelled
                CorbaMessageMediator corbaMsgMediator =
                             (CorbaMessageMediator)call.messageMediator;
                CDRInputObject inputObject =
                           (CDRInputObject)corbaMsgMediator.getInputObject();
                // IMPORTANT: If inputObject is null, then no need to tell
                //            BufferManagerRead to cancel request processing.
                if (inputObject != null) {
                    BufferManagerReadStream bufferManager =
                        (BufferManagerReadStream)inputObject.getBufferManager();
                    int requestId = corbaMsgMediator.getRequestId();
                    bufferManager.cancelProcessing(requestId);
                }
                call.inputObject = null;
                call.exception = systemException;
                call.done.notify();
            }
        }
    }

    public MessageMediator getMessageMediator(int requestId)
    {
        Integer id = new Integer(requestId);
        OutCallDesc call = (OutCallDesc) out_calls.get(id);
        if (call == null) {
            // This can happen when getting early reply fragments for a
            // request which has completed (e.g., client marshaling error).
            return null;
        }
        return call.messageMediator;
    }

    ////////////////////////////////////////////////////
    //
    // Implementation.
    //

    protected void dprint(String msg)
    {
        ORBUtility.dprint("CorbaResponseWaitingRoomImpl", msg);
    }

    protected String opAndId(CorbaMessageMediator mediator)
    {
        return ORBUtility.operationNameAndRequestId(mediator);
    }
}

// End of file.
