mirror of https://github.com/sipwise/sems.git
git-svn-id: http://svn.berlios.de/svnroot/repos/sems/trunk@1102 8eb893ce-cfd4-0310-b710-fb5ebe64c474sayer/1.4-spce2.6
Stefan Sayer
18 years ago
parent
36e26bb406
commit
3ec2c179dc
@ -0,0 +1,122 @@
|
||||
/*
|
||||
* Written by Jae An (jae_young_an@yahoo.com)
|
||||
*
|
||||
* adapted for SEMS by Stefan Sayer stefan.sayer at iptego.com
|
||||
*/
|
||||
#include "XmlRpcServer.h"
|
||||
#include "XmlRpcServerConnection.h"
|
||||
#include "XmlRpcServerMethod.h"
|
||||
#include "XmlRpcSocket.h"
|
||||
#include "XmlRpcUtil.h"
|
||||
#include "XmlRpcException.h"
|
||||
#include "MultithreadXmlRpcServer.h"
|
||||
|
||||
#include "log.h"
|
||||
|
||||
using namespace XmlRpc;
|
||||
WorkerThread::WorkerThread(MultithreadXmlRpcServer* chief)
|
||||
: runcond(false), chief(chief) {
|
||||
}
|
||||
|
||||
// call this method before calling run
|
||||
void WorkerThread::addXmlRpcSource(XmlRpcSource* source,unsigned eventMask)
|
||||
{
|
||||
dispatcher.addSource(source,eventMask);
|
||||
runcond.set(true);
|
||||
}
|
||||
|
||||
void WorkerThread::run()
|
||||
{
|
||||
chief->reportBack(this);
|
||||
|
||||
while (!is_stopped()) {
|
||||
runcond.wait_for();
|
||||
dispatcher.work(-1.0);
|
||||
|
||||
dispatcher.clear(); // close socket and others ...
|
||||
runcond.set(false);
|
||||
/* tell chief we can work again */
|
||||
chief->reportBack(this);
|
||||
}
|
||||
DBG("WorkerThread stopped.\n");
|
||||
}
|
||||
|
||||
void WorkerThread::on_stop() {
|
||||
}
|
||||
|
||||
MultithreadXmlRpcServer::MultithreadXmlRpcServer()
|
||||
: XmlRpcServer(), have_waiting(false) {
|
||||
}
|
||||
|
||||
// Wait until all the worker threads are done.
|
||||
MultithreadXmlRpcServer::~MultithreadXmlRpcServer()
|
||||
{
|
||||
for (std::vector<WorkerThread*>::iterator it=
|
||||
workers.begin(); it != workers.end();it++) {
|
||||
(*it)->stop();
|
||||
(*it)->join();
|
||||
delete *it;
|
||||
}
|
||||
}
|
||||
|
||||
// Accept a client connection request and create a connection to
|
||||
// handle method calls from the client.
|
||||
void MultithreadXmlRpcServer::acceptConnection()
|
||||
{
|
||||
int s = XmlRpcSocket::accept(this->getfd());
|
||||
if (s < 0)
|
||||
{
|
||||
ERROR("MultithreadXmlRpcServer::acceptConnection: Could not accept connection (%s).",
|
||||
XmlRpcSocket::getErrorMsg().c_str());
|
||||
|
||||
}
|
||||
else if ( ! XmlRpcSocket::setNonBlocking(s))
|
||||
{
|
||||
XmlRpcSocket::close(s);
|
||||
ERROR("XmlRpcServer::acceptConnection: Could not set socket "
|
||||
"to non-blocking input mode (%s).\n",
|
||||
XmlRpcSocket::getErrorMsg().c_str());
|
||||
}
|
||||
else // Notify the dispatcher to listen for input on this source when we are in work()
|
||||
{
|
||||
WorkerThread* thr = NULL;
|
||||
while (!thr) {
|
||||
if (!have_waiting.get())
|
||||
have_waiting.wait_for();
|
||||
thr = getIdleThread();
|
||||
}
|
||||
thr->addXmlRpcSource(this->createConnection(s), XmlRpcDispatch::ReadableEvent);
|
||||
|
||||
// // Uh oh.. all threads are busy...
|
||||
// // Just close the connection so that the rejected client would try again.
|
||||
// XmlRpcSocket::close(s);
|
||||
// XmlRpcUtil::error("MultithreadXmlRpcServer::acceptConnection: All threads are busy. Rejected a client");
|
||||
}
|
||||
}
|
||||
|
||||
void MultithreadXmlRpcServer::reportBack(WorkerThread* thr) {
|
||||
waiting_mut.lock();
|
||||
waiting.push(thr);
|
||||
have_waiting.set(true);
|
||||
waiting_mut.unlock();
|
||||
}
|
||||
|
||||
WorkerThread* MultithreadXmlRpcServer::getIdleThread() {
|
||||
WorkerThread* res = NULL;
|
||||
waiting_mut.lock();
|
||||
if (!waiting.empty()) {
|
||||
res = waiting.front();
|
||||
waiting.pop();
|
||||
}
|
||||
have_waiting.set(!waiting.empty());
|
||||
waiting_mut.unlock();
|
||||
return res;
|
||||
}
|
||||
|
||||
void MultithreadXmlRpcServer::createThreads(unsigned int n) {
|
||||
for (unsigned int i=0;i<n;i++) {
|
||||
WorkerThread* thr = new WorkerThread(this);
|
||||
workers.push_back(thr);
|
||||
thr->start();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,66 @@
|
||||
/**
|
||||
* Written by Jae An (jae_young_an@yahoo.com)
|
||||
* http://sourceforge.net/forum/forum.php?thread_id=910224&forum_id=240495
|
||||
*
|
||||
* adapted for SEMS by Stefan Sayer stefan.sayer at iptego.com
|
||||
*/
|
||||
|
||||
#ifndef _MULTI_THREAD_XMLRPCSERVER_H_
|
||||
#define _MULTI_THREAD_XMLRPCSERVER_H_
|
||||
|
||||
#include <AmThread.h>
|
||||
#include "XmlRpcServer.h"
|
||||
#include "XmlRpcDispatch.h"
|
||||
|
||||
#include <queue>
|
||||
#include <vector>
|
||||
|
||||
namespace XmlRpc {
|
||||
class MultithreadXmlRpcServer;
|
||||
|
||||
class WorkerThread : public AmThread {
|
||||
MultithreadXmlRpcServer* chief;
|
||||
AmCondition<bool> runcond;
|
||||
|
||||
public:
|
||||
WorkerThread(MultithreadXmlRpcServer* chief);
|
||||
|
||||
void addXmlRpcSource(XmlRpcSource* source,unsigned eventMask); // call this method to make it run
|
||||
|
||||
void run();
|
||||
void on_stop();
|
||||
private:
|
||||
XmlRpcDispatch dispatcher;
|
||||
};
|
||||
|
||||
|
||||
#define MAX_THREAD_SIZE 8
|
||||
|
||||
//! Multi-threaded sever class to handle XML RPC requests
|
||||
class MultithreadXmlRpcServer : public XmlRpcServer {
|
||||
public:
|
||||
//! Create a server object.
|
||||
MultithreadXmlRpcServer();
|
||||
//! Destructor.
|
||||
virtual ~MultithreadXmlRpcServer();
|
||||
|
||||
/* report back from work */
|
||||
void reportBack(WorkerThread* thr);
|
||||
void createThreads(unsigned int n);
|
||||
|
||||
protected:
|
||||
|
||||
//! Accept a client connection request
|
||||
virtual void acceptConnection();
|
||||
|
||||
private:
|
||||
AmMutex waiting_mut;
|
||||
std::queue<WorkerThread*> waiting;
|
||||
AmCondition<bool> have_waiting;
|
||||
std::vector<WorkerThread*> workers;
|
||||
WorkerThread* getIdleThread();
|
||||
};
|
||||
|
||||
} // namespace XmlRpc
|
||||
|
||||
#endif
|
||||
Loading…
Reference in new issue