00001
00002 #include "XmlRpcDispatch.h"
00003 #include "XmlRpcSource.h"
00004 #include "XmlRpcUtil.h"
00005
00006 #include <math.h>
00007 #include <sys/timeb.h>
00008
00009 #if defined(_WINDOWS)
00010 # include <winsock2.h>
00011
00012 # define USE_FTIME
00013 # if defined(_MSC_VER)
00014 # define timeb _timeb
00015 # define ftime _ftime
00016 # endif
00017 #else
00018 # include <sys/time.h>
00019 #endif // _WINDOWS
00020
00021 #include <iostream>
00022
00023 using namespace XmlRpc;
00024
00025
00026 XmlRpcDispatch::XmlRpcDispatch()
00027 {
00028 _endTime = -1.0;
00029 _doClear = false;
00030 _inWork = false;
00031 }
00032
00033
00034 XmlRpcDispatch::~XmlRpcDispatch()
00035 {
00036 }
00037
00038
00039
00040 void
00041 XmlRpcDispatch::addSource(XmlRpcSource* source, unsigned mask)
00042 {
00043 _sources.push_back(MonitoredSource(source, mask));
00044 }
00045
00046
00047 void
00048 XmlRpcDispatch::removeSource(XmlRpcSource* source)
00049 {
00050 for (SourceList::iterator it=_sources.begin(); it!=_sources.end(); ++it)
00051 if (it->getSource() == source)
00052 {
00053 _sources.erase(it);
00054 break;
00055 }
00056 }
00057
00058
00059
00060 void
00061 XmlRpcDispatch::setSourceEvents(XmlRpcSource* source, unsigned eventMask)
00062 {
00063 for (SourceList::iterator it=_sources.begin(); it!=_sources.end(); ++it)
00064 if (it->getSource() == source)
00065 {
00066 it->getMask() = eventMask;
00067 break;
00068 }
00069 }
00070
00071
00072
00073
00074 void
00075 XmlRpcDispatch::work(double timeout)
00076 {
00077
00078 _endTime = (timeout < 0.0) ? -1.0 : (getTime() + timeout);
00079 _doClear = false;
00080 _inWork = true;
00081
00082
00083 while (_sources.size() > 0) {
00084
00085
00086 fd_set inFd, outFd, excFd;
00087 FD_ZERO(&inFd);
00088 FD_ZERO(&outFd);
00089 FD_ZERO(&excFd);
00090
00091 int maxFd = -1;
00092 SourceList::iterator it;
00093 for (it=_sources.begin(); it!=_sources.end(); ++it) {
00094 int fd = it->getSource()->getfd();
00095 if (it->getMask() & ReadableEvent) FD_SET(fd, &inFd);
00096 if (it->getMask() & WritableEvent) FD_SET(fd, &outFd);
00097 if (it->getMask() & Exception) FD_SET(fd, &excFd);
00098 if (it->getMask() && fd > maxFd) maxFd = fd;
00099 }
00100
00101
00102 int nEvents;
00103 if (timeout < 0.0)
00104 nEvents = select(maxFd+1, &inFd, &outFd, &excFd, NULL);
00105 else
00106 {
00107 struct timeval tv;
00108 tv.tv_sec = (int)floor(timeout);
00109 tv.tv_usec = ((int)floor(1000000.0 * (timeout-floor(timeout)))) % 1000000;
00110 nEvents = select(maxFd+1, &inFd, &outFd, &excFd, &tv);
00111 }
00112
00113 if (nEvents < 0)
00114 {
00115 XmlRpcUtil::error("Error in XmlRpcDispatch::work: error in select (%d).", nEvents);
00116 _inWork = false;
00117 return;
00118 }
00119
00120
00121 for (it=_sources.begin(); it != _sources.end(); )
00122 {
00123 SourceList::iterator thisIt = it++;
00124 XmlRpcSource* src = thisIt->getSource();
00125 int fd = src->getfd();
00126 unsigned newMask = (unsigned) -1;
00127 if (fd <= maxFd) {
00128
00129 if (FD_ISSET(fd, &inFd))
00130 newMask &= src->handleEvent(ReadableEvent);
00131 if (FD_ISSET(fd, &outFd))
00132 newMask &= src->handleEvent(WritableEvent);
00133 if (FD_ISSET(fd, &excFd))
00134 newMask &= src->handleEvent(Exception);
00135
00136 if ( ! newMask) {
00137 _sources.erase(thisIt);
00138 if ( ! src->getKeepOpen())
00139 src->close();
00140 } else if (newMask != (unsigned) -1) {
00141 thisIt->getMask() = newMask;
00142 }
00143 }
00144 }
00145
00146
00147 if (_doClear)
00148 {
00149 SourceList closeList = _sources;
00150 _sources.clear();
00151 for (SourceList::iterator it=closeList.begin(); it!=closeList.end(); ++it) {
00152 XmlRpcSource *src = it->getSource();
00153 src->close();
00154 }
00155
00156 _doClear = false;
00157 }
00158
00159
00160 if (0 <= _endTime && getTime() > _endTime){
00161 std::cout<< "XmlRpc : time out when connect to database " << std::endl;
00162 break;
00163 }
00164 }
00165
00166 _inWork = false;
00167 }
00168
00169
00170
00171
00172 void
00173 XmlRpcDispatch::exit()
00174 {
00175 _endTime = 0.0;
00176 }
00177
00178
00179 void
00180 XmlRpcDispatch::clear()
00181 {
00182 if (_inWork)
00183 _doClear = true;
00184 else
00185 {
00186 SourceList closeList = _sources;
00187 _sources.clear();
00188 for (SourceList::iterator it=closeList.begin(); it!=closeList.end(); ++it)
00189 it->getSource()->close();
00190 }
00191 }
00192
00193
00194 double
00195 XmlRpcDispatch::getTime()
00196 {
00197 #ifdef USE_FTIME
00198 struct timeb tbuff;
00199
00200 ftime(&tbuff);
00201 return ((double) tbuff.time + ((double)tbuff.millitm / 1000.0) +
00202 ((double) tbuff.timezone * 60));
00203 #else
00204 struct timeval tv;
00205 struct timezone tz;
00206
00207 gettimeofday(&tv, &tz);
00208 return (tv.tv_sec + tv.tv_usec / 1000000.0);
00209 #endif
00210 }
00211
00212