binarymessagehandler.cpp

Pegasus is an open-source implementationof the DMTF CIM and WBEM standards. It is designed to be por

//%2006////////////////////////////////////////////////////////////////////////
//
// Copyright (c) 2000, 2001, 2002 BMC Software; Hewlett-Packard Development
// Company, L.P.; IBM Corp.; The Open Group; Tivoli Systems.
// Copyright (c) 2003 BMC Software; Hewlett-Packard Development Company, L.P.;
// IBM Corp.; EMC Corporation, The Open Group.
// Copyright (c) 2004 BMC Software; Hewlett-Packard Development Company, L.P.;
// IBM Corp.; EMC Corporation; VERITAS Software Corporation; The Open Group.
// Copyright (c) 2005 Hewlett-Packard Development Company, L.P.; IBM Corp.;
// EMC Corporation; VERITAS Software Corporation; The Open Group.
// Copyright (c) 2006 Hewlett-Packard Development Company, L.P.; IBM Corp.;
// EMC Corporation; Symantec Corporation; The Open Group.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to
// deal in the Software without restriction, including without limitation the
// rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
// sell copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
// 
// THE ABOVE COPYRIGHT NOTICE AND THIS PERMISSION NOTICE SHALL BE INCLUDED IN
// ALL COPIES OR SUBSTANTIAL PORTIONS OF THE SOFTWARE. THE SOFTWARE IS PROVIDED
// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
//
//==============================================================================
//
//%/////////////////////////////////////////////////////////////////////////////

#include "BinaryMessageHandler.h"
#include <Pegasus/Common/Thread.h>

PEGASUS_NAMESPACE_BEGIN

BinaryMessageHandler::BinaryMessageHandler(MessageQueueService* output_q)
   : Base(PEGASUS_QUEUENAME_BINARY_HANDLER),
     _outputQueue(output_q),
     _msg_q()
{
}

BinaryMessageHandler::~BinaryMessageHandler()
{
}

Boolean BinaryMessageHandler::messageOK(const Message* msg)
{
    return Base::messageOK(msg);
}

void BinaryMessageHandler::handleEnqueue()
{
    Message* msg = dequeue();
    handleEnqueue(msg);
}

void BinaryMessageHandler::handleEnqueue(Message* message)
{
    PEG_METHOD_ENTER(TRC_BINARY_MSG_HANDLER,
        "BinaryMessageHandler::handleEnqueue(Message*)");
    PEGASUS_ASSERT(message != 0);

    AsyncRequest* asyncRequest;
    AsyncOpNode* op;

    message->_async = 0;

    PEG_TRACE_STRING(TRC_BINARY_MSG_HANDLER, Tracer::LEVEL4,
        "Converting legacy message to AsyncLegacyOperationStart");
    op = this->get_op();
    asyncRequest = new AsyncLegacyOperationStart(
        op,
        this->getQueueId(),
        message,
        this->getQueueId());
    op->_flags |= ASYNC_OPFLAGS_FIRE_AND_FORGET;

    _handle_async_request(asyncRequest);
    PEG_METHOD_EXIT();
}

void BinaryMessageHandler::_handle_async_request(AsyncRequest* request)
{
    PEG_METHOD_ENTER(TRC_BINARY_MSG_HANDLER,
        "BinaryMessageHandler::_handle_async_request");

    PEGASUS_ASSERT(request != 0 && request->op != 0 );

    if (request->getType() == async_messages::ASYNC_LEGACY_OP_START ||
        request->getType() == async_messages::ASYNC_LEGACY_OP_RESULT)
    {
        PEG_TRACE_STRING(TRC_BINARY_MSG_HANDLER, Tracer::LEVEL4,
            "Processing ASYNC_LEGACY_OP_* Message.");
        request->op->processing();

        try
        {
            _msg_q.enqueue(request->op);
        }
        catch (ListFull&)
        {
            PEG_TRACE_STRING(TRC_BINARY_MSG_HANDLER, Tracer::LEVEL4,
                "List Full.");
            Base::_handle_async_request(request);
            PEG_METHOD_EXIT();
            return;
        }
        catch (...)
        {
            PEG_TRACE_STRING(TRC_BINARY_MSG_HANDLER, Tracer::LEVEL4,
                "List Error.");
            Base::_handle_async_request(request);
            PEG_METHOD_EXIT();
            return;
        }

        PEG_TRACE_STRING(TRC_BINARY_MSG_HANDLER, Tracer::LEVEL4,
            "Allocating pooled thread to handle binary message.");
        if (_thread_pool->allocate_and_awaken(
                (void*)this, BinaryMessageHandler::handle_binary_message) !=
            PEGASUS_THREAD_OK)
        {
            Logger::put(Logger::STANDARD_LOG, System::CIMSERVER, Logger::TRACE,
                "Not enough threads to handle binary message.");

            Tracer::trace(TRC_BINARY_MSG_HANDLER, Tracer::LEVEL2,
                "Could not allocate thread for %s. " \
                "Queue has %d messages waiting. ",
                getQueueName(),
                _msg_q.count());
        }
    }
    else if (request->getType() == async_messages::CIMSERVICE_STOP)
    {
        PEG_TRACE_STRING(TRC_BINARY_MSG_HANDLER, Tracer::LEVEL4,
            "Handling CIMServer Stop Message");
        Base::_handle_async_request(request);
    }
    else
    {
        // pass all other operations to the default handler
        PEG_TRACE_STRING(TRC_BINARY_MSG_HANDLER, Tracer::LEVEL4,
            "Passing message to parent.");
#ifdef BINARYMESSAGEHANDLER_DEBUG
        PEGASUS_STD(cout) << "Unexpected Message: type " <<
            request->getType() << PEGASUS_STD(endl);
#endif

        Base::_handle_async_request(request);
    }

    PEG_METHOD_EXIT();
}

