blob: c9fc25077b6466b2536cdff23c00d89e674a3861 (
plain)
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
|
// $Id$
#include "orbsvcs/Notify/Structured/StructuredPushConsumer.h"
ACE_RCSID(RT_Notify, TAO_Notify_StructuredPushConsumer, "$Id$")
#include "ace/Bound_Ptr.h"
#include "tao/Stub.h" // For debug messages printing out ORBid.
#include "orbsvcs/Notify/Properties.h"
#include "orbsvcs/Notify/Event.h"
TAO_BEGIN_VERSIONED_NAMESPACE_DECL
TAO_Notify_StructuredPushConsumer::TAO_Notify_StructuredPushConsumer (TAO_Notify_ProxySupplier* proxy)
:TAO_Notify_Consumer (proxy)
{
}
TAO_Notify_StructuredPushConsumer::~TAO_Notify_StructuredPushConsumer ()
{
}
void
TAO_Notify_StructuredPushConsumer::init (CosNotifyComm::StructuredPushConsumer_ptr push_consumer ACE_ENV_ARG_DECL)
{
// Initialize only once
ACE_ASSERT( CORBA::is_nil (this->push_consumer_.in()) );
if (CORBA::is_nil (push_consumer))
{
ACE_THROW (CORBA::BAD_PARAM());
}
if (!TAO_Notify_PROPERTIES::instance()->separate_dispatching_orb ())
{
this->push_consumer_ = CosNotifyComm::StructuredPushConsumer::_duplicate (push_consumer);
this->publish_ = CosNotifyComm::NotifyPublish::_duplicate (push_consumer);
}
else
{
// "Port" consumer's object reference from receiving ORB to dispatching ORB.
CORBA::String_var temp =
TAO_Notify_PROPERTIES::instance()->orb()->object_to_string(push_consumer);
CORBA::Object_var obj =
TAO_Notify_PROPERTIES::instance()->dispatching_orb()->string_to_object(temp.in());
ACE_TRY
{
CosNotifyComm::StructuredPushConsumer_var new_push_consumer =
CosNotifyComm::StructuredPushConsumer::_unchecked_narrow(obj.in() ACE_ENV_ARG_PARAMETER);
ACE_TRY_CHECK;
this->push_consumer_ = CosNotifyComm::StructuredPushConsumer::_duplicate (new_push_consumer.in());
this->publish_ = CosNotifyComm::NotifyPublish::_duplicate (new_push_consumer.in());
//--cj verify dispatching ORB
if (TAO_debug_level >= 10)
{
ACE_DEBUG ((LM_DEBUG,
"(%P|%t) Structured push init dispatching ORB id is %s.\n",
obj->_stubobj()->orb_core()->orbid()));
}
//--cj end
}
ACE_CATCH (CORBA::TRANSIENT, ex)
{
ACE_PRINT_EXCEPTION (ex, "Got a TRANSIENT in NS_StructuredPushConsumer::init");
ACE_DEBUG ((LM_DEBUG, "(%P|%t) got it for NS_StructuredPushConsumer %@\n", this));
}
ACE_CATCHANY
{
// _narrow failed
}
ACE_ENDTRY;
}
}
void
TAO_Notify_StructuredPushConsumer::release (void)
{
delete this;
//@@ inform factory
}
void
TAO_Notify_StructuredPushConsumer::push (const CORBA::Any& event ACE_ENV_ARG_DECL)
{
CosNotification::StructuredEvent notification;
TAO_Notify_Event::translate (event, notification);
this->push_consumer_->push_structured_event (notification ACE_ENV_ARG_PARAMETER);
}
void
TAO_Notify_StructuredPushConsumer::push (const CosNotification::StructuredEvent& event ACE_ENV_ARG_DECL)
{
//--cj verify dispatching ORB
if (TAO_debug_level >= 10) {
ACE_DEBUG ((LM_DEBUG, "(%P|%t) Structured push dispatching ORB id is %s.\n",
this->push_consumer_->_stubobj()->orb_core()->orbid()));
}
//--cj end
this->push_consumer_->push_structured_event (event ACE_ENV_ARG_PARAMETER);
ACE_CHECK;
}
/// Push a batch of events to this consumer.
void
TAO_Notify_StructuredPushConsumer::push (const CosNotification::EventBatch& event ACE_ENV_ARG_DECL_NOT_USED)
{
ACE_ASSERT(false);
ACE_UNUSED_ARG (event);
// TODO exception?
}
void
TAO_Notify_StructuredPushConsumer::reconnect_from_consumer (TAO_Notify_Consumer* old_consumer
ACE_ENV_ARG_DECL)
{
TAO_Notify_StructuredPushConsumer* tmp = dynamic_cast<TAO_Notify_StructuredPushConsumer *> (old_consumer);
ACE_ASSERT(tmp != 0);
this->init(tmp->push_consumer_.in() ACE_ENV_ARG_PARAMETER);
ACE_CHECK;
this->schedule_timer(false);
}
ACE_CString
TAO_Notify_StructuredPushConsumer::get_ior (void) const
{
ACE_CString result;
CORBA::ORB_var orb = TAO_Notify_PROPERTIES::instance()->orb();
ACE_DECLARE_NEW_CORBA_ENV;
ACE_TRY
{
CORBA::String_var ior = orb->object_to_string(this->push_consumer_.in() ACE_ENV_ARG_PARAMETER);
ACE_TRY_CHECK;
result = static_cast<const char*> (ior.in ());
}
ACE_CATCHANY
{
result.fast_clear();
}
ACE_ENDTRY;
return result;
}
TAO_END_VERSIONED_NAMESPACE_DECL
|
