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.2 and 1.5

version 1.2, 2002/02/07 16:46:42

version 1.5, 2005/05/12 13:40:18

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>

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

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

Meta Dispatcher|

Line 119

Line 220

</pre>

</div></div>

The numbered steps above are as follows:

<ol>

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

Line 168

Line 270

<ol>

<li>Define the Service Class</li>

<pre>

<pre>// Define our service class

// Define our service class

class MessageQueueClient : public MessageQueueService

{

Line 214

Line 315

</pre>

<li>Construct the Service</li>

<pre>

<pre>// Create our Service

// Create our Service

MessageQueueClient *q_client =

new MessageQueueClient("test client");

</pre>

<li>Register the Service</li>

<pre>

<pre>// Register our service with the Meta Dispatcher

// Register our service with the Meta Dispatcher

q_client->register_service("test client",

q_client->_client_capabilities,

q_client->_client_mask);

Line 246

Line 345

program:

<pre>

Array<Uint32>; services;

while( services.size() == 0 )

Line 277

Line 375

<ol>

<li>Define the Request and Response Message Pair by Inheriting from AsyncMessage.</li>

<pre>

class test_request : public AsyncRequest

{

Line 352

Line 449

process the reply.

<pre>

void MessageQueueClient::send_test_request(char *greeting, Uint32 qid)

{

</pre>

<li>Construct the Request</li>

<pre>

<pre> test_request *req =

test_request *req =

new test_request(Base::get_next_xid(),

qid, // destination queue ID

Line 371

Line 466

<li>Send the message using <code>MessageQueueService::SendWait</code></li>

<pre>

<pre> AsyncMessage *response = SendWait(req);

AsyncMessage *response = SendWait(req);

</pre>

<li>Process the Response.

<pre>

<pre> if( response != 0 )

if( response != 0 )

{

msg_count++;

delete response;

Line 414

Line 507

before inserting the request on the Service's queue.

<pre>

Boolean MessageQueueServer::messageOK(const Message *msg)

{

if(msg->getMask() & message_mask::ha_async)

Line 447

Line 539

the message.

<pre>

void MessageQueueServer::_handle_incoming_operation(AsyncOpNode *op)

{

if ( operation != 0 )

Line 473

Line 564

this method. If the Service does not handle the Request it

must pass the Request to the Base class by calling <code>Base::_handle_async_request(req)</code>

<pre>

<pre>void MessageQueueServer::_handle_async_request(AsyncRequest *req)

void MessageQueueServer::_handle_async_request(AsyncRequest *req)

{

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

{

Line 500

Line 590

program.

<pre>

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

{

Line 508

Line 597

<ol>

<li>Construct the Reply</li>

<pre>

test_response *resp =

new test_response(msg->getKey(),

msg->getRouting(),

Line 522

Line 610

<li>Complete the Reply by calling the following

helper routine in the Base class</li>

<pre>

<pre> _completeAsyncResponse(msg, resp, ASYNC_OPSTATE_COMPLETE, 0);

_completeAsyncResponse(msg, resp, ASYNC_OPSTATE_COMPLETE, 0);

}

</pre>

Line 560

Line 647

<pre>

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

Message *legacy = new Message(0x11100011,

Line 594

Line 680

<pre>

void MessageQueueService::handle_AsyncLegacyOperationStart(

AsyncLegacyOperationStart *req)

{

Line 635

Line 720

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

<pre>

void MessageQueueServer::_handle_async_request(AsyncRequest *req)

{

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

Line 650

Line 734

}

</pre>

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

<pre>

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

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

{

req->op->processing();

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

Line 664

Line 747

</pre>

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

<pre>

void MessageQueueServer::handle_LegacyOpStart(AsyncLegacyOperationStart *req)

{

Line 694

Line 776

</ol>

<hr>

<h2>Class Definitions</h2>

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

<pre>

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 ;

virtual void handleEnqueue();

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:

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

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;

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;

// CIMOperationRequestDispatcher *_cim_dispatcher;

// CIMOperationResponseEncoder *_cim_encoder;

// CIMOperationRequestDecoder *_cim_decoder;

// CIMRepository *_repository;

};

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>

<h3>Test Program</h3>

<h3>MessageQueueService</h3>

There is a test program that sends and receives

<pre>

non-blocking messages in

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

class message_module;

class PEGASUS_COMMON_LINKAGE MessageQueueService : public MessageQueue

{

public:

typedef MessageQueue Base;

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

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 *req);

virtual Boolean accept_async(AsyncOpNode *op);

virtual Boolean messageOK(const Message *msg) ;

AsyncReply *SendWait(AsyncRequest *request);

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);

void return_op(AsyncOpNode *op);

Uint32 _capabilities;

Uint32 _mask;

AtomicInt _die;

protected:

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

virtual void _handle_async_request(AsyncRequest *req);

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 *);

AtomicInt _incoming_queue_shutdown;

Thread _req_thread;

struct timeval _default_op_timeout;

static AtomicInt _xid;

<h3>SendAsync method</h3>

};

The <code>MessageQueueService</code> class sends

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

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

<pre>

Boolean SendAsync(AsyncOpNode *op,

Uint32 destination,

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

MessageQueue *callback_q,

void <font

color=#990000>*callback_ptr);

</pre>

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

structure that controls the message flow and

consolidates the response data. This data structure is

also passed to the callback function as the first

parameter. The caller must

allocate and free this data structure.

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

the service which will receive the asynchronous

request message.</var>

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

<h3>Asynchronous Messages</h3>

</div>

</div></div>

<pre>

extern const Uint32 CIMOM_Q_ID;

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;

};

class PEGASUS_COMMON_LINKAGE async_messages

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

{

public:

The focus of processing non-blocking messages is to

static const Uint32 HEARTBEAT;

remove the possibility of deadlock. Therefore, all

static const Uint32 REPLY;

message processing is performed by background

static const Uint32 REGISTER_CIM_SERVICE;

threads. All access by services to shared data

static const Uint32 DEREGISTER_CIM_SERVICE;

structures is discrete in the sense that one service

static const Uint32 UPDATE_CIM_SERVICE;

never calls into another service. For example,

static const Uint32 IOCTL;

callback routines are executed by the background

static const Uint32 CIMSERVICE_START;

thread of the receiving service. (As opposed to the

static const Uint32 CIMSERVICE_STOP;

sending service making a callback into the

static const Uint32 CIMSERVICE_PAUSE;

receiving service.)

static const Uint32 CIMSERVICE_RESUME;

static const Uint32 ASYNC_OP_START;

static const Uint32 ASYNC_OP_RESULT;

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;

};

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

class PEGASUS_COMMON_LINKAGE AsyncMessage : public Message

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

{

public:

To use non-blocking messages, a service should

AsyncMessage(Uint32 type,

implement the following virtual method from

Uint32 key,

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

Uint32 routing,

callback function.

Uint32 mask,

AsyncOpNode *operation);

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

virtual ~AsyncMessage(void)

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

{

}

// let everyone know you are working on the request

req->op->processing();

Boolean operator ==(void *key);

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

Boolean operator ==(const AsyncMessage& msg);

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

AsyncOpNode *op;

// initialize the response message however is appropriate

Thread *_myself;

MessageQueue *_service;

};

// put the response to the op node

req->op->put_response(response);

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

// complete the operation

{

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

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

return true;

return false;

}

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

return;

{

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

}

</code></pre>

</div>

class PEGASUS_COMMON_LINKAGE AsyncRequest : public AsyncMessage

{

public:

AsyncRequest(Uint32 type,

Uint32 key,

Uint32 routing,

Uint32 mask,

AsyncOpNode *operation,

Uint32 destination,

Uint32 response,

Boolean blocking);

virtual ~AsyncRequest(void)

{

}

<pre>

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

Uint32 dest;

</pre>

Uint32 resp;

Boolean block;

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

};

incoming shared object that is controlling this

completing operation.

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

the class that has received the complete

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

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

callback is a static class method.

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

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

used as a convenience for the class.

class PEGASUS_COMMON_LINKAGE AsyncReply : public AsyncMessage

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>

void my_class::async_callback_function(AsyncOpNode *op,

MessageQueue *q,

void *parm)

{

public:

AsyncReply(Uint32 type,

Uint32 key,

Uint32 routing,

Uint32 mask,

AsyncOpNode *operation,

Uint32 result_code,

Uint32 destination,

Boolean blocking);

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

AsyncRequest *request = op->get_request();

AsyncResponse *response = op->get_response();

virtual ~AsyncReply(void)

// process response

{

if(op != 0 )

delete op;

delete request;

delete response;

myself->put_op(op);

return;

}

Uint32 result;

</div>

Uint32 dest;

</div>

Boolean block;

</div>

};

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

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>

class PEGASUS_COMMON_LINKAGE RegisterCimService : public AsyncRequest

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

{

public:

RegisterCimService(Uint32 routing,

<ol>

AsyncOpNode *operation,

<li>Write the request generation method.

Boolean blocking,

<pre>

String service_name,

void my_class::generate_request(Uint32 destination)

Uint32 service_capabilities,

Uint32 service_mask,

Uint32 service_queue);

virtual ~RegisterCimService(void)

{

AsyncOpNode *op = get_op();

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

op->put_request(req);

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

return;

}

</pre>

String name;

</li>

Uint32 capabilities;

<li> Write the request handling method.

Uint32 mask;

<pre>

Uint32 queue;

void my_class::handle_response(AsyncOpNode *op)

};

class PEGASUS_COMMON_LINKAGE DeRegisterCimService : public AsyncRequest

{

public:

my_response *res = op->get_response();

DeRegisterCimService(Uint32 routing,

if(res != 0 )

AsyncOpNode *operation,

Boolean blocking,

Uint32 service_queue);

virtual ~DeRegisterCimService(void)

{

// handle response

delete response;

}

put_op(op);

Uint32 queue;

return;

} ;

class PEGASUS_COMMON_LINKAGE UpdateCimService : public AsyncRequest

{

public:

UpdateCimService(Uint32 routing,

AsyncOpNode *operation,

Boolean blocking,

Uint32 service_queue,

Uint32 service_capabilities,

Uint32 service_mask);

virtual ~UpdateCimService(void)

{

}

</pre>

Uint32 queue;

</li>

Uint32 capabilities;

<li> Have your callback method call your response

Uint32 mask;

method.

};

<pre>

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

class PEGASUS_COMMON_LINKAGE AsyncIoctl : public AsyncRequest

{

public:

AsyncIoctl(Uint32 routing,

AsyncOpNode *operation,

Uint32 destination,

Uint32 response,

Boolean blocking,

Uint32 code,

Uint32 int_param,

void *p_param);

virtual ~AsyncIoctl(void)

{

my_class *myself = static_cast<queue>

myself->handle_response(op);

return;

}

</pre>

</li>

</ol>

</div>

enum

{

IO_CLOSE,

IO_OPEN,

IO_SOURCE_QUENCH,

IO_SERVICE_DEFINED

};

Uint32 ctl;

Uint32 intp;

void *voidp;

};

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

class PEGASUS_COMMON_LINKAGE CimServiceStart : public AsyncRequest

This strategy requires two separate code blocks within

{

the request/response method, and conditional execution

public:

depending on the status of the operation.

CimServiceStart(Uint32 routing,

AsyncOpNode *operation,

I think the advantage of this strategy is that it

Uint32 destination,

matches more closely the current

Uint32 response,

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

Boolean blocking);

convention that is in Pegasus.

virtual ~CimServiceStart(void)

<ol>

<li>Write the request generation block.

<pre>

void my_class::handle_operation(AsyncOpNode *op)

{

if(op == NULL)

}

};

class PEGASUS_COMMON_LINKAGE CimServiceStop : public AsyncRequest

{

public:

AsyncOpNode *op = get_op();

CimServiceStop(Uint32 routing,

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

AsyncOpNode *operation,

op->put_request(req);

Uint32 destination,

Uint32 response,

Boolean blocking);

virtual ~CimServiceStop(void)

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

{

}

};

else

class PEGASUS_COMMON_LINKAGE CimServicePause : public AsyncRequest

{

public:

CimServicePause(Uint32 routing,

AsyncOpNode *operation,

Uint32 destination,

Uint32 response,

Boolean blocking);

virtual ~CimServicePause(void)

{

}

};

return;

class PEGASUS_COMMON_LINKAGE CimServiceResume : public AsyncRequest

{

public:

CimServiceResume(Uint32 routing,

AsyncOpNode *operation,

Uint32 destination,

Uint32 response,

Boolean blocking);

virtual ~CimServiceResume(void)

{

}

};

</pre>

</li>

class PEGASUS_COMMON_LINKAGE AsyncOperationStart : public AsyncRequest

<li> Write the request handling block.

<pre>

void my_class::handle_operation(AsyncOpNode *op)

{

public:

if(op == NULL)

AsyncOperationStart(Uint32 routing,

AsyncOpNode *operation,

Uint32 destination,

Uint32 response,

Boolean blocking,

Message *action);

virtual ~AsyncOperationStart(void)

{

AsyncOpNode *op = get_op();

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

op->put_request(req);

}

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

Message *act;

};

class PEGASUS_COMMON_LINKAGE AsyncOperationResult : public AsyncReply

{

public:

AsyncOperationResult(Uint32 key,

Uint32 routing,

AsyncOpNode *operation,

Uint32 result_code,

Uint32 destination,

Uint32 blocking);

virtual ~AsyncOperationResult(void)

{

}

};

else

class PEGASUS_COMMON_LINKAGE AsyncLegacyOperationStart : public AsyncRequest

{

public:

my_response *res = op->get_response();

AsyncLegacyOperationStart(Uint32 routing,

if(res != 0 )

AsyncOpNode *operation,

Uint32 destination,

Message *action);

virtual ~AsyncLegacyOperationStart(void)

{

// handle response

delete response;

}

put_op(op);

Message *act;

};

class PEGASUS_COMMON_LINKAGE AsyncLegacyOperationResult : public AsyncReply

{

public:

AsyncLegacyOperationResult(Uint32 key,

Uint32 routing,

AsyncOpNode *operation,

Message *result);

virtual ~AsyncLegacyOperationResult(void)

{

}

return;

Message *res;

};

class PEGASUS_COMMON_LINKAGE FindServiceQueue : public AsyncRequest

{

public:

FindServiceQueue(Uint32 routing,

AsyncOpNode *operation,

Uint32 response,

Boolean blocking,

String service_name,

Uint32 service_capabilities,

Uint32 service_mask);

virtual ~FindServiceQueue(void)

{

}

String name;

</pre>

Uint32 capabilities;

</li>

Uint32 mask;

<li> Have your callback method call your handler

} ;

method.

<pre>

class PEGASUS_COMMON_LINKAGE FindServiceQueueResult : public AsyncReply

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

{

public:

FindServiceQueueResult(Uint32 key,

Uint32 routing,

AsyncOpNode *operation,

Uint32 result_code,

Uint32 destination,

Boolean blocking,

Array<Uint32> queue_ids);

virtual ~FindServiceQueueResult(void)

{

my_class *myself = static_cast<queue>

myself->handle_operation(op);

return;

}

</pre>

</li>

</ol>

</div>

Array<Uint32> qids;

} ;

class PEGASUS_COMMON_LINKAGE EnumerateService : public AsyncRequest

<h2>Class Definitions</h2>

{

public:

EnumerateService(Uint32 routing,

AsyncOpNode *operation,

Uint32 response,

Boolean blocking,

Uint32 queue_id);

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

virtual ~EnumerateService(void)

{

//%///-*-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;

}

friend class MessageQueueService;

Uint32 qid;

};

};

class PEGASUS_COMMON_LINKAGE EnumerateServiceResponse : public AsyncReply

PEGASUS_NAMESPACE_END

{

public:

EnumerateServiceResponse(Uint32 key,

Uint32 routing,

AsyncOpNode *operation,

Uint32 result_code,

Uint32 response,

Boolean blocking,

String service_name,

Uint32 service_capabilities,

Uint32 service_mask,

Uint32 service_qid);

#endif // CIMOM_include

</tt>

</pre>

virtual ~EnumerateServiceResponse(void)

<h3>MessageQueueService</h3>

{

}

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