ThreadReturnType PEGASUS_THREAD_CDECL
BinaryMessageHandler::handle_binary_message(void* parm)
{
    PEG_METHOD_ENTER(TRC_BINARY_MSG_HANDLER,
        "BinaryMessageHandler::handle_binary_message");

    BinaryMessageHandler* myself =
        reinterpret_cast<BinaryMessageHandler*>(parm);
    PEGASUS_ASSERT(myself != 0);

    AsyncOpNode* op;
    try
    {
        op = myself->_msg_q.dequeue();
    }
    catch (...)
    {
        PEG_TRACE_STRING(TRC_BINARY_MSG_HANDLER, Tracer::LEVEL2,
            "Internal DQueue Error.");
        PEG_METHOD_EXIT();
        return 0;
    }

    PEGASUS_ASSERT(op != 0);

    // we only receive ASYNC_LEGACY_OP_START and
    // ASYNC_LEGACY_OP_RESULT messages

    Message* legacy = 0;
    AsyncMessage* msg = 0;

    try
    {
        // if there is a response, dispatch the response
        if (op->_response.get() != 0)
        {
            msg = static_cast<AsyncMessage*>(op->_response.get());
            legacy =
                static_cast<AsyncLegacyOperationResult*>(msg)->get_result();
        }
        else
        {
            // there is no response so there has to be a request
            if (op->_request.get() == 0)
            {
                PEG_TRACE_STRING(TRC_BINARY_MSG_HANDLER, Tracer::LEVEL2,
                    "Received OpNode with no messages.");
                PEG_METHOD_EXIT();
                return 0;
            }

            // dispatch the request
            msg = static_cast<AsyncMessage*>(op->_request.get());
            legacy =
                static_cast<AsyncLegacyOperationStart*>(msg)->get_action();
        }
        if (msg && legacy)
        {
            legacy->_async = 0;

            switch(legacy->getType())
            {
                case CIM_CREATE_CLASS_REQUEST_MESSAGE:
                    myself->handleCreateClassRequest(
                        op, (CIMCreateClassRequestMessage*)legacy);
                    break;
                case CIM_CREATE_INSTANCE_REQUEST_MESSAGE:
                    myself->handleCreateInstanceRequest(
                        op, (CIMCreateInstanceRequestMessage*)legacy);
                    break;
                case CIM_MODIFY_CLASS_REQUEST_MESSAGE:
                    myself->handleModifyClassRequest(
                        op, (CIMModifyClassRequestMessage*)legacy);
                    break;
            case CIM_MODIFY_INSTANCE_REQUEST_MESSAGE:
                    myself->handleModifyInstanceRequest(
                        op, (CIMModifyInstanceRequestMessage*)legacy);
                    break;
            case CIM_ENUMERATE_CLASSES_REQUEST_MESSAGE:
                    myself->handleEnumerateClassesRequest(
                        op, (CIMEnumerateClassesRequestMessage*)legacy);
                    break;
            case CIM_ENUMERATE_CLASS_NAMES_REQUEST_MESSAGE:
                    myself->handleEnumerateClassNamesRequest(
                        op, (CIMEnumerateClassNamesRequestMessage*)legacy);
                    break;
            case CIM_ENUMERATE_INSTANCES_REQUEST_MESSAGE:
                    myself->handleEnumerateInstancesRequest(
                        op, (CIMEnumerateInstancesRequestMessage*)legacy);
                    break;
            case CIM_ENUMERATE_INSTANCE_NAMES_REQUEST_MESSAGE:
                    myself->handleEnumerateInstanceNamesRequest(
                        op, (CIMEnumerateInstanceNamesRequestMessage*)legacy);
                    break;
            case CIM_EXEC_QUERY_REQUEST_MESSAGE:
                    myself->handleExecQueryRequest(
                        op, (CIMExecQueryRequestMessage*)legacy);
                    break;
            case CIM_ASSOCIATORS_REQUEST_MESSAGE:
                    myself->handleAssociatorsRequest(
                        op, (CIMAssociatorsRequestMessage*)legacy);
                    break;
            case CIM_ASSOCIATOR_NAMES_REQUEST_MESSAGE:
                    myself->handleAssociatorNamesRequest(
                        op, (CIMAssociatorNamesRequestMessage*)legacy);
                    break;
            case CIM_REFERENCES_REQUEST_MESSAGE:
                    myself->handleReferencesRequest(
                        op, (CIMReferencesRequestMessage*)legacy);
                    break;
            case CIM_REFERENCE_NAMES_REQUEST_MESSAGE:
                    myself->handleReferenceNamesRequest(
                        op, (CIMReferenceNamesRequestMessage*)legacy);
                    break;
            case CIM_GET_PROPERTY_REQUEST_MESSAGE:
                    myself->handleGetPropertyRequest(
                        op, (CIMGetPropertyRequestMessage*)legacy);
                    break;
            case CIM_SET_PROPERTY_REQUEST_MESSAGE:
                    myself->handleSetPropertyRequest(
                        op, (CIMSetPropertyRequestMessage*)legacy);
                    break;
            case CIM_GET_QUALIFIER_REQUEST_MESSAGE:
                    myself->handleGetQualifierRequest(
                        op, (CIMGetQualifierRequestMessage*)legacy);
                    break;
            case CIM_SET_QUALIFIER_REQUEST_MESSAGE:
                    myself->handleSetQualifierRequest(
                        op, (CIMSetQualifierRequestMessage*)legacy);
                    break;
            case CIM_DELETE_QUALIFIER_REQUEST_MESSAGE:
                    myself->handleDeleteQualifiersRequest(
                        op, (CIMDeleteQualifierRequestMessage*)legacy);
                    break;
            case CIM_ENUMERATE_QUALIFIERS_REQUEST_MESSAGE:
                    myself->handleEnumerateQualifiersRequest(
                        op, (CIMEnumerateQualifiersRequestMessage*)legacy);
                    break;
            case CIM_INVOKE_METHOD_REQUEST_MESSAGE:
                    myself->handleInvokeMethodRequest(
                        op, (CIMInvokeMethodRequestMessage*)legacy);
                    break;
            //**** response messages ****//
            case CIM_GET_CLASS_RESPONSE_MESSAGE:
                    myself->handleGetClassResponse(
                        op, (CIMGetClassResponseMessage*)legacy);
                    break;
            case CIM_GET_INSTANCE_RESPONSE_MESSAGE:
                    myself->handleGetInstanceResponse(
                        op, (CIMGetInstanceResponseMessage*)legacy);
                    break;
            case CIM_DELETE_CLASS_RESPONSE_MESSAGE:
                    myself->handleDeleteClassResponse(
                        op, (CIMDeleteClassResponseMessage*)legacy);
                    break;
            case CIM_DELETE_INSTANCE_RESPONSE_MESSAGE:
                    myself->handleDeleteInstanceResponse(
                        op, (CIMDeleteInstanceResponseMessage*)legacy);
                    break;
            case CIM_CREATE_CLASS_RESPONSE_MESSAGE:
                    myself->handleCreateClassResponse(
                        op, (CIMCreateClassResponseMessage*)legacy);
                    break;
            case CIM_CREATE_INSTANCE_RESPONSE_MESSAGE:
                    myself->handleCreateInstanceResponse(
                        op, (CIMCreateInstanceResponseMessage*)legacy);
                    break;
            case CIM_MODIFY_CLASS_RESPONSE_MESSAGE:
                    myself->handleModifyClassResponse(
                        op, (CIMModifyClassResponseMessage*)legacy);
                    break;
            case CIM_MODIFY_INSTANCE_RESPONSE_MESSAGE:
                    myself->handleModifyInstanceResponse(
                        op, (CIMModifyInstanceResponseMessage*)legacy);
                    break;
            case CIM_ENUMERATE_CLASSES_RESPONSE_MESSAGE:
                    myself->handleEnumerateClassesResponse(
                        op, (CIMEnumerateClassesResponseMessage*)legacy);
                    break;
            case CIM_ENUMERATE_CLASS_NAMES_RESPONSE_MESSAGE:
                    myself->handleEnumerateClassNamesResponse(
                        op, (CIMEnumerateClassNamesResponseMessage*)legacy);
                    break;
            case CIM_ENUMERATE_INSTANCES_RESPONSE_MESSAGE:
                    myself->handleEnumerateInstancesResponse(
                        op, (CIMEnumerateInstancesResponseMessage*)legacy);
                    break;