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
|
// $Id$
#include "Supplier.h"
ACE_RCSID(Notify, TAO_NS_Supplier, "$id$")
#include "tao/ORB_Core.h"
TAO_NS_Supplier::TAO_NS_Supplier (TAO_NS_ORB_Objects& orb_objects)
: orb_objects_ (orb_objects)
, proxy_consumer_id_ (0)
, expected_consumer_count_ (2)
, consumers_connected_ (lock_)
, consumer_count_ (0)
{
}
TAO_NS_Supplier::~TAO_NS_Supplier ()
{
}
void
TAO_NS_Supplier::init (CosNotifyChannelAdmin::SupplierAdmin_var& admin, int expected_consumer_count ACE_ENV_ARG_DECL)
{
// First initialize the class members.
this->admin_ = admin;
this->expected_consumer_count_ = expected_consumer_count;
this->connect (ACE_ENV_SINGLE_ARG_PARAMETER);
}
void
TAO_NS_Supplier::run (ACE_ENV_SINGLE_ARG_DECL)
{
// The Priority at which we send the first event to the first consumer.
RTCORBA::Priority priority = 1;
{
ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->lock_);
ACE_DEBUG ((LM_DEBUG, "(%P, %t) Waiting for %d consumers to connect...\n", this->expected_consumer_count_));
// Wait till the consumers are ready to go.
while (this->consumer_count_ != this->expected_consumer_count_)
this->consumers_connected_.wait ();
}
// Send an event each to each consumer.
// Each Consumer expects a different priority.
for (int i = 0; i < this->expected_consumer_count_; ++i, ++priority)
{
// Set this threads priority.
this->orb_objects_.current_->the_priority (priority ACE_ENV_ARG_PARAMETER);
ACE_CHECK;
// Make sure the priority was set, get the priority of the current thread.
RTCORBA::Priority thread_priority =
this->orb_objects_.current_->the_priority (ACE_ENV_SINGLE_ARG_PARAMETER);
ACE_CHECK;
// We will send this event.
CosNotification::StructuredEvent event;
// Populate the Priority field so that the consumer can deduce the suppliers priority
// to do a sanity check when it receives the event.
CosNotification::PropertySeq& opt = event.header.variable_header;
opt.length (1);
CORBA::Any buffer;
buffer <<= (CORBA::Short) thread_priority;
opt[0].name = CORBA::string_dup (CosNotification::Priority);
opt[0].value = buffer;
// Set the domain and type nams in the event's fixed header.
char type[BUFSIZ];
ACE_OS::sprintf (type, "TEST_TYPE_%d", thread_priority);
event.header.fixed_header.event_type.domain_name = CORBA::string_dup("TEST_DOMAIN");
event.header.fixed_header.event_type.type_name = CORBA::string_dup(type);
ACE_DEBUG ((LM_DEBUG,
"(%P, %t) Supplier is sending an event of type %s at priority %d\n", type, thread_priority));
// send the event
this->send_event (event ACE_ENV_ARG_PARAMETER);
} // repeat for the next consumer at the next priority.
// Disconnect from the EC
this->disconnect (ACE_ENV_SINGLE_ARG_PARAMETER);
// Deactivate this object.
this->deactivate (ACE_ENV_SINGLE_ARG_PARAMETER);
// we're done. shutdown the ORB to exit the process.
this->orb_objects_.orb_->shutdown (1);
}
void
TAO_NS_Supplier::connect (ACE_ENV_SINGLE_ARG_DECL)
{
// Activate the supplier object.
CosNotifyComm::StructuredPushSupplier_var objref = this->_this (ACE_ENV_SINGLE_ARG_PARAMETER);
ACE_CHECK;
// Obtain the proxy.
CosNotifyChannelAdmin::ProxyConsumer_var proxyconsumer =
this->admin_->obtain_notification_push_consumer (CosNotifyChannelAdmin::STRUCTURED_EVENT
, proxy_consumer_id_ ACE_ENV_ARG_PARAMETER);
ACE_CHECK;
ACE_ASSERT (!CORBA::is_nil (proxyconsumer.in ()));
// narrow
this->proxy_consumer_ =
CosNotifyChannelAdmin::StructuredProxyPushConsumer::_narrow (proxyconsumer.in () ACE_ENV_ARG_PARAMETER);
ACE_CHECK;
ACE_ASSERT (!CORBA::is_nil (proxy_consumer_.in ()));
// connect to the proxyconsumer.
proxy_consumer_->connect_structured_push_supplier (objref.in ()
ACE_ENV_ARG_PARAMETER);
ACE_CHECK;
}
void
TAO_NS_Supplier::disconnect (ACE_ENV_SINGLE_ARG_DECL)
{
ACE_ASSERT (!CORBA::is_nil (this->proxy_consumer_.in ()));
this->proxy_consumer_->disconnect_structured_push_consumer(ACE_ENV_SINGLE_ARG_PARAMETER);
}
void
TAO_NS_Supplier::deactivate (ACE_ENV_SINGLE_ARG_DECL)
{
PortableServer::POA_var poa (this->_default_POA (ACE_ENV_SINGLE_ARG_PARAMETER));
ACE_CHECK;
PortableServer::ObjectId_var id (poa->servant_to_id (this
ACE_ENV_ARG_PARAMETER));
ACE_CHECK;
poa->deactivate_object (id.in()
ACE_ENV_ARG_PARAMETER);
ACE_CHECK;
}
void
TAO_NS_Supplier::subscription_change (const CosNotification::EventTypeSeq & added,
const CosNotification::EventTypeSeq & /*removed */
ACE_ENV_ARG_DECL_NOT_USED)
ACE_THROW_SPEC ((
CORBA::SystemException,
CosNotifyComm::InvalidEventType
))
{
ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->lock_);
// Count the number of consumers connect and signal the supplier thread when the expected count have connected.
if (added.length () > 0)
{
if (++this->consumer_count_ == this->expected_consumer_count_)
this->consumers_connected_.signal ();
}
}
void
TAO_NS_Supplier::send_event (const CosNotification::StructuredEvent& event ACE_ENV_ARG_DECL)
{
ACE_ASSERT (!CORBA::is_nil (this->proxy_consumer_.in ()));
proxy_consumer_->push_structured_event (event ACE_ENV_ARG_PARAMETER);
ACE_CHECK;
}
void
TAO_NS_Supplier::disconnect_structured_push_supplier (ACE_ENV_SINGLE_ARG_DECL)
ACE_THROW_SPEC ((
CORBA::SystemException
))
{
this->deactivate (ACE_ENV_SINGLE_ARG_PARAMETER);
}
|
