359 mday 1.1
360
361 </pre>
362
363 The code sample above shows how to find services by their
364 name. The api also allows finding services by their capabilities
365 or the messages they support. Note that the return is an array
366 of Queue IDs. It is possible, for example, to find multiple
367 services.
368 </p>
369
370 <h2>Sending an Asynchronous Message to Another Service</h2>
371 <p>
372 The "handle" for a services is its Queue ID. Once you have the
373 Queue ID you can send a message to that service. The example
374 above shows one way to get a service's Queue ID. Here is an
375 example that shows how to send that service a message.
376
377 <ol>
378 <li><b>Define the Request and Response Message Pair by Inheriting from AsyncMessage.</b></li>
379 <pre>
380 mday 1.1
381 class test_request : public AsyncRequest
382 {
383
384 public:
385 typedef AsyncRequest Base;
386
387 test_request(Uint32 routing,
388 AsyncOpNode *op,
389 Uint32 destination,
390 Uint32 response,
391 char *message)
392 : Base(0x04100000,
393 Message::getNextKey(),
394 routing,
395 0,
396 op,
397 destination,
398 response,
399 true),
400 greeting(message)
401 mday 1.1 {
402
403 }
404
405 virtual ~test_request(void)
406 {
407
408
409 }
410
411 String greeting;
412 };
413
414
415 class test_response : public AsyncReply
416 {
417 public:
418 typedef AsyncReply Base;
419
420
421 test_response(Uint32 key,
422 mday 1.1 Uint32 routing,
423 AsyncOpNode *op,
424 Uint32 result,
425 Uint32 destination,
426 char *message)
427 : Base(0x04200000,
428 key,
429 routing,
430 0,
431 op,
432 result,
433 destination,
434 true),
435 greeting(message)
436 {
437
438 }
439
440 virtual ~test_response(void)
441 {
442
443 mday 1.1 }
444
445 String greeting;
446 };
447
448 </pre>
449
450 The function <code>send_test_request</code> shows everything
451 that is necessary to send a message to another service and
452 process the reply.
453
454 <pre>
455
456 void MessageQueueClient::send_test_request(char *greeting, Uint32 qid)
457 {
458
459 </pre>
460 <li><b>Construct the Request</b></li>
461
462 <pre>
463 test_request *req =
464 mday 1.1 new test_request(Base::get_next_xid(),
465 0,
466 qid, // destination queue ID
467 _queueId, // my own queue ID
468 greeting); // message parameter
469
470 </pre>
471
472 <li><b>Send the message using <code>MessageQueueService::SendWait</code></b></li>
473
474 <pre>
475 AsyncMessage *response = SendWait(req);
476
477 </pre>
478
479 <li><b>Process the Response.</b></i>
480 <pre>
481 if( response != 0 )
482 {
483 msg_count++;
484 delete response;
485 mday 1.1 cout << " test message " << msg_count.value() << endl;
486
487 }
488 delete req;
489 }
490
491 </pre>
492
493 <li><b>Delete the Request and the Response. The
494 <code>SendWait</code> interface creates and disposes of
495 everything else.</b></li>
496
497 </ol>
498 </p>
499
500
501 <h2>Handling an Incoming Message </h2>
502
503 <p>
504 To handle messages the service needs to implement the
505 following methods.
506 mday 1.1
507 <ol>
508
509 <li><b><code>virtual Boolean MessageOK(const Message
510 *)</code></b></li>
511
512 This method allows the Service to accept or reject the
513 message. The Meta Dispatcher will always call this method
514 before inserting the request on the Service's queue.
515
516 <pre>
517
518 Boolean MessageQueueServer::messageOK(const Message *msg)
519 {
520 if(msg->getMask() & message_mask::ha_async)
521 {
522 if( msg->getType() == 0x04100000 ||
523 msg->getType() == async_messages::CIMSERVICE_STOP ||
524 msg->getType() == async_messages::CIMSERVICE_PAUSE ||
525 msg->getType() == async_messages::CIMSERVICE_RESUME )
526 return true;
527 mday 1.1 }
528 return false;
529 }
530
531 </pre>
532
533 <li><b><code>virtual Boolean accept_async(AsyncOpNode
534 *operation)</code> (optional) </b></li>
535
536 This method executes on the Meta Dispatcher's thread and links
537 the incoming message to the Service's queue. <br><br>
538
539
540
541 <li><b><code>virtual void _handle_incoming_operation(AsyncOpNode
542 *)</code></b></li><br>
543
544
545 This method is called by the Service's background thread. Here is an
546 example implementation that just does some sanity checking on
547 the message.
548 mday 1.1
549 <pre>
550
551 void MessageQueueServer::_handle_incoming_operation(AsyncOpNode *op)
552 {
553 if ( operation != 0 )
554 {
555 Message *rq = operation->get_request();
556 PEGASUS_ASSERT(rq != 0 );
557 PEGASUS_ASSERT(rq->getMask() & message_mask::ha_async );
558 PEGASUS_ASSERT(rq->getMask() & message_mask::ha_request);
559 _handle_async_request(static_cast<AsyncRequest *>(rq));
560 }
561
562 return;
563
564 }
565
566
567 </pre>
568
569 mday 1.1 <li><b><code>virtual void _handle_async_request(AsyncRequest *)</code></b></li><br>
570 <br>
571
572 This method handles the request. The Service must implement
573 this method. <b>If the Service does not handle the Request it
574 must pass the Request to the Base class by calling <code>Base::_handle_async_request(req)</code></b>
575
576 <pre>
577 void MessageQueueServer::_handle_async_request(AsyncRequest *req)
578 {
579 if (req->getType() == 0x04100000 )
580 {
581 req->op->processing();
582 handle_test_request(req); // Message Handler
583 }
584 else if ( req->getType() == async_messages::CIMSERVICE_STOP )
585 {
586 req->op->processing();
587 handle_CimServiceStop(static_cast<CimServiceStop *>(req));
588 }
589
590 mday 1.1 else
591 Base::_handle_async_request(req); // Give it to the Base !!
592 }
593
594 </pre>
595
596 <li><b>Specific Message Handlers</b>
597
598 Each Message handler will be defined by the format of the
599 Request/Response pair. Here is an example from the test
600 program.
601
602 <pre>
603
604 if( msg->getType() == 0x04100000 )
605 {
606
607 </pre>
608 <ol>
609 <li><b>Construct the Reply</b></li>
610 <pre>
611 mday 1.1
612 test_response *resp =
613 new test_response(msg->getKey(),
614 msg->getRouting(),
615 msg->op,
616 async_results::OK,
617 msg->dest,
618 "i am a test response");
619
620
621 </pre>
622 <li><b>Complete the Reply</b> by calling the following
623 helper routine in the Base class</li>
624
625 <pre>
626 _completeAsyncResponse(msg, resp, ASYNC_OPSTATE_COMPLETE, 0);
627
628 }
629 </pre>
630
631 </ol>
632 mday 1.1 </p>
633
|
634 mday 1.2 <h2>Handling CIMMessage and Other Pre-existing Message Classes</h2>
635 <p>
636 Existing Messages, including all of the <code>CIMMessage</code>
637 derivitives, are not configured to be asynchronous
638 request/reply pairs. They are designed to travel through
639 Pegasus as events that trigger other processing events,
640 which is the end of their lifetime. This is not an optimal
641 use model for asynchronous operation because the
642 originator of the event does not require nor receive any
643 completion notification. Further, there is not a
644 one-to-one correspondence of "event messages" to replies.
645 </p>
646
647 <h3>AsyncLegacyOperationStart Message</h3>
648 <p>
649 The AsyncLegacyOperationStart message is an envelope that
650 allows a <code>MessageQueueService</code>-based service to
651 send, receive, and process pre-existing "legacy"
652 messages.
653 </p>
654 <p>
655 mday 1.2 The <code>AsyncLegacyOperationStart</code> Message allows
656 an asynchronous service to create, package, and send a
657 "legacy" message to another service or, indirectly,
658 enqueue it to a non-asynchronous message queue. The code
659 example below shows how this works:
660 </p>
661
662 <pre>
663
664 cout << " sending LEGACY to test server" << endl;
665
666 Message *legacy = new Message(0x11100011,
667 Message::getNextKey());
668
669 AsyncLegacyOperationStart *req =
670 new AsyncLegacyOperationStart(q_client->get_next_xid(),
671 0,
672 services[0],
673 legacy,
674 q_client->getQueueId());
675 reply = q_client->SendWait(req);
676 mday 1.2 delete req;
677 delete reply;
678
679 </pre>
680 <p>
681 The code sample above shows a <code>Message</code> object
682 being embedded inside an
683 <code>AsyncLegacyOperationStart</code> message and sent
684 using the <code>SendWait</code>API.
685 </p>
686
687 <h3>Default Handler for Legacy Messages</h3>
688 <p>
689 The <code>MessageQueueService</code> class has a default
690 handler for legacy messages that extracts the
691 <code>Message</code> out of its asynchronous "envelope"
692 and dispatches it using the pre-existing synchronous
693 interface, as shown below.
694 </p>
695
696 <pre>
697 mday 1.2
698 void MessageQueueService::handle_AsyncLegacyOperationStart(
699 AsyncLegacyOperationStart *req)
700 {
701 // remove the legacy message from the request and enqueue it to its destination
702 Uint32 result = async_results::CIM_NAK;
703
704 Message *legacy = req->act;
705 if ( legacy != 0 )
706 {
707 MessageQueue* queue = MessageQueue::lookup(req->legacy_destination);
708 if( queue != 0 )
709 {
710 // Enqueue the response:
711 queue->enqueue(legacy);
712 result = async_results::OK;
713 }
714 }
715 _make_response(req, result);
716 }
717
718 mday 1.2 </pre>
719
720 <p>
721 The default handler shown above extracts the legacy
722 message and attempts to <code>enqueue</code> that message
723 syncrhonously using the pre-existing interface.
724 </p>
725
726 <h3>Example of Custom Handler for Legacy Messages</h3>
727 <p>
728 By implementing the virtual
729 <code>_handle_async_request</code> method,
730 a service can choose to implement its own handler for
731 Legacy messages, as the code below shows:
732 </p>
733
734 <ol>
735
736 <li><b>Implement the virtual <code>_handle_async_request</code> method.</b></li>
737 <pre>
738
739 mday 1.2 void MessageQueueServer::_handle_async_request(AsyncRequest *req)
740 {
741 if (req->getType() == 0x04100000 )
742 {
743 req->op->processing();
744 handle_test_request(req);
745 }
746 else if ( req->getType() == async_messages::CIMSERVICE_STOP )
747 {
748 req->op->processing();
749 handle_CimServiceStop(static_cast<CimServiceStop *>(req));
750 }
751 </pre>
752 <li><b>Implement a dispatcher for <code>ASYNC_LEGACY_OP_START</code></b></li>
753 <pre>
754 else if ( req->getType() == async_messages::ASYNC_LEGACY_OP_START )
755 {
756 req->op->processing();
757 handle_LegacyOpStart(static_cast<AsyncLegacyOperationStart *>(req));
758 }
759
760 mday 1.2 else
761 Base::_handle_async_request(req);
762 }
763
764 </pre>
765 <li><b>Implement a dispatcher for <code>ASYNC_LEGACY_OP_START</code></b></li>
766 <pre>
767
768 void MessageQueueServer::handle_LegacyOpStart(AsyncLegacyOperationStart *req)
769 {
770
771 Message *legacy = req->act;
772 cout << " ### handling legacy messages " << endl;
773
774
775 AsyncReply *resp =
776 new AsyncReply(async_messages::REPLY,
777 req->getKey(),
778 req->getRouting(),
779 0,
780 req->op,
781 mday 1.2 async_results::OK,
782 req->resp,
783 req->block);
784 _completeAsyncResponse(req, resp, ASYNC_OPSTATE_COMPLETE, 0 );
785
786 if (legacy != 0 )
787 cout << " legacy msg type: " << legacy->getType() << endl;
788
789 }
790
791 </pre>
792
793
794 </ol>
|
855 mday 1.1 <h2>Class Definitions</h2>
856
857 <h3>cimom (Meta Dispatcher)</h3>
858 <pre>
859 class PEGASUS_COMMON_LINKAGE cimom : public MessageQueue
860 {
861 public :
862 cimom(void);
863
864 virtual ~cimom(void) ;
865
866 Boolean moduleChange(struct timeval last);
867
868 Uint32 getModuleCount(void);
869 Uint32 getModuleIDs(Uint32 *ids, Uint32 count) throw(IPCException);
870
871 AsyncOpNode *get_cached_op(void) throw(IPCException);
872 void cache_op(AsyncOpNode *op) throw(IPCException);
873
874 void set_default_op_timeout(const struct timeval *buffer);
875 void get_default_op_timeout(struct timeval *timeout) const ;
876 mday 1.1
877 virtual void handleEnqueue();
878 void register_module(RegisterCimService *msg);
879 void deregister_module(Uint32 quid);
880 void update_module(UpdateCimService *msg );
881 void ioctl(AsyncIoctl *msg );
882
883 void find_service_q(FindServiceQueue *msg );
884 void enumerate_service(EnumerateService *msg );
885 Boolean route_async(AsyncOpNode *operation);
886 void _shutdown_routed_queue(void);
887
888
889 protected:
890 Uint32 get_module_q(const String & name);
891 void _make_response(AsyncRequest *req, Uint32 code);
892 void _completeAsyncResponse(AsyncRequest *request,
893 AsyncReply *reply,
894 Uint32 state,
895 Uint32 flag);
896 private:
897 mday 1.1 struct timeval _default_op_timeout;
898 struct timeval _last_module_change;
899 DQueue<message_module> _modules;
900
901 DQueue<AsyncOpNode> _recycle;
902
903 AsyncDQueue<AsyncOpNode> _routed_ops;
904 DQueue<AsyncOpNode> _internal_ops;
905
906 static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL _routing_proc(void *);
907
908 Thread _routing_thread;
909
910 static Uint32 get_xid(void);
911 void _handle_cimom_op(AsyncOpNode *op, Thread *thread, MessageQueue *queue);
912 Uint32 _ioctl(Uint32, Uint32, void *);
913
914
915 AtomicInt _die;
916 AtomicInt _routed_queue_shutdown;
917
918 mday 1.1 static AtomicInt _xid;
919
920 // CIMOperationRequestDispatcher *_cim_dispatcher;
921 // CIMOperationResponseEncoder *_cim_encoder;
922 // CIMOperationRequestDecoder *_cim_decoder;
923 // CIMRepository *_repository;
924
925 };
926
927 </pre>
928
929
930 <h3>MessageQueueService</h3>
931
932 <pre>
933
934 class message_module;
935
936 class PEGASUS_COMMON_LINKAGE MessageQueueService : public MessageQueue
937 {
938 public:
939 mday 1.1
940 typedef MessageQueue Base;
941
942 MessageQueueService(const char *name, Uint32 queueID, Uint32 capabilities, Uint32 mask) ;
943
944 virtual ~MessageQueueService(void);
945
946 virtual void handle_heartbeat_request(AsyncRequest *req);
947 virtual void handle_heartbeat_reply(AsyncReply *rep);
948
949 virtual void handle_AsyncIoctl(AsyncIoctl *req);
950 virtual void handle_CimServiceStart(CimServiceStart *req);
951 virtual void handle_CimServiceStop(CimServiceStop *req);
952 virtual void handle_CimServicePause(CimServicePause *req);
953 virtual void handle_CimServiceResume(CimServiceResume *req);
954
955 virtual void handle_AsyncOperationStart(AsyncOperationStart *req);
956 virtual void handle_AsyncOperationResult(AsyncOperationResult *req);
957 virtual Boolean accept_async(AsyncOpNode *op);
958 virtual Boolean messageOK(const Message *msg) ;
959
960 mday 1.1 AsyncReply *SendWait(AsyncRequest *request);
961
962 void _completeAsyncResponse(AsyncRequest *request,
963 AsyncReply *reply,
964 Uint32 state,
965 Uint32 flag);
966 Boolean register_service(String name, Uint32 capabilities, Uint32 mask);
967 Boolean update_service(Uint32 capabilities, Uint32 mask);
968 Boolean deregister_service(void);
969 virtual void _shutdown_incoming_queue(void);
970 void find_services(String name,
971 Uint32 capabilities,
972 Uint32 mask,
973 Array<Uint32> *results);
974 void enumerate_service(Uint32 queue, message_module *result);
975 Uint32 get_next_xid(void);
976 AsyncOpNode *get_op(void);
977 void return_op(AsyncOpNode *op);
978 Uint32 _capabilities;
979 Uint32 _mask;
980 AtomicInt _die;
981 mday 1.1 protected:
982
983 virtual void _handle_incoming_operation(AsyncOpNode *operation, Thread *thread, MessageQueue *queue);
984 virtual void _handle_async_request(AsyncRequest *req);
985 virtual void _make_response(AsyncRequest *req, Uint32 code);
986 cimom *_meta_dispatcher;
987
988 private:
989 void handleEnqueue();
990 DQueue<AsyncOpNode> _pending;
991 AsyncDQueue<AsyncOpNode> _incoming;
992
993 static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL _req_proc(void *);
994 AtomicInt _incoming_queue_shutdown;
995
996 Thread _req_thread;
997
998 struct timeval _default_op_timeout;
999
1000 static AtomicInt _xid;
1001
1002 mday 1.1 };
1003
1004
1005 </pre>
1006
1007 <h3>Asynchronous Messages</h3>
1008
1009 <pre>
1010
1011 extern const Uint32 CIMOM_Q_ID;
1012
1013 class AsyncOpNode;
1014
1015 class PEGASUS_COMMON_LINKAGE async_results
1016 {
1017 public:
1018 static const Uint32 OK;
1019 static const Uint32 PARAMETER_ERROR;
1020 static const Uint32 MODULE_ALREADY_REGISTERED;
1021 static const Uint32 MODULE_NOT_FOUND;
1022 static const Uint32 INTERNAL_ERROR;
1023 mday 1.1
1024 static const Uint32 ASYNC_STARTED;
1025 static const Uint32 ASYNC_PROCESSING;
1026 static const Uint32 ASYNC_COMPLETE;
1027 static const Uint32 ASYNC_CANCELLED;
1028 static const Uint32 ASYNC_PAUSED;
1029 static const Uint32 ASYNC_RESUMED;
1030
1031 static const Uint32 CIM_SERVICE_STARTED;
1032 static const Uint32 CIM_SERVICE_STOPPED;
1033 static const Uint32 CIM_SERVICE_PAUSED;
1034
1035 static const Uint32 CIM_SERVICE_RESUMED;
1036 static const Uint32 CIM_NAK;
1037
1038 static const Uint32 ASYNC_PHASE_COMPLETE;
1039 static const Uint32 ASYNC_CHILD_COMPLETE;
1040 static const Uint32 ASYNC_PHASE_STARTED;
1041 static const Uint32 ASYNC_CHILD_STARTED;
1042 static const Uint32 CIM_PAUSED;
1043 static const Uint32 CIM_STOPPED;
1044 mday 1.1
1045 };
1046
1047
1048 class PEGASUS_COMMON_LINKAGE async_messages
1049 {
1050 public:
1051 static const Uint32 HEARTBEAT;
1052 static const Uint32 REPLY;
1053 static const Uint32 REGISTER_CIM_SERVICE;
1054 static const Uint32 DEREGISTER_CIM_SERVICE;
1055 static const Uint32 UPDATE_CIM_SERVICE;
1056 static const Uint32 IOCTL;
1057 static const Uint32 CIMSERVICE_START;
1058 static const Uint32 CIMSERVICE_STOP;
1059 static const Uint32 CIMSERVICE_PAUSE;
1060 static const Uint32 CIMSERVICE_RESUME;
1061
1062 static const Uint32 ASYNC_OP_START;
1063 static const Uint32 ASYNC_OP_RESULT;
1064 static const Uint32 ASYNC_LEGACY_OP_START;
1065 mday 1.1 static const Uint32 ASYNC_LEGACY_OP_RESULT;
1066
1067 static const Uint32 FIND_SERVICE_Q;
1068 static const Uint32 FIND_SERVICE_Q_RESULT;
1069 static const Uint32 ENUMERATE_SERVICE;
1070 static const Uint32 ENUMERATE_SERVICE_RESULT;
1071 };
1072
1073
1074 class PEGASUS_COMMON_LINKAGE AsyncMessage : public Message
1075 {
1076 public:
1077 AsyncMessage(Uint32 type,
1078 Uint32 key,
1079 Uint32 routing,
1080 Uint32 mask,
1081 AsyncOpNode *operation);
1082
1083 virtual ~AsyncMessage(void)
1084 {
1085
1086 mday 1.1 }
1087
1088 Boolean operator ==(void *key);
1089 Boolean operator ==(const AsyncMessage& msg);
1090
1091 AsyncOpNode *op;
1092 Thread *_myself;
1093 MessageQueue *_service;
1094 };
1095
1096
1097 inline Boolean AsyncMessage::operator ==(void *key)
1098 {
1099 if( key == reinterpret_cast<void *>(this))
1100 return true;
1101 return false;
1102 }
1103
1104 inline Boolean AsyncMessage::operator ==(const AsyncMessage& msg)
1105 {
1106 return this->operator==(reinterpret_cast<void *>(const_cast<AsyncMessage *>(&msg)));
1107 mday 1.1 }
1108
1109
1110 class PEGASUS_COMMON_LINKAGE AsyncRequest : public AsyncMessage
1111 {
1112 public:
1113 AsyncRequest(Uint32 type,
1114 Uint32 key,
1115 Uint32 routing,
1116 Uint32 mask,
1117 AsyncOpNode *operation,
1118 Uint32 destination,
1119 Uint32 response,
1120 Boolean blocking);
1121
1122
1123 virtual ~AsyncRequest(void)
1124 {
1125
1126 }
1127
1128 mday 1.1 Uint32 dest;
1129 Uint32 resp;
1130 Boolean block;
1131 };
1132
1133 class PEGASUS_COMMON_LINKAGE AsyncReply : public AsyncMessage
1134 {
1135 public:
1136 AsyncReply(Uint32 type,
1137 Uint32 key,
1138 Uint32 routing,
1139 Uint32 mask,
1140 AsyncOpNode *operation,
1141 Uint32 result_code,
1142 Uint32 destination,
1143 Boolean blocking);
1144
1145
1146 virtual ~AsyncReply(void)
1147 {
1148 if(op != 0 )
1149 mday 1.1 delete op;
1150
1151 }
1152
1153 Uint32 result;
1154 Uint32 dest;
1155 Boolean block;
1156 };
1157
1158
1159
1160 class PEGASUS_COMMON_LINKAGE RegisterCimService : public AsyncRequest
1161 {
1162 public:
1163 RegisterCimService(Uint32 routing,
1164 AsyncOpNode *operation,
1165 Boolean blocking,
1166 String service_name,
1167 Uint32 service_capabilities,
1168 Uint32 service_mask,
1169 Uint32 service_queue);
1170 mday 1.1
1171 virtual ~RegisterCimService(void)
1172 {
1173
1174 }
1175
1176 String name;
1177 Uint32 capabilities;
1178 Uint32 mask;
1179 Uint32 queue;
1180 };
1181
1182 class PEGASUS_COMMON_LINKAGE DeRegisterCimService : public AsyncRequest
1183 {
1184 public:
1185 DeRegisterCimService(Uint32 routing,
1186 AsyncOpNode *operation,
1187 Boolean blocking,
1188 Uint32 service_queue);
1189
1190
1191 mday 1.1 virtual ~DeRegisterCimService(void)
1192 {
1193
1194 }
1195
1196 Uint32 queue;
1197 } ;
1198
1199 class PEGASUS_COMMON_LINKAGE UpdateCimService : public AsyncRequest
1200 {
1201 public:
1202 UpdateCimService(Uint32 routing,
1203 AsyncOpNode *operation,
1204 Boolean blocking,
1205 Uint32 service_queue,
1206 Uint32 service_capabilities,
1207 Uint32 service_mask);
1208
1209 virtual ~UpdateCimService(void)
1210 {
1211
1212 mday 1.1 }
1213
1214 Uint32 queue;
1215 Uint32 capabilities;
1216 Uint32 mask;
1217 };
1218
1219
1220 class PEGASUS_COMMON_LINKAGE AsyncIoctl : public AsyncRequest
1221 {
1222 public:
1223 AsyncIoctl(Uint32 routing,
1224 AsyncOpNode *operation,
1225 Uint32 destination,
1226 Uint32 response,
1227 Boolean blocking,
1228 Uint32 code,
1229 Uint32 int_param,
1230 void *p_param);
1231
1232 virtual ~AsyncIoctl(void)
1233 mday 1.1 {
1234
1235 }
1236
1237 enum
1238 {
1239 IO_CLOSE,
1240 IO_OPEN,
1241 IO_SOURCE_QUENCH,
1242 IO_SERVICE_DEFINED
1243 };
1244
1245
1246
1247 Uint32 ctl;
1248 Uint32 intp;
1249 void *voidp;
1250
1251 };
1252
1253 class PEGASUS_COMMON_LINKAGE CimServiceStart : public AsyncRequest
1254 mday 1.1 {
1255 public:
1256 CimServiceStart(Uint32 routing,
1257 AsyncOpNode *operation,
1258 Uint32 destination,
1259 Uint32 response,
1260 Boolean blocking);
1261
1262 virtual ~CimServiceStart(void)
1263 {
1264
1265 }
1266 };
1267
1268
1269 class PEGASUS_COMMON_LINKAGE CimServiceStop : public AsyncRequest
1270 {
1271 public:
1272 CimServiceStop(Uint32 routing,
1273 AsyncOpNode *operation,
1274 Uint32 destination,
1275 mday 1.1 Uint32 response,
1276 Boolean blocking);
1277
1278 virtual ~CimServiceStop(void)
1279 {
1280
1281 }
1282 };
1283
1284 class PEGASUS_COMMON_LINKAGE CimServicePause : public AsyncRequest
1285 {
1286 public:
1287 CimServicePause(Uint32 routing,
1288 AsyncOpNode *operation,
1289 Uint32 destination,
1290 Uint32 response,
1291 Boolean blocking);
1292
1293
1294 virtual ~CimServicePause(void)
1295 {
1296 mday 1.1
1297 }
1298 };
1299
1300 class PEGASUS_COMMON_LINKAGE CimServiceResume : public AsyncRequest
1301 {
1302 public:
1303 CimServiceResume(Uint32 routing,
1304 AsyncOpNode *operation,
1305 Uint32 destination,
1306 Uint32 response,
1307 Boolean blocking);
1308
1309
1310 virtual ~CimServiceResume(void)
1311 {
1312
1313 }
1314 };
1315
1316 class PEGASUS_COMMON_LINKAGE AsyncOperationStart : public AsyncRequest
1317 mday 1.1 {
1318 public:
1319 AsyncOperationStart(Uint32 routing,
1320 AsyncOpNode *operation,
1321 Uint32 destination,
1322 Uint32 response,
1323 Boolean blocking,
1324 Message *action);
1325
1326
1327 virtual ~AsyncOperationStart(void)
1328 {
1329
1330 }
1331
1332 Message *act;
1333 };
1334
1335 class PEGASUS_COMMON_LINKAGE AsyncOperationResult : public AsyncReply
1336 {
1337 public:
1338 mday 1.1 AsyncOperationResult(Uint32 key,
1339 Uint32 routing,
1340 AsyncOpNode *operation,
1341 Uint32 result_code,
1342 Uint32 destination,
1343 Uint32 blocking);
1344
1345
1346 virtual ~AsyncOperationResult(void)
1347 {
1348
1349 }
1350 };
1351
1352
1353 class PEGASUS_COMMON_LINKAGE AsyncLegacyOperationStart : public AsyncRequest
1354 {
1355 public:
1356 AsyncLegacyOperationStart(Uint32 routing,
1357 AsyncOpNode *operation,
1358 Uint32 destination,
1359 mday 1.1 Message *action);
1360
1361
1362 virtual ~AsyncLegacyOperationStart(void)
1363 {
1364
1365 }
1366
1367 Message *act;
1368 };
1369
1370 class PEGASUS_COMMON_LINKAGE AsyncLegacyOperationResult : public AsyncReply
1371 {
1372 public:
1373 AsyncLegacyOperationResult(Uint32 key,
1374 Uint32 routing,
1375 AsyncOpNode *operation,
1376 Message *result);
1377
1378 virtual ~AsyncLegacyOperationResult(void)
1379 {
1380 mday 1.1
1381 }
1382
1383 Message *res;
1384 };
1385
1386
1387 class PEGASUS_COMMON_LINKAGE FindServiceQueue : public AsyncRequest
1388 {
1389 public:
1390 FindServiceQueue(Uint32 routing,
1391 AsyncOpNode *operation,
1392 Uint32 response,
1393 Boolean blocking,
1394 String service_name,
1395 Uint32 service_capabilities,
1396 Uint32 service_mask);
1397
1398 virtual ~FindServiceQueue(void)
1399 {
1400
1401 mday 1.1 }
1402
1403 String name;
1404 Uint32 capabilities;
1405 Uint32 mask;
1406 } ;
1407
1408 class PEGASUS_COMMON_LINKAGE FindServiceQueueResult : public AsyncReply
1409 {
1410 public:
1411 FindServiceQueueResult(Uint32 key,
1412 Uint32 routing,
1413 AsyncOpNode *operation,
1414 Uint32 result_code,
1415 Uint32 destination,
1416 Boolean blocking,
1417 Array<Uint32> queue_ids);
1418
1419
1420 virtual ~FindServiceQueueResult(void)
1421 {
1422 mday 1.1
1423 }
1424
1425 Array<Uint32> qids;
1426 } ;
1427
1428 class PEGASUS_COMMON_LINKAGE EnumerateService : public AsyncRequest
1429 {
1430 public:
1431 EnumerateService(Uint32 routing,
1432 AsyncOpNode *operation,
1433 Uint32 response,
1434 Boolean blocking,
1435 Uint32 queue_id);
1436
1437
1438 virtual ~EnumerateService(void)
1439 {
1440
1441 }
1442
1443 mday 1.1 Uint32 qid;
1444 };
1445
1446 class PEGASUS_COMMON_LINKAGE EnumerateServiceResponse : public AsyncReply
1447 {
1448 public:
1449 EnumerateServiceResponse(Uint32 key,
1450 Uint32 routing,
1451 AsyncOpNode *operation,
1452 Uint32 result_code,
1453 Uint32 response,
1454 Boolean blocking,
1455 String service_name,
1456 Uint32 service_capabilities,
1457 Uint32 service_mask,
1458 Uint32 service_qid);
1459
1460
1461 virtual ~EnumerateServiceResponse(void)
1462 {
1463
1464 mday 1.1 }
1465
1466 String name;
1467 Uint32 capabilities;
1468 Uint32 mask;
1469 Uint32 qid;
1470 };
1471
1472 </pre>
1473
1474 <h3>AsyncOPNode</h3>
1475
1476 <pre>
1477 #define ASYNC_OPFLAGS_UNKNOWN 0x00000000
1478 #define ASYNC_OPFLAGS_INTERVAL_REPEAT 0x00000010
1479 #define ASYNC_OPFLAGS_INDICATION 0x00000020
1480 #define ASYNC_OPFLAGS_REMOTE 0x00000040
1481 #define ASYNC_OPFLAGS_LOCAL_OUT_OF_PROC 0x00000080
1482 #define ASYNC_OPFLAGS_PHASED 0x00000001
1483 #define ASYNC_OPFLAGS_PARTIAL 0x00000002
1484 #define ASYNC_OPFLAGS_NORMAL 0x00000000
1485 mday 1.1 #define ASYNC_OPFLAGS_SINGLE 0x00000008
1486 #define ASYNC_OPFLAGS_MULTIPLE 0x00000010
1487 #define ASYNC_OPFLAGS_TOTAL 0x00000020
1488 #define ASYNC_OPFLAGS_META_DISPATCHER 0x00000040
1489
1490 #define ASYNC_OPSTATE_UNKNOWN 0x00000000
1491 #define ASYNC_OPSTATE_OFFERED 0x00000001
1492 #define ASYNC_OPSTATE_DECLINED 0x00000002
1493 #define ASYNC_OPSTATE_STARTED 0x00000004
1494 #define ASYNC_OPSTATE_PROCESSING 0x00000008
1495 #define ASYNC_OPSTATE_DELIVER 0x00000010
1496 #define ASYNC_OPSTATE_RESERVE 0x00000020
1497 #define ASYNC_OPSTATE_COMPLETE 0x00000040
1498 #define ASYNC_OPSTATE_TIMEOUT 0x00000080
1499 #define ASYNC_OPSTATE_CANCELLED 0x00000100
1500 #define ASYNC_OPSTATE_PAUSED 0x00000200
1501 #define ASYNC_OPSTATE_SUSPENDED 0x00000400
1502 #define ASYNC_OPSTATE_RESUMED 0x00000800
1503 #define ASYNC_OPSTATE_ORPHANED 0x00001000
1504 #define ASYNC_OPSTATE_RELEASED 0x00002000
1505
1506 mday 1.1 class Cimom;
1507
1508 class PEGASUS_COMMON_LINKAGE AsyncOpNode
1509 {
1510 public:
1511
1512 AsyncOpNode(void);
1513 ~AsyncOpNode(void);
1514
1515 Boolean operator == (const void *key) const;
1516 Boolean operator == (const AsyncOpNode & node) const;
1517
1518 void get_timeout_interval(struct timeval *buffer) ;
1519 void set_timeout_interval(const struct timeval *interval);
1520
1521 Boolean timeout(void) ;
1522
1523 OperationContext & get_context(void) ;
1524
1525 void put_request(const Message *request) ;
1526 Message *get_request(void) ;
1527 mday 1.1
1528 void put_response(const Message *response) ;
1529 Message *get_response(void) ;
1530
1531 Uint32 read_state(void) ;
1532 void write_state(Uint32) ;
1533
1534 Uint32 read_flags(void);
1535 void write_flags(Uint32);
1536
1537 void lock(void) throw(IPCException);
1538 void unlock(void) throw(IPCException);
1539 void udpate(void) throw(IPCException);
1540 void deliver(const Uint32 count) throw(IPCException);
1541 void reserve(const Uint32 size) throw(IPCException);
1542 void processing(void) throw(IPCException) ;
1543 void processing(OperationContext *context) throw(IPCException);
1544 void complete(void) throw(IPCException) ;
1545 void complete(OperationContext *context) throw(IPCException);
1546 void release(void);
1547 void wait(void);
1548 mday 1.1
1549
1550 private:
1551 Semaphore _client_sem;
1552 Mutex _mut;
1553 unlocked_dq<Message> _request;
1554 unlocked_dq<Message> _response;
1555
1556 OperationContext _operation_list;
1557 Uint32 _state;
1558 Uint32 _flags;
1559 Uint32 _offered_count;
1560 Uint32 _total_ops;
1561 Uint32 _completed_ops;
1562 Uint32 _user_data;
1563
1564 struct timeval _start;
1565 struct timeval _lifetime;
1566 struct timeval _updated;
1567 struct timeval _timeout_interval;
1568
1569 mday 1.1 AsyncOpNode *_parent;
1570 unlocked_dq<AsyncOpNode> _children;
1571
1572 void _reset(unlocked_dq<AsyncOpNode> *dst_q);
1573
1574 // the lifetime member is for cache management by the cimom
1575 void _set_lifetime(struct timeval *lifetime) ;
1576 Boolean _check_lifetime(void) ;
1577
1578 Boolean _is_child(void) ;
1579 Uint32 _is_parent(void) ;
1580
1581 Boolean _is_my_child(const AsyncOpNode & caller) const;
1582 void _make_orphan( AsyncOpNode & parent) ;
1583 void _adopt_child(AsyncOpNode *child) ;
1584 void _disown_child(AsyncOpNode *child) ;
1585 friend class cimom;
1586 friend class MessageQueueService;
1587
1588 };
1589 </pre>
1590 mday 1.1
1591 <hr>
1592 <address><a href="mailto:mdday@us.ibm.com">Michael Day</a></address>
1593 <!-- Created: Tue Feb 5 13:21:55 EST 2002 -->
1594 <!-- hhmts start -->
|
Return to
MessageQueueService.html
CVS log
Up to [Pegasus] / pegasus / doc