pegasus/doc/MessageQueueService.html - diff

Return to MessageQueueService.html CVS log

Up to [Pegasus] / pegasus / doc

Diff for /pegasus/doc/MessageQueueService.html between version 1.1 and 1.4

version 1.1, 2002/02/05 23:14:54

version 1.4, 2002/03/13 17:33:36

Line 1

<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN">

<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN">

<META http-equiv="Content-Type" content="text/html;

charset=ISO-8859-1"><meta http-equiv="Content-Type"

content="text/html;

charset=ISO-8859-1"><title>Pegasus

Meta Dispatcher </title><style type="text/css">

body {

margin: 2em 1em 2em;

font-family: sans-serif;

color: black;

background: white;

background-position: top left;

background-attachment: fixed;

background-repeat: no-repeat;

}

:link { color: #00C; background: transparent }

:visited { color: #609; background: transparent }

:active { color: #C00; background: transparent }

th, td { /* ns 4 */

font-family: sans-serif;

}

h1, h2, h3, h4, h5, h6 { text-align: left }

/* background should be transparent, but WebTV has a bug */

h1, h2, h3 { color: #005A9C; background: white }

h1 { font: 170% sans-serif }

h2 { font: 140% sans-serif }

h3 { font: 120% sans-serif }

h4 { font: bold 100% sans-serif }

h5 { font: italic 100% sans-serif }

h6 { font: small-caps 100% sans-serif }

.hide { display: none }

div.head { margin-bottom: 1em }

div.head h1 { margin-top: 2em; clear: both }

div.head table { margin-left: 2em; margin-top: 2em }

div.head img { color: white; border: none } /* remove border from top image */

p.copyright { font-size: small }

p.copyright small { font-size: small }

pre { margin-left: 2em }

p {

margin-top: 0.6em;

margin-bottom: 0.6em;

}

dt, dd { margin-top: 0; margin-bottom: 0 } /* opera 3.50 */

dt { font-weight: bold }

pre, code { font-family: monospace } /* navigator 4 requires this */

ul.toc {

list-style: disc; /* Mac NS has problem with 'none' */

list-style: none;

}

code { font-family: monospace; }

var {font-family: monospace; font-style: normal; font-weight: bold;}

div.constraint,

div.issue,

div.note,

div.notice { margin-left: 2em; }

li p { margin-top: 0.8em;

margin-bottom: 0.8em; }

li {margin-top: 0.4em; margin-bottom: 0.4em;}

div.exampleInner pre { margin-left: 1em;

margin-top: 0em; margin-bottom: 0em}

div.exampleOuter {border: 4px double gray;

margin: 0em; padding: 0em}

div.exampleInner { background-color: #d5dee3;

border-top-width: 4px;

border-top-style: double;

border-top-color: #d3d3d3;

border-bottom-width: 4px;

border-bottom-style: double;

border-bottom-color: #d3d3d3;

padding: 4px; margin: 0em }

div.exampleWrapper { margin: 4px }

div.exampleHeader { font-weight: bold;

margin: 4px}

table { background-color: #d5dee3;

width: 85% ;

border-style: double;

border-width: 4px;

border-color: #d3d3d3;

}

</style>

<html>

<head>

<title>Pegasus Meta Dispatcher</title>

</head>

<body>

<h1>Pegasus Meta Dispatcher</h1>

The Pegasus Meta Dispatcher is a set of classes that extend the

Line 12

Line 110

asynchronous, and multithreaded. The primary classes consist of the

folowing:

<table>

<tr align="left"><th>Class</th><th>Derived from</th><th>Source file</th></tr>

<tr><td>cimom</td><td>MessageQueue</td><td>Pegasus/Common/Cimom.h</td></tr>

<tr><td><div class="exampleInner">cimom</div></td><td>MessageQueue</td><td>Pegasus/Common/Cimom.h</td></tr>

<tr><td>MessageQueueService</td><td>MessageQueue</td><td>Pegasus/Common/MessageQueueServices.h</td></tr>

<tr><td>CimomMessage</td><td>Message</td><td>Pegasus/Common/CimomMessage.h</td></tr>

<tr><td>AsyncOpNode</td><td>n/a</td><td>Pegasus/Common/AsyncOpNode.h</td></tr>

<tr><td>AsyncDQueue</td><td>unlocked_dq</td><td>Pegasus/Common/DQueue.h</td></tr>

<tr><td>IPC classes</td><td>n/a</td><td>Pegasus/Common/IPC.h</td></tr>

<tr><td>Threading classes</td><td>n/a</td><td>Pegasus/Common/Thread.h</td></tr>

</table>

<a href = "#non-blocking">Jump to the non-blocking message interface.</a>

<h2>Purposes of Meta Dispatcher</h2>

The Meta Dispatcher has two primary goals:

The Meta Dispatcher has three primary goals:

<ol>

<li>Provide for orderly asynchronous message-based communication

among a dynamic set of Pegasus Services.</li>

Line 103

Line 205

The Meta Dispatcher therefore acts as a central message

hub. Services communicate with each other via the Meta

Dispatcher.

<pre>

Service A--Message----1----> (block on semaphore)

Line 119

Line 221

</pre>

</div></div>

The numbered steps above are as follows:

<ol>

<li>Service A creates a new <code>AsyncMessage</code> and

Line 247

Line 350

<pre>

Array<Uint32> services;

Array<Uint32>; services;

while( services.size() == 0 )

{

q_client->find_services(String("test server"), 0, 0, &services);

pegasus_yield();

}

cout << "found server at " << services[0] << endl;

</pre>

Line 525

Line 628

<pre>

_completeAsyncResponse(msg, resp, ASYNC_OPSTATE_COMPLETE, 0);

}

</pre>

</ol>

<h2>Handling CIMMessage and Other Pre-existing Message Classes</h2>

Existing Messages, including all of the <code>CIMMessage</code>

derivitives, are not configured to be asynchronous

request/reply pairs. They are designed to travel through

Pegasus as events that trigger other processing events,

which is the end of their lifetime. This is not an optimal

use model for asynchronous operation because the

originator of the event does not require nor receive any

completion notification. Further, there is not a

one-to-one correspondence of "event messages" to replies.

<hr>

<h3>AsyncLegacyOperationStart Message</h3>

<h2>Class Definitions</h2>

The AsyncLegacyOperationStart message is an envelope that

allows a <code>MessageQueueService</code>-based service to

send, receive, and process pre-existing "legacy"

messages.

The <code>AsyncLegacyOperationStart</code> Message allows

an asynchronous service to create, package, and send a

"legacy" message to another service or, indirectly,

enqueue it to a non-asynchronous message queue. The code

example below shows how this works:

<h3>cimom (Meta Dispatcher)</h3>

<pre>

class PEGASUS_COMMON_LINKAGE cimom : public MessageQueue

{

public :

cimom(void);

virtual ~cimom(void) ;

cout << " sending LEGACY to test server" << endl;

Boolean moduleChange(struct timeval last);

Message *legacy = new Message(0x11100011,

Message::getNextKey());

Uint32 getModuleCount(void);

AsyncLegacyOperationStart *req =

Uint32 getModuleIDs(Uint32 *ids, Uint32 count) throw(IPCException);

new AsyncLegacyOperationStart(q_client->get_next_xid(),

services[0],

legacy,

q_client->getQueueId());

reply = q_client->SendWait(req);

delete req;

delete reply;

AsyncOpNode *get_cached_op(void) throw(IPCException);

</pre>

void cache_op(AsyncOpNode *op) throw(IPCException);

The code sample above shows a <code>Message</code> object

being embedded inside an

<code>AsyncLegacyOperationStart</code> message and sent

using the <code>SendWait</code>API.

void set_default_op_timeout(const struct timeval *buffer);

<h3>Default Handler for Legacy Messages</h3>

void get_default_op_timeout(struct timeval *timeout) const ;

The <code>MessageQueueService</code> class has a default

handler for legacy messages that extracts the

<code>Message</code> out of its asynchronous "envelope"

and dispatches it using the pre-existing synchronous

interface, as shown below.

virtual void handleEnqueue();

<pre>

void register_module(RegisterCimService *msg);

void deregister_module(Uint32 quid);

void update_module(UpdateCimService *msg );

void ioctl(AsyncIoctl *msg );

void find_service_q(FindServiceQueue *msg );

void MessageQueueService::handle_AsyncLegacyOperationStart(

void enumerate_service(EnumerateService *msg );

AsyncLegacyOperationStart *req)

Boolean route_async(AsyncOpNode *operation);

{

void _shutdown_routed_queue(void);

// remove the legacy message from the request and enqueue it to its destination

Uint32 result = async_results::CIM_NAK;

Message *legacy = req->act;

if ( legacy != 0 )

{

MessageQueue* queue = MessageQueue::lookup(req->legacy_destination);

if( queue != 0 )

{

// Enqueue the response:

queue->enqueue(legacy);

result = async_results::OK;

}

_make_response(req, result);

}

protected:

</pre>

Uint32 get_module_q(const String & name);

void _make_response(AsyncRequest *req, Uint32 code);

void _completeAsyncResponse(AsyncRequest *request,

AsyncReply *reply,

Uint32 state,

Uint32 flag);

private:

struct timeval _default_op_timeout;

struct timeval _last_module_change;

DQueue<message_module> _modules;

DQueue<AsyncOpNode> _recycle;

The default handler shown above extracts the legacy

message and attempts to <code>enqueue</code> that message

syncrhonously using the pre-existing interface.

AsyncDQueue<AsyncOpNode> _routed_ops;

<h3>Example of Custom Handler for Legacy Messages</h3>

DQueue<AsyncOpNode> _internal_ops;

By implementing the virtual

<code>_handle_async_request</code> method,

a service can choose to implement its own handler for

Legacy messages, as the code below shows:

static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL _routing_proc(void *);

<ol>

Thread _routing_thread;

<li>Implement the virtual <code>_handle_async_request</code> method.</li>

<pre>

static Uint32 get_xid(void);

void MessageQueueServer::_handle_async_request(AsyncRequest *req)

void _handle_cimom_op(AsyncOpNode *op, Thread *thread, MessageQueue *queue);

{

Uint32 _ioctl(Uint32, Uint32, void *);

if (req->getType() == 0x04100000 )

{

req->op->processing();

handle_test_request(req);

}

else if ( req->getType() == async_messages::CIMSERVICE_STOP )

{

req->op->processing();

handle_CimServiceStop(static_cast<CimServiceStop *>(req));

}

</pre>

<li>Implement a dispatcher for <code>ASYNC_LEGACY_OP_START</code></li>

<pre>

else if ( req->getType() == async_messages::ASYNC_LEGACY_OP_START )

{

req->op->processing();

handle_LegacyOpStart(static_cast<AsyncLegacyOperationStart *>(req));

}

else

Base::_handle_async_request(req);

}

</pre>

<li>Implement a dispatcher for <code>ASYNC_LEGACY_OP_START</code></li>

<pre>

void MessageQueueServer::handle_LegacyOpStart(AsyncLegacyOperationStart *req)

{

AtomicInt _die;

Message *legacy = req->act;

AtomicInt _routed_queue_shutdown;

cout << " ### handling legacy messages " << endl;

static AtomicInt _xid;

// CIMOperationRequestDispatcher *_cim_dispatcher;

AsyncReply *resp =

// CIMOperationResponseEncoder *_cim_encoder;

new AsyncReply(async_messages::REPLY,

// CIMOperationRequestDecoder *_cim_decoder;

req->getKey(),

// CIMRepository *_repository;

req->getRouting(),

req->op,

async_results::OK,

req->resp,

req->block);

_completeAsyncResponse(req, resp, ASYNC_OPSTATE_COMPLETE, 0 );

};

if (legacy != 0 )

cout << " legacy msg type: " << legacy->getType() << endl;

}

</pre>

<h3>MessageQueueService</h3>

</ol>

<pre>

<hr>

class message_module;

class PEGASUS_COMMON_LINKAGE MessageQueueService : public MessageQueue

<a name="non-blocking"><h2>Sending Messages without Blocking (Async with Callback)</h2></a>

{

public:

typedef MessageQueue Base;

Whenever there is a possibility that the processing of

one message may generate a nested message (message

generated within the handler of a message) it is

necessary to send messages without blocking, and to

receive responses via callback routines. The diagram

below shows the (more complicated) flow of

non-blocking messages.

<pre>

Service A--Message----1---->

. <-----------(return)-+----->-(loop)--->-+

. | Meta Dispatcher |

. +----<-----<---<---+

. Message---2-->Service B

. |

. <--Response--3------+

. |

. +--<--<-----<-----+--(return)---->

. | Meta Dispatcher |

Service A <--Callback--4---+--->-(loop)-->---+

| ^

+-------+

</pre>

</div></div>

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

<h3>Test Program</h3>

virtual ~MessageQueueService(void);

There is a test program that sends and receives

non-blocking messages in

<code>$(PEGASUS_ROOT)/src/Pegasus/Common/tests/async_callback/</code>

virtual void handle_heartbeat_request(AsyncRequest *req);

<h3>SendAsync method</h3>

virtual void handle_heartbeat_reply(AsyncReply *rep);

virtual void handle_AsyncIoctl(AsyncIoctl *req);

virtual void handle_CimServiceStart(CimServiceStart *req);

The <code>MessageQueueService</code> class sends

virtual void handle_CimServiceStop(CimServiceStop *req);

non-blocking messages using the <code>SendAsync</code>

virtual void handle_CimServicePause(CimServicePause *req);

method from <code>MessageQueueService.h</code>.

virtual void handle_CimServiceResume(CimServiceResume *req);

virtual void handle_AsyncOperationStart(AsyncOperationStart *req);

virtual void handle_AsyncOperationResult(AsyncOperationResult *req);

virtual Boolean accept_async(AsyncOpNode *op);

virtual Boolean messageOK(const Message *msg) ;

AsyncReply *SendWait(AsyncRequest *request);

<pre>

void _completeAsyncResponse(AsyncRequest *request,

Boolean SendAsync(AsyncOpNode *op,

AsyncReply *reply,

Uint32 destination,

Uint32 state,

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

Uint32 flag);

MessageQueue *callback_q,

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

void <font

Boolean update_service(Uint32 capabilities, Uint32 mask);

color=#990000>*callback_ptr);

Boolean deregister_service(void);

</pre>

virtual void _shutdown_incoming_queue(void);

void find_services(String name,

Uint32 capabilities,

<var>AsyncOpNode *op (In)</var> is the shared data

Uint32 mask,

structure that controls the message flow and

Array<Uint32> *results);

consolidates the response data. This data structure is

void enumerate_service(Uint32 queue, message_module *result);

also passed to the callback function as the first

Uint32 get_next_xid(void);

parameter. The caller must

AsyncOpNode *get_op(void);

allocate and free this data structure.

void return_op(AsyncOpNode *op);

Uint32 _capabilities;

<var>Uint32 destination (In)</var> is the queue ID of

Uint32 _mask;

the service which will receive the asynchronous

AtomicInt _die;

request message.</var>

protected:

<var>void (*callback)(AsyncOpNode *, MessageQueue *,

void *) (In)</var> is the static class method that

will be called when the request/response pair is complete.

The callback is always passed <var>op</var> as the

first parameter. The second parameter is the

<var>MessageQueue *</var> object instance that is

executing the callback function. Because callback

functions must be static class methods, the callback

can use the <var>MessageQueue *</var> parameter as a

pseudo <var>this</var> pointer.

<var>MessageQueue *callback_q (In)</var> is the

queue pointer that will be passed to the callback

routine as the second parameter. It also controls

which queue instance receives the callback.

<var>void *callback_ptr (In)</var> is a pointer that will

be passed to the callback function.

virtual void _handle_incoming_operation(AsyncOpNode *operation, Thread *thread, MessageQueue *queue);

</div>

virtual void _handle_async_request(AsyncRequest *req);

</div></div>

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

cimom *_meta_dispatcher;

private:

void handleEnqueue();

DQueue<AsyncOpNode> _pending;

AsyncDQueue<AsyncOpNode> _incoming;

static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL _req_proc(void *);

<h3>How the Meta Dispatcher Processes Non-Blocking Messages</h3>

AtomicInt _incoming_queue_shutdown;

The focus of processing non-blocking messages is to

remove the possibility of deadlock. Therefore, all

message processing is performed by background

threads. All access by services to shared data

structures is discrete in the sense that one service

never calls into another service. For example,

callback routines are executed by the background

thread of the receiving service. (As opposed to the

sending service making a callback into the

receiving service.)

Thread _req_thread;

<ol>

<li>The requesting service creates a request, an

<code>AsyncOpNode</code> and calls

<code>MessageQueueService::SendAsync(...)</code>.</li>

<li><code>SendAsync</code> marks the

<code>AsyncOpNode</code> as

<code>ASYNC_OPFLAGS_CALLBACK</code> with the

<code>ASYNC_OPSTATE_COMPLETE</code> bit

clear.</li> <li>The Meta Dispatcher routes

the <code>AsyncOpNode</code> to the responding

service.</li> <li>The responding service's

background thread pulls the message off its

internal queue and processes the request. This

will frequently entail creating and inserting a

response message into the <code>AsyncOpNode</code>

by calling

<code>AsyncOpNode::put_response(AsyncMessage

*)</code>.</li> <li>The responding service

completes the response by calling

<code>MessageQueueService::_complete_op_node(AsyncOpNode

*, Uint32, Uint32, Uint32)</code>.</li>

<li><code>_complete_op_node</code> passes the

operation back to the Meta Dispatcher, which

routes it back to the requesting

service.</li> <li>The requesting service's

background thread pulls the

<code>AsyncOpNode</code> off its internal queue

and calls

<code>AsyncOpNode::_async_callback</code>, which

is a function pointer that holds the address of the

requesting service's static callback method..</li>

<li>The requesting service deallocates the op

node, the request message, and the response message.</li>

</ol>

struct timeval _default_op_timeout;

<h3>Virtual Methods for Non-Blocking Messages</h3>

To use non-blocking messages, a service should

implement the following virtual method from

<code>MessageQueueService</code>, plus a static class

callback function.

<pre>

virtual void _handle_async_request(AsyncRequest *req);

</pre>

<var>AsyncRequest *req (In)</var> is the incoming

request. <code>req->op</code> is the

<code>AsyncOpNode</code> that is controlling this

operation.

This method is called by the service's background

thread whenever an <code>AsyncRequest</code>

message is sent to the service. This includes

both blocking and non-blocking request messages.

Within <code>_handle_async_request</code> you

can determine whether the message is blocking or

non-blocking by using the following code

segment:

static AtomicInt _xid;

if(req->op->read_flags() & ASYNC_OPFLAGS_CALLBACK )

{

};

// let everyone know you are working on the request

req->op->processing();

// this request is part of a callback (non-blocking) operation

AsyncResponse *response = new AsyncResponse(...);

</pre>

// initialize the response message however is appropriate

<h3>Asynchronous Messages</h3>

// put the response to the op node

req->op->put_response(response);

<pre>

// complete the operation

_complete_op_node(req->op, 0, 0, async_results::OK);

extern const Uint32 CIMOM_Q_ID;

return;

}

</code></pre>

</div>

class AsyncOpNode;

class PEGASUS_COMMON_LINKAGE async_results

{

public:

static const Uint32 OK;

static const Uint32 PARAMETER_ERROR;

static const Uint32 MODULE_ALREADY_REGISTERED;

static const Uint32 MODULE_NOT_FOUND;

static const Uint32 INTERNAL_ERROR;

static const Uint32 ASYNC_STARTED;

static const Uint32 ASYNC_PROCESSING;

static const Uint32 ASYNC_COMPLETE;

static const Uint32 ASYNC_CANCELLED;

static const Uint32 ASYNC_PAUSED;

static const Uint32 ASYNC_RESUMED;

static const Uint32 CIM_SERVICE_STARTED;

static const Uint32 CIM_SERVICE_STOPPED;

static const Uint32 CIM_SERVICE_PAUSED;

static const Uint32 CIM_SERVICE_RESUMED;

static const Uint32 CIM_NAK;

static const Uint32 ASYNC_PHASE_COMPLETE;

static const Uint32 ASYNC_CHILD_COMPLETE;

static const Uint32 ASYNC_PHASE_STARTED;

static const Uint32 ASYNC_CHILD_STARTED;

static const Uint32 CIM_PAUSED;

static const Uint32 CIM_STOPPED;

};

<pre>

static void async_callback_function(AsyncOpNode *op, MessageQueue *, void *);

class PEGASUS_COMMON_LINKAGE async_messages

</pre>

{

public:

<var>AsyncOpNode *op (In)</var> is the

static const Uint32 HEARTBEAT;

incoming shared object that is controlling this

static const Uint32 REPLY;

completing operation.

static const Uint32 REGISTER_CIM_SERVICE;

static const Uint32 DEREGISTER_CIM_SERVICE;

<var>MessageQueue *q (In)</var> is a pointer to

static const Uint32 UPDATE_CIM_SERVICE;

the class that has received the complete

static const Uint32 IOCTL;

<var>op</var>. This parameter is meant to be used

static const Uint32 CIMSERVICE_START;

as a <code>this</code> pointer because the

static const Uint32 CIMSERVICE_STOP;

callback is a static class method.

static const Uint32 CIMSERVICE_PAUSE;

static const Uint32 CIMSERVICE_RESUME;

<var>void *parm (In)</var> is a pointer that the

static const Uint32 ASYNC_OP_START;

class passed to <code>SendAsync</code>. It can be

static const Uint32 ASYNC_OP_RESULT;

used as a convenience for the class.

static const Uint32 ASYNC_LEGACY_OP_START;

static const Uint32 ASYNC_LEGACY_OP_RESULT;

static const Uint32 FIND_SERVICE_Q;

static const Uint32 FIND_SERVICE_Q_RESULT;

static const Uint32 ENUMERATE_SERVICE;

static const Uint32 ENUMERATE_SERVICE_RESULT;

};

This method is only called when a

non-blocking operation is complete, meaning that

the responding service has added a response

message to the <var>op</var> and set the

<code>ASYNC_OPSTATE_COMPLETE</code> bit.

<pre>

class PEGASUS_COMMON_LINKAGE AsyncMessage : public Message

void my_class::async_callback_function(AsyncOpNode *op,

MessageQueue *q,

void *parm)

{

public:

AsyncMessage(Uint32 type,

Uint32 key,

Uint32 routing,

Uint32 mask,

AsyncOpNode *operation);

virtual ~AsyncMessage(void)

my_class *myself = static_cast<my_class *>(q);

{

AsyncRequest *request = op->get_request();

AsyncResponse *response = op->get_response();

// process response

delete request;

delete response;

myself->put_op(op);

return;

}

Boolean operator ==(void *key);

</div>

Boolean operator ==(const AsyncMessage& msg);

</div>

AsyncOpNode *op;

<h3>Strategies for Handling Non-Blocking Responses</h3>

Thread *_myself;

MessageQueue *_service;

All of the legacy message handling code in Pegasus is

};

syncrhonous, meaning that a service can send a request

and receive the response in two adjacent lines of

code. Non-blocking messages are different because the

timing and existence of a response message is

undetermined.

There are couple of possible strategies for handing

non-blocking response messages.

<ol>

<li>Seperate the request creation and sending code

from the response handling code into distinct methods.</li>

<li>Handle request creation and sending and response

handling in the same method using different code blocks.

</ol>

<h4>Separate Request and Response Handling Methods</h4>

<ol>

<li>Write the request generation method.

<pre>

inline Boolean AsyncMessage::operator ==(void *key)

void my_class::generate_request(Uint32 destination)

{

if( key == reinterpret_cast<void *>(this))

AsyncOpNode *op = get_op();

return true;

my_request *req = new my_request(...);

return false;

op->put_request(req);

SendAsync(op, destination, my_callback, this, (void *)0);

return;

}

</pre>

</li>

<li> Write the request handling method.

<pre>

inline Boolean AsyncMessage::operator ==(const AsyncMessage& msg)

void my_class::handle_response(AsyncOpNode *op)

{

my_response *res = op->get_response();

if(res != 0 )

{

return this->operator==(reinterpret_cast<void *>(const_cast<AsyncMessage *>(&msg)));

// handle response

delete response;

}

put_op(op);

return;

}

</pre>

</li>

<li> Have your callback method call your response

method.

<pre>

void my_class:my_callback(AsyncOpNode *op, MessageQueue *queue, void *parm)

class PEGASUS_COMMON_LINKAGE AsyncRequest : public AsyncMessage

{

public:

my_class *myself = static_cast<queue>

AsyncRequest(Uint32 type,

Uint32 key,

Uint32 routing,

Uint32 mask,

AsyncOpNode *operation,

Uint32 destination,

Uint32 response,

Boolean blocking);

myself->handle_response(op);

return;

}

</pre>

</li>

</ol>

</div>

virtual ~AsyncRequest(void)

{

}

<h4>Singe Request and Response Handling Method</h4>

This strategy requires two separate code blocks within

the request/response method, and conditional execution

depending on the status of the operation.

I think the advantage of this strategy is that it

matches more closely the current

<code>handleEnqueue(Message *msg)</code> code

convention that is in Pegasus.

Uint32 dest;

Uint32 resp;

Boolean block;

<ol>

};

<li>Write the request generation block.

<pre>

class PEGASUS_COMMON_LINKAGE AsyncReply : public AsyncMessage

void my_class::handle_operation(AsyncOpNode *op)

{

public:

if(op == NULL)

AsyncReply(Uint32 type,

{

Uint32 key,

AsyncOpNode *op = get_op();

Uint32 routing,

my_request *req = new my_request(...);

Uint32 mask,

op->put_request(req);

AsyncOpNode *operation,

Uint32 result_code,

Uint32 destination,

Boolean blocking);

SendAsync(op, destination, my_callback, this, (void *)0);

virtual ~AsyncReply(void)

}

else

{

if(op != 0 )

delete op;

}

return;

}

</pre>

</li>

<li> Write the request handling block.

<pre>

Uint32 result;

void my_class::handle_operation(AsyncOpNode *op)

Uint32 dest;

{

Boolean block;

if(op == NULL)

};

{

AsyncOpNode *op = get_op();

my_request *req = new my_request(...);

op->put_request(req);

SendAsync(op, destination, my_callback, this, (void *)0);

class PEGASUS_COMMON_LINKAGE RegisterCimService : public AsyncRequest

}

else

{

public:

my_response *res = op->get_response();

RegisterCimService(Uint32 routing,

if(res != 0 )

AsyncOpNode *operation,

Boolean blocking,

String service_name,

Uint32 service_capabilities,

Uint32 service_mask,

Uint32 service_queue);

virtual ~RegisterCimService(void)

{

// handle response

delete response;

}

put_op(op);

}

return;

}

String name;

</pre>

Uint32 capabilities;

</li>

Uint32 mask;

<li> Have your callback method call your handler

Uint32 queue;

method.

};

<pre>

class PEGASUS_COMMON_LINKAGE DeRegisterCimService : public AsyncRequest

void my_class:my_callback(AsyncOpNode *op, MessageQueue *queue, void *parm)

{

public:

my_class *myself = static_cast<queue>

DeRegisterCimService(Uint32 routing,

AsyncOpNode *operation,

myself->handle_operation(op);

Boolean blocking,

return;

Uint32 service_queue);

}

</pre>

</li>

</ol>

</div>

virtual ~DeRegisterCimService(void)

<h2>Class Definitions</h2>

{

}

<h3>cimom (Meta Dispatcher)</h3>

Uint32 queue;

<pre>

} ;

<tt>

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

//

// The Open Group, Tivoli Systems

//

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

//

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

//

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

//

// Modified By:

//

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

#ifndef CIMOM_include

#define CIMOM_include

#include <Pegasus/Common/Config.h>

#include <Pegasus/Common/Exception.h>

#include <Pegasus/Common/MessageQueue.h>

#include <Pegasus/Common/DQueue.h>

#include <Pegasus/Common/Thread.h>

#include <Pegasus/Common/Array.h>

#include <Pegasus/Common/AsyncOpNode.h>

#include <Pegasus/Common/CimomMessage.h>

#include <Pegasus/Common/MessageQueueService.h>

PEGASUS_NAMESPACE_BEGIN

extern const Uint32 CIMOM_Q_ID;

class PEGASUS_COMMON_LINKAGE module_capabilities

{

public:

static Uint32 async;

static Uint32 remote;

static Uint32 trusted;

static Uint32 paused;

static Uint32 stopped;

} ;

class PEGASUS_COMMON_LINKAGE cimom;

class PEGASUS_COMMON_LINKAGE message_module

{

public:

message_module(void)

: _name(), _capabilities(0),

_mask(0), _q_id(0) { }

message_module(const String & name,

Uint32 capabilities,

Uint32 mask,

Uint32 queue)

: _name(name), _capabilities(capabilities),

_mask(mask), _q_id(queue) { }

Boolean operator == (const message_module *mm) const;

Boolean operator == (const String & name ) const ;

Boolean operator == (const message_module & mm ) const ;

Boolean operator == (const void *) const;

Boolean operator == (Uint32) const;

const String & get_name(void) const ;

Uint32 get_capabilities(void) const ;

Uint32 get_mask(void) const ;

Uint32 get_queue(void) const ;

void put_name(String & name);

void put_capabilities(Uint32 capabilities);

void put_mask(Uint32 mask);

void put_queue(Uint32 queue) ;

private:

String _name;

Uint32 _capabilities;

Uint32 _mask;

struct timeval _heartbeat;

Uint32 _q_id;

friend class cimom;

};

class MessageQueueService;

class PEGASUS_COMMON_LINKAGE cimom : public MessageQueue

{

public :

cimom(void);

virtual ~cimom(void) ;

Boolean moduleChange(struct timeval last);

Uint32 getModuleCount(void);

Uint32 getModuleIDs(Uint32 *ids, Uint32 count) throw(IPCException);

AsyncOpNode *get_cached_op(void) throw(IPCException);

void cache_op(AsyncOpNode *op) throw(IPCException);

void set_default_op_timeout(const struct timeval *buffer);

void get_default_op_timeout(struct timeval *timeout) const ;

void register_module(RegisterCimService *msg);

void deregister_module(Uint32 quid);

void update_module(UpdateCimService *msg );

void ioctl(AsyncIoctl *msg );

void find_service_q(FindServiceQueue *msg );

void enumerate_service(EnumerateService *msg );

Boolean route_async(AsyncOpNode *operation);

void _shutdown_routed_queue(void);

protected:

Uint32 get_module_q(const String & name);

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

static void _completeAsyncResponse(AsyncRequest *request,

AsyncReply *reply,

Uint32 state,

Uint32 flag);

static void _complete_op_node(AsyncOpNode *op, Uint32 state, Uint32 flag, Uint32 code);

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

private:

struct timeval _default_op_timeout;

struct timeval _last_module_change;

DQueue<message_module> _modules;

DQueue<AsyncOpNode> _recycle;

AsyncDQueue<AsyncOpNode> _routed_ops;

DQueue<AsyncOpNode> _internal_ops;

static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL _routing_proc(void *);

Thread _routing_thread;

static Uint32 get_xid(void);

void _handle_cimom_op(AsyncOpNode *op, Thread *thread, MessageQueue *queue);

Uint32 _ioctl(Uint32, Uint32, void *);

AtomicInt _die;

AtomicInt _routed_queue_shutdown;

static AtomicInt _xid;

static cimom *_global_this;

class PEGASUS_COMMON_LINKAGE UpdateCimService : public AsyncRequest

friend class MessageQueueService;

{

public:

UpdateCimService(Uint32 routing,

AsyncOpNode *operation,

Boolean blocking,

Uint32 service_queue,

Uint32 service_capabilities,

Uint32 service_mask);

virtual ~UpdateCimService(void)

};

{

}

PEGASUS_NAMESPACE_END

Uint32 queue;

#endif // CIMOM_include

Uint32 capabilities;

</tt>

Uint32 mask;

</pre>

};

<h3>MessageQueueService</h3>

class PEGASUS_COMMON_LINKAGE AsyncIoctl : public AsyncRequest

<pre>

{

<tt>

public:

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

AsyncIoctl(Uint32 routing,

//

AsyncOpNode *operation,

Uint32 destination,

//

Uint32 response,

// Permission is hereby granted, free of charge, to any person obtaining a copy

Boolean blocking,

// of this software and associated documentation files (the "Software"), to

Uint32 code,

// deal in the Software without restriction, including without limitation the

Uint32 int_param,

// rights to use, copy, modify, merge, publish, distribute, sublicense, and/or

void *p_param);