1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
|
<!DOCTYPE HTML PUBLIC "-//IETF//DTD HTML//EN">
<html>
<head>
<title>TAO's RT Event Service</title>
<!-- $Id$ -->
</head>
<BODY text = "#000000"
link="#000fff"
vlink="#ff0f0f"
bgcolor="#ffffff">
<h3>TAO's RT Event Service</h3>
<P>We have already explored how to use
<A HREF="../Event_Service/index.html">
TAO's COS Event Service
</A>
to receive updated stock prices,
but what if we are not interested in all the stocks?
One approach is to use multiple event channels, each one
carrying different traffic.
For example,
each event channel could carry only a subset of the stocks.
In this section we will explore another solution,
using TAO's real-time Event Service to perform filtering for us.
TAO's real-time Event Service can do many other things,
like preserving priority end-to-end,
using multicast to conserve network resources,
generating timeout and interval events,
and it can collaborate with TAO's Scheduling Service to
analyze the schedulability of your system.
</P>
<H3>Getting the Price Changes</H3>
<P>For this example we will use the same data structures that we
used in the previous example,
i.e., the events will be identical.
TAO's RT Event Service can be configured to carry your events in
a type-safe manner,
or you can use custom marshaling to send non-IDL structures in
the event,
but it is easier to use it like the COS Event Service first.
</P>
<H3>Connecting as a consumer</H3>
<P>Connecting as a consumer is very similar. Some of the base
classes and signatures change, but it is basically the same
idea:
First let us define the consumer object:
</P>
<PRE>
class Stock_Consumer : public POA_RtecEventComm::PushConsumer {
public:
Stock_Consumer ();
void push (const RtecEventComm::EventSet& data);
void disconnect_push_consumer (void);
// details omitted
};
</PRE>
<P>Notice that we receive an event set instead of a single event.
The event channel can use this feature to queue multiple events
and push them in a single operation.
First we need to extract the event data from the any:
</P>
<PRE>
void
Stock_Consumer::push (const RtecEventComm::EventSet &data)
{
for (CORBA::ULong i = 0; i != data.length (); ++i) {
RtecEventComm::Event &e = data[i];
Quoter::Event *event;
if ((e.data.any_value >>= event) == 0)
continue; // Invalid event
</PRE>
<P>Notice that the events have more structure,
they have a clearly separated header and data,
and the data has more than just an any.
The header is used to provide filtering, and
the event data field can be configured at compile time to carry
whatever IDL structures you want.
Now we can print out the new stock price:
</P>
<PRE>
std::cout << "The new price for one stock in \""
<< event->full_name.in ()
<< "\" (" << event->symbol.in ()
<< ") is " << event->price << std::endl;
}
</PRE>
<P>We also need to implement the disconnect callback:
</P>
<PRE>
void
Stock_Consumer::disconnect_push_consumer (void)
{
this->supplier_proxy_ = CosEventChannelAdmin::ProxyPushSupplier::_nil ();
}
</PRE>
<P>As with the COS Event Channel we can voluntarily disconnect,
too:
</P>
<PRE>
void
Stock_Consumer::disconnect ()
{
// Do not receive any more events...
this->supplier_proxy_->disconnect_push_supplier ();
}
</PRE>
<H4>How to connect to the RT event channel</H4>
<P>Connecting to the RT event channel is very similar to
connecting to the regular event channel.
The only difference is that we must specify the events that we
want to receive. This is described using a fairly complex IDL
structure, but TAO provides a helper class to generate it.
We will assume that we are using the naming service or
something similar to obtain a reference to the event service:
</P>
<PRE>
CORBA::Object_var tmp = naming_context->resolve (name);
RtecEventChannelAdmin::EventChannel_var event_channel =
RtecEventChannelAdmin::EventChannel::_narrow (tmp);
</PRE>
<P>Now we use the event channel to obtain the factory used for
consumer connections:
</P>
<PRE>
RtecEventChannelAdmin::ConsumerAdmin_var consumer_admin =
event_channel->for_consumers ();
</PRE>
<P>And use the factory to obtain a proxy:
</P>
<PRE>
void
Stock_Consumer::connect (RtecEventChannelAdmin::ConsumerAdmin_ptr consumer_admin)
{
this->supplier_proxy_ =
consumer_admin->obtain_push_supplier ();
</PRE>
<P>Now we list the events that we want to receive.
We use a simple algorithm to assign an event type to each stock
symbol:
</P>
<PRE>
CORBA::ULong rhat_event_type =
(int('R') << 24) | (int('H') << 16) | (int('A') << 8) | int('T');
CORBA::ULong aaaa_event_type =
(int('A') << 24) | (int('A') << 16) | (int('A') << 8) | int('A');
</PRE>
<P>Now we create the subscription:
</P>
<PRE>
ACE_ConsumerQOS_Factory subscriptions;
subscriptions.insert_type (rhat_event_type, 0);
subscriptions.insert_type (aaaa_event_type, 0);
</PRE>
<P>And connect to the proxy:
</P>
<PRE>
RtecEventComm::PushConsumer_var myself = this->_this ();
this->supplier_proxy_->connect_push_consumer (
myself.in (),
subscriptions.get_ConsumerQOS ());
}
</PRE>
<H3>Notifying the Price Changes</H3>
<P>As with the COS Event Channel example we will make our
implementation of the <CODE>Modify_Stock</CODE> interface
generate events whenever the price changes:
</P>
<PRE>
class Quoter_Modify_Stock_i : public POA_Quoter::Modify_Stock {
public:
Quoter_Modify_Stock_i (const char *symbol,
const char *full_name,
CORBA::Double price);
void set_price (CORBA::Double new_price);
void disconnect_push_supplier (void);
private:
Quoter::Event data_;
RtecEventChannelAdmin::ProxyPushConsumer_var consumer_proxy_;
POA_RtecEventComm::PushSupplier_tie < Quoter_Stock_i > supplier_personality_;
};
</PRE>
<P>The implementation of the <CODE>set_price()</CODE> method is
very similar.
First we store the new price:
</P>
<PRE>
void
Quoter_Stock_i::set_price (CORBA::Double new_price)
{
this->data_.price = new_price;
</PRE>
<P>Next we prepare the event.
This time we must create a sequence, but we just have one
element in it:
</P>
<PRE>
RtecEventComm::EventSet event (1);
event.length (1);
</PRE>
<P>We set the event type based on the stock symbol:
</P>
<PRE>
RtecEventComm::Event &e = event[0];
const char *symbol = this->data_.symbol;
e.header.type =
((int(symbol[0]) << 24)
| (int(symbol[1]) << 16)
| (int(symbol[2]) << 8)
| int(symbol[3]));
e.header.source = 1;
</PRE>
<P>The event source is not used in this example, but it must be
non-zero. Now we can set the data:
</P>
<PRE>
e.data.any_value <<= this->data_;
</PRE>
<P>and send the event to the event channel:
</P>
<PRE>
this->consumer_proxy_->push (event);
}
</PRE>
<H3>Connecting to the Event Service as a Supplier</H3>
<P>As in the COS Event Channel case we need a supplier personality
to connect to it.
We gain access to the Event Service, for example using the
naming service:
</P>
<PRE>
CORBA::Object_var tmp = naming_context->resolve (name);
RtecEventChannelAdmin::EventChannel_var event_channel =
RtecEventChannelAdmin::EventChannel::_narrow (tmp);
</PRE>
<P>Now we use the event channel to obtain the factory used for
supplier connections:
</P>
<PRE>
RtecEventChannelAdmin::SupplierAdmin_var supplier_admin =
event_channel->for_suppliers ();
</PRE>
<P>And use the factory to obtain a proxy:
</P>
<PRE>
this->consumer_proxy_ =
supplier_admin->obtain_push_consumer ();
</PRE>
<P>We build our publications so the event channel can match
consumers and suppliers based on their common events:
</P>
<PRE>
const char *symbol = this->data_.symbol;
CORBA::ULong type =
((int(symbol[0]) << 24)
| (int(symbol[1]) << 16)
| (int(symbol[2]) << 8)
| int(symbol[3]));
CORBA::ULong source = 1;
ACE_SupplierQOS_Factory publications;
publications.insert_type (type, source, 0, 1);
</PRE>
<P>Finally we connect to the consumer proxy:
</P>
<PRE>
RtecEventComm::PushSupplier_var supplier =
this->supplier_personality_._this ();
this->consumer_proxy_->connect_push_supplier (supplier);
</PRE>
<P>The implementation of the disconnect callback is as before:
</P>
<PRE>
void
Quoter_Stock_i::disconnect_push_supplier (void)
{
// Forget about the consumer it is not there anymore
this->consumer_proxy_ =
RtecEventChannelAdmin::ProxyPushConsumer::_nil ();
}
</PRE>
<H3>Exercise 1</H3>
<P>Implement a consumer that receives the price update events,
</P>
<P>The
<A HREF="Stock_Consumer.h">header file</A>
is already provided,
along with a sample
<A HREF="client.cpp">client.cpp</A>.
And other support files
<A HREF="../Event_Service/Quoter.idl">Quoter.idl</A>,
<A HREF="GNUMakefile">Makefile</A>,
<A HREF="Stock_i.h">Stock_i.h</A>,
<A HREF="Stock_i.cpp">Stock_i.cpp</A>,
<A HREF="Stock_Factory_i.h">Stock_Factory_i.h</A>,
<A HREF="Stock_Factory_i.cpp">Stock_Factory_i.cpp</A>,
and <A HREF="server.cpp">server.cpp</A>.
</P>
<H4>Solution</H4>
<P>Compare your solution with
<A HREF="Stock_Consumer.cpp">Stock_Consumer.cpp</A>.
</P>
<H4>Testing</H4>
<P>To test your changes you need to run three programs,
first TAO's Naming Service:
<PRE>
$ $TAO_ROOT/orbsvcs/Naming_Service/Naming_Service
</PRE>
<P>then run TAO's Real-time Event Service
</P>
<PRE>
$ $TAO_ROOT/orbsvcs/Event_Service/Event_Service
</PRE>
<P>Now you can run your client:
<PRE>
$ client AAAA CCCC
</PRE>
<P>and finally the server:
</P>
<PRE>
$ server MSFT BBBB CCCC < stock_list.txt
</PRE>
<P>Here is the
<A HREF="../Event_Service/stock_list.txt">stock_list.txt file</A>.
</P>
<H3>Exercise 2</H3>
<P>Run the same configuration as above,
but this time run multiple clients and servers:
</P>
<PRE>
$ client AAAA MSFT
$ client PPPP
$ server AAAA < stock_list1.txt
$ server QQQQ < stock_list2.txt
</PRE>
<P>Do the clients receive all the events from both servers?
</P>
<P>Here are the
<A HREF="../Event_Service/stock_list1.txt">stock_list1.txt</A>
and
<A HREF="../Event_Service/stock_list2.txt">stock_list2.txt</A>
files.
</P>
<hr>
<address><a href="mailto:coryan@cs.wustl.edu">Carlos O'Ryan</a></address>
<!-- Created: Sat Nov 27 15:47:01 CST 1999 -->
<!-- hhmts start -->
Last modified: Wed May 16 10:13:34 PDT 2001
<!-- hhmts end -->
</body>
</html>
|
