/**********************************************************************
// @@@ START COPYRIGHT @@@
//
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.
//
// @@@ END COPYRIGHT @@@
**********************************************************************/
/* -*-C++-*-
*****************************************************************************
*
* File:         UdrStreams.cpp
* Description:  Server-side message stream classes for UDRs
* Created:      06/10/2001
* Language:     C++
*
*
*****************************************************************************
*/

#include "UdrStreams.h"
#include "udrglobals.h"
#include "udrdefs.h"
#include "udrutil.h"
#include "sql_buffer.h"
#include "UdrDebug.h"
#include "spinfo.h"
#include "udrdecs.h"

#include "rosetta/rosgen.h"
#include "nsk/nskprocess.h"
#include "zsysc.h"

extern void processAnInvokeMessage(UdrGlobals *UdrGlob,
                                   UdrServerDataStream &msgStream,
                                   UdrDataBuffer &request);

extern void processAnRSFetchMessage(UdrGlobals *udrGlob,
                                    UdrServerDataStream &msgStream,
				    UdrHandle udrHandle,
				    RSHandle rsHandle,
                                    UdrDataBuffer *request);

extern void processAnRSContinueMessage(UdrGlobals *udrGlob,
                                       UdrServerDataStream &msgStream,
                                       UdrRSContinueMsg &request);

