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
|
From: =?utf-8?q?Picca_Fr=C3=A9d=C3=A9ric-Emmanuel?= <picca@debian.org>
Date: Mon, 25 Nov 2013 20:41:39 +0100
Subject: fix upstream p812_2
---
lib/cpp/client/api_util.cpp | 4 ++++
lib/cpp/client/event.cpp | 11 ++++++-----
lib/cpp/client/eventconsumer.h | 1 +
lib/cpp/client/eventkeepalive.cpp | 9 +++++++++
lib/cpp/client/zmqeventconsumer.cpp | 21 ++++++++++++++++-----
5 files changed, 36 insertions(+), 10 deletions(-)
diff --git a/lib/cpp/client/api_util.cpp b/lib/cpp/client/api_util.cpp
index 6e38825..3ed6d83 100644
--- a/lib/cpp/client/api_util.cpp
+++ b/lib/cpp/client/api_util.cpp
@@ -257,6 +257,10 @@ void ApiUtil::create_orb()
if (sigaction(SIGPIPE,NULL,&sa) == -1)
sa.sa_handler = NULL;
+#else
+ WORD rel = 0x0202;
+ WSADATA dat;
+ WSAStartup(rel,&dat);
#endif
//
diff --git a/lib/cpp/client/event.cpp b/lib/cpp/client/event.cpp
index 75b8c33..f3fecf1 100644
--- a/lib/cpp/client/event.cpp
+++ b/lib/cpp/client/event.cpp
@@ -202,12 +202,12 @@ EventConsumer::EventConsumer(ApiUtil *api_ptr)
{
break;
}
+ }
- if (j == env_var_fqdn_prefix.size())
- {
- string prefix = "tango://" + vs[i] + '/';
- env_var_fqdn_prefix.push_back(prefix);
- }
+ if (j == env_var_fqdn_prefix.size())
+ {
+ string prefix = "tango://" + vs[i] + '/';
+ env_var_fqdn_prefix.push_back(prefix);
}
#endif
}
@@ -1441,6 +1441,7 @@ int EventConsumer::connect_event(DeviceProxy *device,
new_event_callback.device_idl = 0;
}
new_event_callback.ctr = 0;
+ new_event_callback.discarded_event = false;
if (zmq_used == true)
new_event_callback.endpoint = dvlsa->svalue[1].in();
diff --git a/lib/cpp/client/eventconsumer.h b/lib/cpp/client/eventconsumer.h
index f403fb2..20c8604 100644
--- a/lib/cpp/client/eventconsumer.h
+++ b/lib/cpp/client/eventconsumer.h
@@ -298,6 +298,7 @@ typedef struct event_callback_zmq
DevLong device_idl;
DevULong ctr;
string endpoint;
+ bool discarded_event;
}EventCallBackZmq;
typedef struct event_callback: public EventCallBackBase, public EventCallBackZmq
diff --git a/lib/cpp/client/eventkeepalive.cpp b/lib/cpp/client/eventkeepalive.cpp
index e204fae..2d93241 100644
--- a/lib/cpp/client/eventkeepalive.cpp
+++ b/lib/cpp/client/eventkeepalive.cpp
@@ -452,6 +452,10 @@ void EventConsumerKeepAliveThread::reconnect_to_zmq_event(EvChanIte &ipos,EventC
event_consumer->connect_event_system(d_name,epos->second.attr_name,epos->second.event_name,vs,ipos,ecbs,dd);
+ const DevVarLongStringArray *dvlsa;
+ dd >> dvlsa;
+ epos->second.endpoint = dvlsa->svalue[1].in();
+
cout3 << "Reconnected to ZMQ event" << endl;
}
catch(...)
@@ -588,6 +592,11 @@ void *EventConsumerKeepAliveThread::run_undetached(TANGO_UNUSED(void *arg))
{
try
{
+ if (notifd_event_consumer == NULL)
+ {
+ ApiUtil::instance()->create_notifd_event_consumer();
+ notifd_event_consumer = ApiUtil::instance()->get_notifd_event_consumer();
+ }
notifd_event_consumer->connect_event(vpos->device,vpos->attribute,vpos->event_type,
vpos->callback,
vpos->ev_queue,
diff --git a/lib/cpp/client/zmqeventconsumer.cpp b/lib/cpp/client/zmqeventconsumer.cpp
index 8fa8ce0..6ebdfc8 100644
--- a/lib/cpp/client/zmqeventconsumer.cpp
+++ b/lib/cpp/client/zmqeventconsumer.cpp
@@ -859,6 +859,7 @@ bool ZmqEventConsumer::process_ctrl(zmq::message_t &received_ctrl,zmq::pollitem_
event_sub_sock->setsockopt(ZMQ_RCVHWM,&sub_hwm,sizeof(sub_hwm));
event_sub_sock->connect(endpoint);
+
if (force_connect == 0)
connected_pub.push_back(endpoint);
}
@@ -935,9 +936,10 @@ bool ZmqEventConsumer::process_ctrl(zmq::message_t &received_ctrl,zmq::pollitem_
vector<string>::iterator pos;
pos = find(connected_pub.begin(),connected_pub.end(),endpoint_str);
if (pos != connected_pub.end())
+ {
connected_pub.erase(pos);
-
- event_sub_sock->disconnect(endpoint);
+ event_sub_sock->disconnect(endpoint);
+ }
#endif
}
else
@@ -1930,12 +1932,21 @@ void ZmqEventConsumer::push_zmq_event(string &ev_name,unsigned char endian,zmq::
if (missed_event >= 2)
{
err_missed_event = true;
+ evt_cb.discarded_event = false;
}
else if (missed_event == 0)
{
- map_modification_lock.readerOut();
- return;
- }
+ if (evt_cb.discarded_event == false)
+ {
+ evt_cb.discarded_event = true;
+ map_modification_lock.readerOut();
+ return;
+ }
+ else
+ evt_cb.discarded_event = false;
+ }
+ else
+ evt_cb.discarded_event = false;
evt_cb.ctr = ds_ctr;
|