pegasus/src/Pegasus/Common/MessageQueueService.h - diff

Return to MessageQueueService.h CVS log

Up to [Pegasus] / pegasus / src / Pegasus / Common

Diff for /pegasus/src/Pegasus/Common/MessageQueueService.h between version 1.1 and 1.61

version 1.1, 2002/01/15 21:47:13

version 1.61, 2007/07/18 19:46:23

Line 1

//%////-*-c++-*-////////////////////////////////////////////////////////////////

//%2006////////////////////////////////////////////////////////////////////////

// Company, L.P.; IBM Corp.; The Open Group; Tivoli Systems.

// IBM Corp.; EMC Corporation, The Open Group.

// IBM Corp.; EMC Corporation; VERITAS Software Corporation; The Open Group.

// EMC Corporation; VERITAS Software Corporation; The Open Group.

// 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

Line 20

Line 29

//==============================================================================

// Author: Mike Day (mdday@us.ibm.com)

// Modified By:

//%/////////////////////////////////////////////////////////////////////////////

#ifndef Pegasus_MessageQueue_Service_h

Line 31

Line 36

#include <Pegasus/Common/Config.h>

#include <Pegasus/Common/Message.h>

#include <Pegasus/Common/Exception.h>

#include <Pegasus/Common/InternalException.h>

#include <Pegasus/Common/IPC.h>

#include <Pegasus/Common/Thread.h>

#include <Pegasus/Common/ThreadPool.h>

#include <Pegasus/Common/AsyncOpNode.h>

#include <Pegasus/Common/Cimom.h>

#include <Pegasus/Common/Mutex.h>

#include <Pegasus/Common/CimomMessage.h>

#include <Pegasus/Common/Linkage.h>

#include <Pegasus/Common/Mutex.h>

PEGASUS_NAMESPACE_BEGIN

extern const Uint32 CIMOM_Q_ID;

class message_module;

class cimom;

class PEGASUS_COMMON_LINKAGE MessageQueueService : public MessageQueue

class PEGASUS_COMMON_LINKAGE MessageQueueService :

public Linkable, public MessageQueue