void UdrServerDataStream::actOnReceive(IpcConnection *conn)
{
  const char *moduleName = "UdrServerDataStream::actOnReceive()";

  IpcMessageObjType t;
  NABoolean somethingArrived = FALSE;

  //
  // A note about control flow in this method. It was originally
  // written under the assumption that only one message arrives at a
  // time. It would call getNextReceiveMsg() only once, then process
  // incoming objects, then return. It turns out this creates a memory
  // leak. Internally within the stream, the incoming message buffer
  // never moved from the "receive" list to the "in use" list and
  // never got cleaned up. Now this method calls getNextReceiveMsg()
  // until it returns FALSE. The result is that after the incoming
  // message has been processed, the next getNextReceiveMsg() call
  // will move the request buffer to the "in use" list where it later
  // gets cleaned up by a call to cleanupBuffers() or
  // releaseBuffers().
  //

  while (getNextReceiveMsg(t))
  {
    somethingArrived = TRUE;

    while (getNextObjType(t))
    {
      switch (t)
      {
        case UDR_MSG_DATA_HEADER:
        {
          // Do nothing for now except extract the object from the stream
          UdrDataHeader *h = new (receiveMsgObj()) UdrDataHeader(this);
          
        } // case UDR_MSG_DATA_HEADER
        break;
        
        case UDR_MSG_CONTINUE_REQUEST:
        {
          // Extract the object from the stream and call SPInfo's work()
          UdrContinueMsg *m = new (receiveMsgObj()) UdrContinueMsg(this);
          
          switch (spinfo_->getParamStyle())
            {
            case COM_STYLE_SQLROW_TM:
            case COM_STYLE_CPP_OBJ:
            case COM_STYLE_JAVA_OBJ:
              spinfo_->workTM();
              break;

            default:
              spinfo_->work();
              break;
            }
          
        } // case UDR_MSG_CONTINUE_REQUEST
        break;
        
        case UDR_MSG_DATA_REQUEST:
        {
          UdrDataBuffer *request = new (receiveMsgObj()) UdrDataBuffer(this);
          
          if (udrGlob_->verbose_ &&
              udrGlob_->traceLevel_ >= TRACE_DETAILS &&
              udrGlob_->showMain_)
          {
            ServerDebug( "");
            ServerDebug("[UdrServ (%s)] Invoke Request has %ld tupps.",
                        moduleName,
                        request->getSqlBuffer()->getTotalTuppDescs());
            ServerDebug("[UdrServ (%s)] Invoke Request SQL Buffer",
                        moduleName);
            displaySqlBuffer(request->getSqlBuffer(),
                             (Lng32) request->getSqlBufferLength());
          }
          
          udrGlob_->numReqInvokeSP_++;
          
          // Let the SPInfo process the request
	  spinfo_->setCurrentRequest(request);
	  spinfo_->work();

        } // case UDR_MSG_DATA_REQUEST
        break;

        case UDR_MSG_TMUDF_DATA_HEADER:
	{
	  // Extract the object. UDR Handle and RS Handle are
	  // interesting things in this message.
          UdrTmudfDataHeader *h = new (receiveMsgObj()) UdrTmudfDataHeader(this);
	  UdrHandle udrHandle = h->getHandle();

          NABoolean anyMore = getNextObjType(t);
          UDR_ASSERT(anyMore, "DATA_REQUEST must follow TMUDF DATA_HEADER");
          UDR_ASSERT(t == UDR_MSG_DATA_REQUEST,
                     "DATA_REQUEST must follow TMUDF DATA_HEADER");

          UdrDataBuffer *request = new (receiveMsgObj()) 
                                    UdrDataBuffer(this,
                                                  FALSE); //Avoid unpack.

          // Let the SPInfo process the request
	  spinfo_->setCurrentRequest(request);
	  spinfo_->workTM();
	}
        break;

        case UDR_MSG_RS_DATA_HEADER:
	{
	  // Extract the object. UDR Handle and RS Handle are
	  // interesting things in this message.
          UdrRSDataHeader *h = new (receiveMsgObj()) UdrRSDataHeader(this);
	  UdrHandle udrHandle = h->getHandle();
	  RSHandle rsHandle = h->getRSHandle();

          NABoolean anyMore = getNextObjType(t);
          UDR_ASSERT(anyMore, "DATA_REQUEST must follow RS_DATA_HEADER");
          UDR_ASSERT(t == UDR_MSG_DATA_REQUEST,
                     "DATA_REQUEST must follow RS_DATA_HEADER");

          UdrDataBuffer *request = new (receiveMsgObj()) UdrDataBuffer(this);

	  udrGlob_->numReqRSFetch_++;
	  processAnRSFetchMessage(udrGlob_, *this, udrHandle,
                                  rsHandle, request);

          // We need the request buffer in a state where the stream
          // knows it can be freed. We call SqlBuffer::bufferFull() to
          // accomplish this even though the method name is a bit
          // misleading.
          SqlBuffer *sqlBuf = request->getSqlBuffer();
          UDR_ASSERT(sqlBuf,
                     "UDR request buffer is corrupt or contains no data");
          sqlBuf->bufferFull();
	}
        break;

        case UDR_MSG_RS_CONTINUE:
	{
	  UdrRSContinueMsg *rsContinue =
            new (receiveMsgObj()) UdrRSContinueMsg(this);

	  udrGlob_->numReqRSContinue_++;
	  processAnRSContinueMessage(udrGlob_, *this, *rsContinue);
	}
	break;

        default:
        {
          UDR_ASSERT(FALSE,
                     "Unknown message type arrived on UDR data stream");
        } // default
        break;
        
      } // switch (t)
      
    } // while (getNextObjType(t))
  } // while (getNextReceiveMsg(t))

  // Make sure all reply buffers have been given to the connection. If
  // the only objects that arrived during this callback were continue
  // requests, then this action allows reply buffers associated with
  // those continue requests to propagate out of the stream and onto
  // the connection.
  //
  // If numOfInputBuffers() is > 0 then we do not need to do anything
  // at this point. This callback will be invoked again when the
  // incoming message is complete and ready to be processed.
  Lng32 numInputBuffers = numOfInputBuffers();
  if (somethingArrived && numInputBuffers == 0 &&
      spinfo_->getCurrentRequest() == NULL)
  {
    responseDone();

    if (udrGlob_->verbose_ &&
        udrGlob_->traceLevel_ >= TRACE_IPMS &&
        udrGlob_->showMain_)
    {
      ServerDebug("[UdrServ (%s)] All messages marked as replied", moduleName);
    }

    // Cleanup any unpacked message buffers containing only objects
    // that are no longer in use, as determined by the virtual method
    // IpcMessageObj::msgObjIsFree()
    releaseBuffers();  // Do final garbage collection
  }
  
} // UdrServerDataStream::actOnReceive()

