/*
 * Copyright (c) 2001, 2012, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.  Oracle designates this
 * particular file as subject to the "Classpath" exception as provided
 * by Oracle in the LICENSE file that accompanied this code.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */

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

import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

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.
    final private Map<Integer, OutCallDesc> out_calls;

    public CorbaResponseWaitingRoomImpl(ORB orb, CorbaConnection connection)
    {
        this.orb = orb;
        wrapper = ORBUtilSystemException.get( orb,
            CORBALogDomains.RPC_TRANSPORT ) ;
        this.connection = connection;
        out_calls =
            Collections.synchronizedMap(new HashMap<Integer, OutCallDesc>());
    }

    ////////////////////////////////////////////////////
    //
    // 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 = 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 = 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()
    {
        return out_calls.size();
    }

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

    public void signalExceptionToAllWaiters(SystemException systemException)
    {

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

        synchronized (out_calls) {
            if (orb.transportDebugFlag) {
                dprint(".signalExceptionToAllWaiters: out_calls size :" + 
                       out_calls.size());
            }

            for (OutCallDesc call : out_calls.values()) { 
                if (orb.transportDebugFlag) {
                    dprint(".signalExceptionToAllWaiters: signaling " + 
                            call);
                }
                synchronized(call.done) {
                    try {
                        // 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);
                        }
                    } catch (Exception e) {
                    } finally {
                        // attempt to wake up waiting threads in all cases
                        call.inputObject = null;
                        call.exception = systemException;
                        call.done.notifyAll();
                    }
                }
            }
        }
    }

    public MessageMediator getMessageMediator(int requestId)
    {
        Integer id = new Integer(requestId);
        OutCallDesc call = 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.