{

public:

typedef MessageQueue Base;

MessageQueueService(const char *name, Uint32 queueID, Uint32 capabilities, Uint32 mask) ;

MessageQueueService(

const char* name,

Uint32 queueID = MessageQueue::getNextQueueId(),

Uint32 capabilities = 0,

Uint32 mask = MessageMask::ha_request |

MessageMask::ha_reply |

MessageMask::ha_async);

virtual ~MessageQueueService();

virtual Boolean isAsync() const { return true; }

// enqueue may throw an IPCException

virtual void enqueue(Message *);

AsyncReply* SendWait(AsyncRequest* request);

Boolean SendAsync(AsyncOpNode* op,

Uint32 destination,

void (*callback)(AsyncOpNode*, MessageQueue*, void*),

MessageQueue* callback_q,

void* callback_ptr);

Boolean SendAsync(Message *msg,

Uint32 destination,

void (*callback)(Message* response, void* handle, void* parameter),

void* handle,

void* parameter);

Boolean SendForget(Message* msg);

Boolean register_service(String name, Uint32 capabilities, Uint32 mask);

Boolean update_service(Uint32 capabilities, Uint32 mask);

Boolean deregister_service();

void find_services(

String name,

Uint32 capabilities,

Uint32 mask,

Array<Uint32>* results);

void enumerate_service(Uint32 queue, message_module* result);

static AsyncOpNode* get_op();

void return_op(AsyncOpNode* op);

virtual ~MessageQueueService(void);

static ThreadPool* get_thread_pool();

Uint32 _mask;

AtomicInt _die;

AtomicInt _threads;

Uint32 getIncomingCount() {return _incoming.count(); }

protected:

virtual void _shutdown_incoming_queue();

virtual Boolean accept_async(AsyncOpNode* op);

virtual Boolean messageOK(const Message* msg);

virtual void handleEnqueue() = 0;

virtual void handleEnqueue(Message *) = 0;

Boolean _enqueueResponse(Message *, Message *);

// virtual void _handle_incoming_operation(

// AsyncOpNode* operation,

// Thread* thread,

// MessageQueue* queue);

virtual void _handle_incoming_operation(AsyncOpNode *);

virtual void _handle_async_request(AsyncRequest* req);

virtual void _handle_async_callback(AsyncOpNode* operation);

virtual void _make_response(Message* req, Uint32 code);

// don't allow derived classes to override

void handleEnqueue();

virtual void handle_heartbeat_request(AsyncRequest *req);

virtual void handle_heartbeat_reply(AsyncReply *rep);

Line 63

Line 139

virtual void handle_CimServiceResume(CimServiceResume *req);

virtual void handle_AsyncOperationStart(AsyncOperationStart *req);

virtual void handle_AsyncOperationResult(AsyncOperationResult *req);

virtual void handle_AsyncOperationResult(AsyncOperationResult* rep);

virtual void handle_AsyncLegacyOperationStart(

AsyncLegacyOperationStart* req);

virtual void handle_AsyncLegacyOperationResult(

AsyncLegacyOperationResult* rep);

// virtual Boolean accept_async(Message *message) throw(IPCException);

void _completeAsyncResponse(

// virtual Boolean messageOK(const Message *msg) ;

AsyncRequest* request,

// virtual Message *openEnvelope(Message *msg);

void SendWait(AsyncRequest *request, unlocked_dq<AsyncMessage> *reply_list);

Boolean SendAsync(AsyncMessage *msg);

void _enqueueAsyncResponse(AsyncRequest *request,

AsyncReply *reply,

Uint32 state,

Uint32 flag);

Boolean register_service(String name, Uint32 capabilities, Uint32 mask);

void _complete_op_node(AsyncOpNode *, Uint32, Uint32, Uint32);

Boolean update_service(Uint32 capabilities, Uint32 mask);

Boolean deregister_service(void);

void find_services(String name,

static cimom* _meta_dispatcher;

Uint32 capabilities,

static AtomicInt _service_count;

Uint32 mask,

static Mutex _meta_dispatcher_mutex;

Array<Uint32> *results);

static ThreadPool* _thread_pool;

void enumerate_service(Uint32 queue, message_module *result);

Uint32 get_next_xid(void);

AsyncOpNode *get_op(void);

void return_op(AsyncOpNode *op);

Uint32 _capabilities;

Uint32 _mask;

AtomicInt _die;

protected:

// handle all your messages. call Base:_handle_async_msg to

private:

// deal with messages you don't handle

static ThreadReturnType PEGASUS_THREAD_CDECL polling_routine(void *);

virtual void _handle_async_msg(AsyncMessage *msg);

AsyncQueue<AsyncOpNode> _incoming;

static Thread* _polling_thread;

static Semaphore _polling_sem;

static AtomicInt _stop_polling;

cimom *_meta_dispatcher;

typedef List<MessageQueueService, Mutex> PollingList;

private:

static PollingList* _polling_list;

static Mutex _polling_list_mutex;

PollingList* _get_polling_list();

static ThreadReturnType PEGASUS_THREAD_CDECL _req_proc(void *);

static void _sendwait_callback(AsyncOpNode *, MessageQueue *, void *);

AtomicInt _incoming_queue_shutdown;

struct timeval _default_op_timeout;

static AtomicInt _xid;

friend class cimom;

void _handle_async_request(AsyncRequest *req);

void _handle_async_reply(AsyncReply *rep);

void _make_response(AsyncRequest *req, Uint32 code);

};

PEGASUS_NAMESPACE_END

#endif /* Pegasus_MessageQueue_Service_h */

Legend:

Removed from v.1.1
changed lines
	Added in v.1.61

No CVS admin address has been configured