mirror of
https://github.com/signalwire/freeswitch.git
synced 2026-07-22 12:11:50 +00:00
FS-2746 --resolve large xmlrpc update thanks garmt
This commit is contained in:
Jeff Lenk
@@ -9,24 +9,27 @@
|
||||
is an Xmlrpc-c invention. It is an almost trivial representation of
|
||||
a sequence of packets on a byte stream.
|
||||
|
||||
You can create a pstream server from any file descriptor from which
|
||||
you can read and write a bidirectional character stream. Typically,
|
||||
it's a TCP socket. Such a server talks to one client its entire life.
|
||||
|
||||
Some day, we'll also have a version that you create from a "listening"
|
||||
socket, which can talk to multiple clients serially (a client connects,
|
||||
does some RPCs, and disconnects).
|
||||
|
||||
By Bryan Henderson 07.05.12.
|
||||
By Bryan Henderson 09.03.22
|
||||
|
||||
Contributed to the public domain by its author.
|
||||
=============================================================================*/
|
||||
|
||||
#include "xmlrpc_config.h"
|
||||
#if MSVCRT
|
||||
#ifndef _CRT_SECURE_NO_WARNINGS
|
||||
#define _CRT_SECURE_NO_WARNINGS
|
||||
#endif
|
||||
#include <winsock.h>
|
||||
typedef int socklen_t;
|
||||
#else
|
||||
#include <sys/socket.h>
|
||||
#endif
|
||||
#include <errno.h>
|
||||
#include <cstring>
|
||||
#include <memory>
|
||||
|
||||
#include "xmlrpc-c/girerr.hpp"
|
||||
using girerr::throwf;
|
||||
#include "xmlrpc-c/packetsocket.hpp"
|
||||
|
||||
#include "xmlrpc-c/server_pstream.hpp"
|
||||
|
||||
@@ -35,20 +38,52 @@ using namespace std;
|
||||
namespace xmlrpc_c {
|
||||
|
||||
|
||||
serverPstreamConn::constrOpt::constrOpt() {
|
||||
struct serverPstream::constrOpt_impl {
|
||||
|
||||
present.socketFd = false;
|
||||
present.registryP = false;
|
||||
present.registryPtr = false;
|
||||
constrOpt_impl();
|
||||
|
||||
struct value {
|
||||
xmlrpc_c::registryPtr registryPtr;
|
||||
const xmlrpc_c::registry * registryP;
|
||||
XMLRPC_SOCKET socketFd;
|
||||
} value;
|
||||
struct {
|
||||
bool registryPtr;
|
||||
bool registryP;
|
||||
bool socketFd;
|
||||
} present;
|
||||
};
|
||||
|
||||
|
||||
|
||||
serverPstream::constrOpt_impl::constrOpt_impl() {
|
||||
|
||||
this->present.socketFd = false;
|
||||
this->present.registryP = false;
|
||||
this->present.registryPtr = false;
|
||||
}
|
||||
|
||||
|
||||
|
||||
serverPstream::constrOpt::constrOpt() {
|
||||
|
||||
this->implP = new serverPstream::constrOpt_impl();
|
||||
}
|
||||
|
||||
|
||||
|
||||
serverPstream::constrOpt::~constrOpt() {
|
||||
|
||||
delete(this->implP);
|
||||
}
|
||||
|
||||
|
||||
|
||||
#define DEFINE_OPTION_SETTER(OPTION_NAME, TYPE) \
|
||||
serverPstreamConn::constrOpt & \
|
||||
serverPstreamConn::constrOpt::OPTION_NAME(TYPE const& arg) { \
|
||||
this->value.OPTION_NAME = arg; \
|
||||
this->present.OPTION_NAME = true; \
|
||||
serverPstream::constrOpt & \
|
||||
serverPstream::constrOpt::OPTION_NAME(TYPE const& arg) { \
|
||||
this->implP->value.OPTION_NAME = arg; \
|
||||
this->implP->present.OPTION_NAME = true; \
|
||||
return *this; \
|
||||
}
|
||||
|
||||
@@ -60,8 +95,63 @@ DEFINE_OPTION_SETTER(registryPtr, xmlrpc_c::registryPtr);
|
||||
|
||||
|
||||
|
||||
struct serverPstream_impl {
|
||||
|
||||
serverPstream_impl(serverPstream::constrOpt_impl const& opt);
|
||||
|
||||
~serverPstream_impl();
|
||||
|
||||
void
|
||||
establishRegistry(serverPstream::constrOpt_impl const& opt);
|
||||
|
||||
// 'registryP' is what we actually use; 'registryHolder' just holds a
|
||||
// reference to 'registryP' so the registry doesn't disappear while
|
||||
// this server exists. But note that if the creator doesn't supply
|
||||
// a registryPtr, 'registryHolder' is just a placeholder variable and
|
||||
// the creator is responsible for making sure the registry doesn't
|
||||
// go anywhere while the server exists.
|
||||
|
||||
registryPtr registryHolder;
|
||||
const registry * registryP;
|
||||
|
||||
XMLRPC_SOCKET listenSocketFd;
|
||||
// The socket on which we accept connections from clients. This comes
|
||||
// to us from the creator, already bound and in listen mode. That
|
||||
// way, this object doesn't have to know anything about socket
|
||||
// addresses or listen parameters such as the maximum connection
|
||||
// backlog size.
|
||||
|
||||
bool termRequested;
|
||||
// User has requested that the run method return ASAP; i.e. that
|
||||
// the server cease servicing RPCs.
|
||||
};
|
||||
|
||||
|
||||
|
||||
serverPstream_impl::serverPstream_impl(
|
||||
serverPstream::constrOpt_impl const& opt) {
|
||||
|
||||
this->establishRegistry(opt);
|
||||
|
||||
if (!opt.present.socketFd)
|
||||
throwf("You must provide a 'socketFd' constructor option.");
|
||||
|
||||
this->listenSocketFd = opt.value.socketFd;
|
||||
|
||||
this->termRequested = false;
|
||||
}
|
||||
|
||||
|
||||
|
||||
serverPstream_impl::~serverPstream_impl() {
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
void
|
||||
serverPstreamConn::establishRegistry(constrOpt const& opt) {
|
||||
serverPstream_impl::establishRegistry(
|
||||
serverPstream::constrOpt_impl const& opt) {
|
||||
|
||||
if (!opt.present.registryP && !opt.present.registryPtr)
|
||||
throwf("You must specify the 'registryP' or 'registryPtr' option");
|
||||
@@ -79,93 +169,75 @@ serverPstreamConn::establishRegistry(constrOpt const& opt) {
|
||||
}
|
||||
|
||||
|
||||
|
||||
void
|
||||
serverPstreamConn::establishPacketSocket(constrOpt const& opt) {
|
||||
|
||||
if (!opt.present.socketFd)
|
||||
throwf("You must provide a 'socketFd' constructor option.");
|
||||
|
||||
auto_ptr<packetSocket> packetSocketAP;
|
||||
|
||||
try {
|
||||
auto_ptr<packetSocket> p(new packetSocket(opt.value.socketFd));
|
||||
packetSocketAP = p;
|
||||
} catch (exception const& e) {
|
||||
throwf("Unable to create packet socket out of file descriptor %d. %s",
|
||||
opt.value.socketFd, e.what());
|
||||
}
|
||||
this->packetSocketP = packetSocketAP.get();
|
||||
packetSocketAP.release();
|
||||
}
|
||||
|
||||
|
||||
|
||||
serverPstreamConn::serverPstreamConn(constrOpt const& opt) {
|
||||
|
||||
this->establishRegistry(opt);
|
||||
|
||||
this->establishPacketSocket(opt);
|
||||
}
|
||||
|
||||
|
||||
|
||||
serverPstreamConn::~serverPstreamConn() {
|
||||
|
||||
delete(this->packetSocketP);
|
||||
}
|
||||
|
||||
|
||||
|
||||
void
|
||||
processCall(const registry * const registryP,
|
||||
packetPtr const& callPacketP,
|
||||
packetPtr * const responsePacketPP) {
|
||||
|
||||
string const callXml(reinterpret_cast<char *>(callPacketP->getBytes()),
|
||||
callPacketP->getLength());
|
||||
|
||||
string responseXml;
|
||||
|
||||
registryP->processCall(callXml, &responseXml);
|
||||
|
||||
*responsePacketPP = packetPtr(new packet(responseXml.c_str(),
|
||||
responseXml.length()));
|
||||
}
|
||||
|
||||
|
||||
|
||||
void
|
||||
serverPstreamConn::runOnce(volatile const int * const interruptP,
|
||||
bool * const eofP) {
|
||||
/*----------------------------------------------------------------------------
|
||||
Get and execute one RPC from the client.
|
||||
|
||||
Unless *interruptP gets set nonzero first.
|
||||
/*-----------------------------------------------------------------------------
|
||||
serverPstream::shutdown is a derived class of registry::shutdown. You give
|
||||
it to the registry object to allow XML-RPC method 'system.shutdown' to
|
||||
-----------------------------------------------------------------------------*/
|
||||
bool gotPacket;
|
||||
packetPtr callPacketP;
|
||||
|
||||
try {
|
||||
this->packetSocketP->readWait(interruptP, eofP, &gotPacket,
|
||||
&callPacketP);
|
||||
} catch (exception const& e) {
|
||||
throwf("Error reading a packet from the packet socket. %s",
|
||||
e.what());
|
||||
}
|
||||
if (gotPacket) {
|
||||
packetPtr responsePacketP;
|
||||
try {
|
||||
processCall(this->registryP, callPacketP, &responsePacketP);
|
||||
} catch (exception const& e) {
|
||||
throwf("Error executing received packet as an XML-RPC RPC. %s",
|
||||
e.what());
|
||||
}
|
||||
try {
|
||||
this->packetSocketP->writeWait(responsePacketP);
|
||||
} catch (exception const& e) {
|
||||
throwf("Failed to write the response to the packet socket. %s",
|
||||
e.what());
|
||||
serverPstream::shutdown::shutdown(serverPstream * const serverPstreamP) :
|
||||
serverPstreamP(serverPstreamP) {}
|
||||
|
||||
|
||||
|
||||
serverPstream::shutdown::~shutdown() {}
|
||||
|
||||
|
||||
|
||||
void
|
||||
serverPstream::shutdown::doit(string const&,
|
||||
void * const) const {
|
||||
|
||||
this->serverPstreamP->terminate();
|
||||
}
|
||||
/*---------------------------------------------------------------------------*/
|
||||
|
||||
|
||||
|
||||
serverPstream::serverPstream(constrOpt const& opt) {
|
||||
|
||||
this->implP = new serverPstream_impl(*opt.implP);
|
||||
}
|
||||
|
||||
|
||||
|
||||
serverPstream::~serverPstream() {
|
||||
|
||||
delete(this->implP);
|
||||
}
|
||||
|
||||
|
||||
|
||||
void
|
||||
serverPstream::runSerial(volatile const int * const interruptP) {
|
||||
|
||||
while (!this->implP->termRequested && !*interruptP) {
|
||||
struct sockaddr peerAddr;
|
||||
socklen_t size = sizeof(peerAddr);
|
||||
int rc;
|
||||
|
||||
rc = accept(this->implP->listenSocketFd, &peerAddr, &size);
|
||||
|
||||
if (!*interruptP) {
|
||||
if (rc < 0)
|
||||
if (errno == EINTR) {
|
||||
// system call was interrupted, but user doesn't want
|
||||
// to interrupt the server, so just keep trying
|
||||
} else
|
||||
throwf("Failed to accept a connection "
|
||||
"on the listening socket. accept() failed "
|
||||
"with errno %d (%s)", errno, strerror(errno));
|
||||
else {
|
||||
int const acceptedFd = rc;
|
||||
|
||||
serverPstreamConn connectionServer(
|
||||
xmlrpc_c::serverPstreamConn::constrOpt()
|
||||
.socketFd(acceptedFd)
|
||||
.registryP(this->implP->registryP));
|
||||
|
||||
callInfo_serverPstream callInfo(this, peerAddr, size);
|
||||
|
||||
connectionServer.run(&callInfo, interruptP);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -173,15 +245,34 @@ serverPstreamConn::runOnce(volatile const int * const interruptP,
|
||||
|
||||
|
||||
void
|
||||
serverPstreamConn::runOnce(bool * const eofP) {
|
||||
/*----------------------------------------------------------------------------
|
||||
Get and execute one RPC from the client.
|
||||
-----------------------------------------------------------------------------*/
|
||||
serverPstream::runSerial() {
|
||||
|
||||
int const interrupt(0); // Never interrupt
|
||||
|
||||
this->runOnce(&interrupt, eofP);
|
||||
this->runSerial(&interrupt);
|
||||
}
|
||||
|
||||
|
||||
|
||||
void
|
||||
serverPstream::terminate() {
|
||||
|
||||
this->implP->termRequested = true;
|
||||
}
|
||||
|
||||
|
||||
|
||||
callInfo_serverPstream::callInfo_serverPstream(
|
||||
serverPstream * const serverP,
|
||||
struct sockaddr const clientAddr,
|
||||
socklen_t const clientAddrSize) :
|
||||
|
||||
serverP(serverP),
|
||||
clientAddr(clientAddr),
|
||||
clientAddrSize(clientAddrSize)
|
||||
|
||||
{}
|
||||
|
||||
|
||||
|
||||
} // namespace
|
||||
|
||||
Reference in New Issue
Block a user