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.16.2.1 and 1.39.6.1

version 1.16.2.1, 2002/03/04 11:57:39

version 1.39.6.1, 2003/10/29 22:10:02

Line 1

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

//%2003////////////////////////////////////////////////////////////////////////

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

// IBM Corp.; EMC 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 31

Line 34

#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/AsyncOpNode.h>

#include <Pegasus/Common/Cimom.h>

#include <Pegasus/Common/CimomMessage.h>

#include <Pegasus/Common/Linkage.h>

PEGASUS_NAMESPACE_BEGIN

Line 50

Line 54

typedef MessageQueue Base;

MessageQueueService(const char *name, Uint32 queueID,

MessageQueueService(const char *name,

Uint32 queueID = MessageQueue::getNextQueueId(),

Uint32 capabilities = 0,

Uint32 mask = message_mask::type_cimom |

message_mask::type_service |

message_mask::ha_request |

message_mask::ha_reply |

message_mask::ha_async ) ;

message_mask::ha_async,

int threads=0) ;

virtual ~MessageQueueService(void);

virtual void handle_heartbeat_request(AsyncRequest *req);

virtual void handle_heartbeat_reply(AsyncReply *rep);

virtual void handle_AsyncIoctl(AsyncIoctl *req);

virtual void handle_CimServiceStart(CimServiceStart *req);

virtual void handle_CimServiceStop(CimServiceStop *req);

virtual void handle_CimServicePause(CimServicePause *req);

virtual void handle_CimServiceResume(CimServiceResume *req);

virtual void handle_AsyncOperationStart(AsyncOperationStart *req);

virtual void handle_AsyncOperationResult(AsyncOperationResult *rep);

virtual void handle_AsyncLegacyOperationStart(AsyncLegacyOperationStart *req);

virtual void handle_AsyncLegacyOperationResult(AsyncLegacyOperationResult *rep);

virtual Boolean isAsync(void) { return true; }

virtual Boolean accept_async(AsyncOpNode *op);

virtual Boolean messageOK(const Message *msg) ;

virtual void enqueue(Message *) throw(IPCException);

AsyncReply *SendWait(AsyncRequest *request);

Boolean SendAsync(AsyncOpNode *op,

Uint32 destination,

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

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

void ReplyAsync(AsyncOpNode *op, Uint32 destination);

MessageQueue *callback_q,

void *callback_ptr);

Boolean SendAsync(Message *msg,

Uint32 destination,

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

void *handle,

void *parameter);

Uint32 get_pending_callback_count(void);

Boolean SendForget(Message *msg);

void ForwardRequest(AsyncOpNode *, MessageQueue *);

Boolean ForwardOp(AsyncOpNode *, Uint32 destination);

void ForwardResponse(AsyncOpNode *, MessageQueue *);

void _completeAsyncResponse(AsyncRequest *request,

AsyncReply *reply,

Uint32 state,

Uint32 flag);

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

Boolean update_service(Uint32 capabilities, Uint32 mask);

Boolean deregister_service(void);

virtual void _shutdown_incoming_queue(void);

void find_services(String name,

Uint32 capabilities,

Uint32 mask,

Array<Uint32> *results);

void enumerate_service(Uint32 queue, message_module *result);

Uint32 get_next_xid(void);

AsyncOpNode *get_op(void);

static AsyncOpNode *get_op(void);

void return_op(AsyncOpNode *op);

Boolean operator ==(const MessageQueueService & svce)

{

return operator==((const void *)&svce);

}

Boolean operator ==(const void *svce)

{

if((void *)this == svce)

return true;

return false;

}

static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL polling_routine(void *);

static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL kill_idle_threads(void *);

static int pooled_threads(void)

{

return _thread_pool->running_count() + _thread_pool->dead_count() + _thread_pool->pool_count();

}

static ThreadPool *get_thread_pool(void);

static void force_shutdown(Boolean destroy_flag = false);

Uint32 _mask;

AtomicInt _die;

protected:

virtual Boolean accept_async(AsyncOpNode *op);

virtual Boolean messageOK(const Message *msg) ;

virtual void handleEnqueue(void) = 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 *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);

virtual void handle_heartbeat_request(AsyncRequest *req);

virtual void handle_heartbeat_reply(AsyncReply *rep);

virtual void handle_AsyncIoctl(AsyncIoctl *req);

virtual void handle_CimServiceStart(CimServiceStart *req);

virtual void handle_CimServiceStop(CimServiceStop *req);

virtual void handle_CimServicePause(CimServicePause *req);

virtual void handle_CimServiceResume(CimServiceResume *req);

virtual void handle_AsyncOperationStart(AsyncOperationStart *req);

virtual void handle_AsyncOperationResult(AsyncOperationResult *rep);

virtual void handle_AsyncLegacyOperationStart(AsyncLegacyOperationStart *req);

virtual void handle_AsyncLegacyOperationResult(AsyncLegacyOperationResult *rep);

void _completeAsyncResponse(AsyncRequest *request,

AsyncReply *reply,

Uint32 state,

Uint32 flag);

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

static cimom *_meta_dispatcher;

static AtomicInt _service_count;

static Mutex _meta_dispatcher_mutex;

private:

DQueue<AsyncOpNode> _pending;

AsyncDQueue<AsyncOpNode> _incoming;

DQueue<AsyncOpNode> _callback;

static Thread* _polling_thread;

static Semaphore _polling_sem;

static AtomicInt _stop_polling;

static AtomicInt _check_idle_flag;

static DQueue<MessageQueueService> _polling_list;

static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL _req_proc(void *);

static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL _callback_proc(void *);

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

AtomicInt _incoming_queue_shutdown;

Semaphore _callback_ready;

Thread _req_thread;

Thread _callback_thread;

protected:

static ThreadPool *_thread_pool;

private:

struct timeval _default_op_timeout;

static AtomicInt _xid;

friend class cimom;

friend class CIMServer;

friend class monitor_2;

};

Legend:

Removed from v.1.16.2.1
changed lines
	Added in v.1.39.6.1

No CVS admin address has been configured