void
UdrGuaControlConnection::actOnSystemMessage(
  short                  messageNum,
  IpcMessageBufferPtr    sysMsg,
  IpcMessageObjSize      sysMsgLen,
  short                  clientFileNumber,
  const GuaProcessHandle &clientPhandle,
  GuaConnectionToClient  *connection)
{
  const char * moduleName = "UdrGuaControlConnection::actOnSystemMessage";

  switch (messageNum)
  {
    case ZSYS_VAL_SMSG_OPEN:
      {
        if (udrGlob_->verbose_ &&
            udrGlob_->traceLevel_ >= TRACE_IPMS &&
            udrGlob_->showMain_)
        {
          ServerDebug(
           "[UdrServ (%s)] A new connection %p is opened to the Server.",
           moduleName,
           connection);
        }

        // Create a new message stream. Link the connection.
        // Then call receive on it to bring it into receiving mode.
        UdrServerControlStream *newControlStream =
          new (udrGlob_->getIpcHeap())
            UdrServerControlStream(udrGlob_->getIpcEnvironment(),
                                   udrGlob_,
                                   UDR_STREAM_SERVER_CONTROL,
                                   UdrServerControlStreamVersionNumber);
        newControlStream->addRecipient(connection);
        newControlStream->receive(FALSE);
      }
      break;

    default:
      // do nothing for all other kinds of system messages
      break;
  } // switch

  // The parent class already handles the job of closing all connections
  // who lost their client process by failed processes, failed CPUs and
  // failed systems or networks. Check here that we die if all our
  // requestors go away, but don't die if the first system message is
  // something other than an OPEN message.
  if (getNumRequestors() == 0 AND initialized_)
  {
    NAExit(0);
  }
  else if (NOT initialized_ AND getNumRequestors() > 0)
  {
    // the first requestor came in
    initialized_ = TRUE;
  }

} // UdrGuaControlConnection::actOnSystemMessage()

void
UdrServerControlStream::actOnReceive(IpcConnection *conn)
{
  const char *moduleName = "UdrServerControlStream::actOnReceive";

  NABoolean doTrace = (udrGlob_->verbose_ &&
                       udrGlob_->traceLevel_ >= TRACE_IPMS &&
                       udrGlob_->showMain_) ? TRUE : FALSE;
  if (doTrace)
  {
    ServerDebug("[UdrServ (%s)]", moduleName);
    ServerDebug("        A new message arrived on connection %p.", conn);
  }

  // Create a new stream and give the message to that stream
  UdrServerReplyStream *newReplyStream =
    new (udrGlob_->getUdrHeap()) UdrServerReplyStream(
      udrGlob_->getIpcEnvironment(),
      udrGlob_,
      UDR_STREAM_SERVER_REPLY,
      UdrServerReplyStreamVersionNumber);

  giveMessageTo(*newReplyStream, conn);

  // Add the stream to Udr Globals list of streams
  udrGlob_->addReplyStream(newReplyStream);


  if (doTrace)
  {
    ServerDebug("        The new message is scheduled for processing.");
    ServerDebug("        The connection %p will be placed on receive mode.",
                 conn);
    ServerDebug("[UdrServ (%s)] Done", moduleName);
  }

  // We have retrieved the message from stream, make it ready for
  // next message.
  receive(FALSE);

} // UdrServerControlStream::actOnReceive()


void UdrServerDataStream::activateCurrentMsgTransaction()
{
  udrGlob_->getControlConnection()->setUserTransReplyTag(getReplyTag());
  return;
}

void UdrServerReplyStream::routeMessage(UdrServerDataStream &other)
{
  // Store reply tag in data stream before routing message to it.
  other.setReplyTag(getReplyTag());
  giveReceiveMsgTo(other);
}

void UdrServerReplyStream::activateCurrentMsgTransaction()
{
  udrGlob_->getControlConnection()->setUserTransReplyTag(getReplyTag());
  return;
}