XmlRpcDispatch.cpp
Go to the documentation of this file.
1 
4 #include "xmlrpcpp/XmlRpcUtil.h"
5 
6 #include "ros/time.h"
7 
8 #include <math.h>
9 #include <errno.h>
10 #include <sys/timeb.h>
11 #include <sys/poll.h>
12 
13 #if defined(_WINDOWS)
14 # include <winsock2.h>
15 static inline int poll( struct pollfd *pfd, int nfds, int timeout)
16 {
17  return WSAPoll(pfd, nfds, timeout);
18 }
19 
20 # define USE_FTIME
21 # if defined(_MSC_VER)
22 # define timeb _timeb
23 # define ftime _ftime_s
24 # endif
25 #else
26 # include <sys/time.h>
27 #endif // _WINDOWS
28 
29 
30 using namespace XmlRpc;
31 
32 
34 {
35  _endTime = -1.0;
36  _doClear = false;
37  _inWork = false;
38 }
39 
40 
42 {
43 }
44 
45 // Monitor this source for the specified events and call its event handler
46 // when the event occurs
47 void
48 XmlRpcDispatch::addSource(XmlRpcSource* source, unsigned mask)
49 {
50  _sources.push_back(MonitoredSource(source, mask));
51 }
52 
53 // Stop monitoring this source. Does not close the source.
54 void
56 {
57  for (SourceList::iterator it=_sources.begin(); it!=_sources.end(); ++it)
58  if (it->getSource() == source)
59  {
60  _sources.erase(it);
61  break;
62  }
63 }
64 
65 
66 // Modify the types of events to watch for on this source
67 void
68 XmlRpcDispatch::setSourceEvents(XmlRpcSource* source, unsigned eventMask)
69 {
70  for (SourceList::iterator it=_sources.begin(); it!=_sources.end(); ++it)
71  if (it->getSource() == source)
72  {
73  it->getMask() = eventMask;
74  break;
75  }
76 }
77 
78 
79 
80 // Watch current set of sources and process events
81 void
82 XmlRpcDispatch::work(double timeout)
83 {
84  // Loosely based on `man select` > Correspondence between select() and poll() notifications
85  // and cloudius-systems/osv#35, cloudius-systems/osv@b53d39a using poll to emulate select
86  const unsigned POLLIN_REQ = POLLIN; // Request read
87  const unsigned POLLIN_CHK = (POLLIN | POLLHUP | POLLERR); // Readable or connection lost
88  const unsigned POLLOUT_REQ = POLLOUT; // Request write
89  const unsigned POLLOUT_CHK = (POLLOUT | POLLERR); // Writable or connection lost
90  const unsigned POLLEX_CHK = (POLLPRI | POLLNVAL); // Exception or invalid fd
91 
92  // Compute end time
93  _endTime = (timeout < 0.0) ? -1.0 : (getTime() + timeout);
94  _doClear = false;
95  _inWork = true;
96  int timeout_ms = static_cast<int>(floor(timeout * 1000.));
97 
98  // Only work while there is something to monitor
99  while (_sources.size() > 0) {
100 
101  // Construct the sets of descriptors we are interested in
102  const unsigned source_cnt = _sources.size();
103  pollfd fds[source_cnt];
104  XmlRpcSource * sources[source_cnt];
105 
106  SourceList::iterator it;
107  std::size_t i = 0;
108  for (it=_sources.begin(); it!=_sources.end(); ++it, ++i) {
109  sources[i] = it->getSource();
110  fds[i].fd = sources[i]->getfd();
111  fds[i].revents = 0; // some platforms may not clear this in poll()
112  fds[i].events = 0;
113  if (it->getMask() & ReadableEvent) fds[i].events |= POLLIN_REQ;
114  if (it->getMask() & WritableEvent) fds[i].events |= POLLOUT_REQ;
115  }
116 
117  // Check for events
118  int nEvents = poll(fds, source_cnt, (timeout_ms < 0) ? -1 : timeout_ms);
119 
120  if (nEvents < 0)
121  {
122  if(errno != EINTR)
123  XmlRpcUtil::error("Error in XmlRpcDispatch::work: error in poll (%d).", nEvents);
124  _inWork = false;
125  return;
126  }
127 
128  // Process events
129  for (i=0; i < source_cnt; ++i)
130  {
131  XmlRpcSource* src = sources[i];
132  pollfd & pfd = fds[i];
133  unsigned newMask = (unsigned) -1;
134  // Only handle requested events to avoid being prematurely removed from dispatch
135  bool readable = (pfd.events & POLLIN_REQ) == POLLIN_REQ;
136  bool writable = (pfd.events & POLLOUT_REQ) == POLLOUT_REQ;
137  if (readable && (pfd.revents & POLLIN_CHK))
138  newMask &= src->handleEvent(ReadableEvent);
139  if (writable && (pfd.revents & POLLOUT_CHK))
140  newMask &= src->handleEvent(WritableEvent);
141  if (pfd.revents & POLLEX_CHK)
142  newMask &= src->handleEvent(Exception);
143 
144  // Find the source iterator. It may have moved as a result of the way
145  // that sources are removed and added in the call stack starting
146  // from the handleEvent() calls above.
147  SourceList::iterator thisIt;
148  for (thisIt = _sources.begin(); thisIt != _sources.end(); thisIt++)
149  {
150  if(thisIt->getSource() == src)
151  break;
152  }
153  if(thisIt == _sources.end())
154  {
155  XmlRpcUtil::error("Error in XmlRpcDispatch::work: couldn't find source iterator");
156  continue;
157  }
158 
159  if ( ! newMask) {
160  _sources.erase(thisIt); // Stop monitoring this one
161  if ( ! src->getKeepOpen())
162  src->close();
163  } else if (newMask != (unsigned) -1) {
164  thisIt->getMask() = newMask;
165  }
166  }
167 
168  // Check whether to clear all sources
169  if (_doClear)
170  {
171  SourceList closeList = _sources;
172  _sources.clear();
173  for (SourceList::iterator it=closeList.begin(); it!=closeList.end(); ++it) {
174  XmlRpcSource *src = it->getSource();
175  src->close();
176  }
177 
178  _doClear = false;
179  }
180 
181  // Check whether end time has passed
182  if (0 <= _endTime && getTime() > _endTime)
183  break;
184  }
185 
186  _inWork = false;
187 }
188 
189 
190 // Exit from work routine. Presumably this will be called from
191 // one of the source event handlers.
192 void
194 {
195  _endTime = 0.0; // Return from work asap
196 }
197 
198 // Clear all sources from the monitored sources list
199 void
201 {
202  if (_inWork)
203  _doClear = true; // Finish reporting current events before clearing
204  else
205  {
206  SourceList closeList = _sources;
207  _sources.clear();
208  for (SourceList::iterator it=closeList.begin(); it!=closeList.end(); ++it)
209  it->getSource()->close();
210  }
211 }
212 
213 
214 double
216 {
217 #ifdef USE_FTIME
218  struct timeb tbuff;
219 
220  ftime(&tbuff);
221  return ((double) tbuff.time + ((double)tbuff.millitm / 1000.0) +
222  ((double) tbuff.timezone * 60));
223 #else
224  uint32_t sec, nsec;
225 
226  ros::ros_steadytime(sec, nsec);
227  return ((double)sec + (double)nsec / 1e9);
228 #endif /* USE_FTIME */
229 }
230 
231 
Definition: XmlRpc.h:39
bool getKeepOpen() const
Return whether the file descriptor should be kept open if it is no longer monitored.
Definition: XmlRpcSource.h:32
connected/data can be written without blocking
ROSTIME_DECL void ros_steadytime(uint32_t &sec, uint32_t &nsec)
void removeSource(XmlRpcSource *source)
static void error(const char *fmt,...)
Dump error messages somewhere.
Definition: XmlRpcUtil.cpp:95
void clear()
Clear all sources from the monitored sources list. Sources are closed.
int getfd() const
Return the file descriptor being monitored.
Definition: XmlRpcSource.h:27
virtual void close()
Close the owned fd. If deleteOnClose was specified at construction, the object is deleted...
void work(double msTime)
XmlRpcDispatch()
Constructor.
An RPC source represents a file descriptor to monitor.
Definition: XmlRpcSource.h:16
void setSourceEvents(XmlRpcSource *source, unsigned eventMask)
Modify the types of events to watch for on this source.
void addSource(XmlRpcSource *source, unsigned eventMask)
std::list< MonitoredSource > SourceList
void exit()
Exit from work routine.
virtual unsigned handleEvent(unsigned eventType)=0
Return true to continue monitoring this source.


xmlrpcpp
Author(s): Chris Morley, Konstantin Pilipchuk, Morgan Quigley
autogenerated on Mon Jun 10 2019 14:48:08