diff --git a/apps/examples/xmlrpc2di/Readme.xmlrpc2di b/apps/examples/xmlrpc2di/Readme.xmlrpc2di index 7fa45ffd..ab162adb 100644 --- a/apps/examples/xmlrpc2di/Readme.xmlrpc2di +++ b/apps/examples/xmlrpc2di/Readme.xmlrpc2di @@ -5,6 +5,12 @@ by component modules accessible from XMLRPC. Additionaly the builtin methods calls, get_loglevel and set_loglevel are implemented (like in the stats UDP server). +Additionally, it can be used as client to access XMLRPC servers. Applications +can use the DI function newConnection to add a new server entry, and sendRequest +to send a request. If sendRequest is executed, an active server is selected from +the list and used to send the request. If sending the request failes, the server +is put inactive for a (configurable) while, and only then retried. + This module uses the XmlRpc++ library (http://xmlrpcpp.sourceforge.net/). In order to compile it, the xmlrpc library is needed; install it by make install-xmlrpcpp, or by hand: @@ -31,8 +37,55 @@ Configuration parameters direct_export none search these interfaces for methods to export -Exporting functions from a DI interface ---------------------------------------- + run_server yes start an XMLRPC server + + server_retry_after 10 retry a failed server after n seconds + +Using XMLRPC2DI client over DI +------------------------------ + +DI method "newConnection" + arguments: + "ssis": app_name, server, port, uri + example: + -- snip -- + AmDynInvoke* xmlrpc2di; + AmDynInvokeFactory* xfFactory = AmPlugIn::instance()->getFactory4Di("xmlrpc2di"); + if (NULL == xfFactory || + NULL == (xmlrpc2di = xfFactory->getInstance())){ + ERROR("could not get xmlrpc2di. please load the xmlrpc2di module\n"); + return; + } + + AmArg cargs, cret; + cargs.push("conf_auth"); + cargs.push("192.168.0.1"); + cargs.push(8102); + cargs.push(""); + xmlrpc2di->invoke("newConnection", cargs, cret); + -- snip -- + +DI method "sendRequest" + arguments: + "ssa": app_name, method, args + example: + -- snip -- + AmArg args, ret; + args.push("conf_auth"); + args.push("authPin"); // method name + args.assertArray(3); + args[2].push("fancy_parameter"); // some parameters + args[2].push("1234"); + xmlrpc2di->invoke("sendRequest", args, ret); + if (ret[0].asInt() == 0) { + DBG("status %d, description %s, uri %s", + ret[2][0].asInt(), ret[2][1].asCStr(), ret[2][3].asCStr()); + } + -- snip -- + + +Exporting functions from a DI interface to XMLRPC +------------------------------------------------- The xmlrpc2di module searches the interfaces configured by the 'direct_export' configuration variable for functions to export via XMLRPC. The interface must provide a function '_list' which diff --git a/apps/examples/xmlrpc2di/XMLRPC2DI.cpp b/apps/examples/xmlrpc2di/XMLRPC2DI.cpp index a7252898..2e0502da 100644 --- a/apps/examples/xmlrpc2di/XMLRPC2DI.cpp +++ b/apps/examples/xmlrpc2di/XMLRPC2DI.cpp @@ -38,17 +38,46 @@ #define XMLRPC_PORT "8090" // default port EXPORT_PLUGIN_CLASS_FACTORY(XMLRPC2DI, MOD_NAME) - XMLRPC2DI::XMLRPC2DI(string mod_name) - : AmDynInvokeFactory(mod_name) +XMLRPC2DI* XMLRPC2DI::_instance=0; + +// retry a failed server after 10 seconds +unsigned int XMLRPC2DI::ServerRetryAfter = 10; + +XMLRPC2DI* XMLRPC2DI::instance() +{ + if(_instance == NULL){ + _instance = new XMLRPC2DI(MOD_NAME); + } + return _instance; +} + +XMLRPC2DI::XMLRPC2DI(const string& mod_name) + : AmDynInvokeFactory(mod_name), configured(false) { } int XMLRPC2DI::onLoad() { + return instance()->load(); +} + +int XMLRPC2DI::load() { + if (configured) // load only once + return 0; + configured = true; AmConfigReader cfg; if(cfg.loadFile(AmConfig::ModConfigPath + string(MOD_NAME ".conf"))) return -1; + ServerRetryAfter = cfg.getParameterInt("server_retry_after", 10); + DBG("retrying failed server after %u seconds\n", ServerRetryAfter); + + string run_server = cfg.getParameter("run_server","yes"); + if (run_server != "yes") { + DBG("XMLRPC server will not be started.\n"); + return 0; + } + string conf_xmlrpc_port = cfg.getParameter("xmlrpc_port",XMLRPC_PORT); if (conf_xmlrpc_port.empty()) { ERROR("configuration: xmlrpc_port must be defined!\n"); @@ -83,6 +112,114 @@ int XMLRPC2DI::onLoad() { return 0; } +XMLRPCServerEntry::XMLRPCServerEntry(string s, int p, string u) + : last_try(0), active(true), server(s), port(p), uri(u) +{ } + +XMLRPCServerEntry::~XMLRPCServerEntry() +{ } + +bool XMLRPCServerEntry::is_active() { + if (!active && + ((unsigned int)(last_try + XMLRPC2DI::ServerRetryAfter) + < (unsigned int)time(NULL))) + active = true; + + return active; +} + +void XMLRPCServerEntry::set_failed() { + active = false; + time(&last_try); +} + +void XMLRPC2DI::newConnection(const AmArg& args, AmArg& ret) { + string app_name = args.get(0).asCStr(); + string server_name = args.get(1).asCStr(); + int port = args.get(2).asInt(); + string uri = args.get(3).asCStr(); + DBG("adding XMLRPC server http://%s:%d%s for application '%s'\n", + server_name.c_str(), port, uri.c_str(), app_name.c_str()); + + XMLRPCServerEntry* sc = new XMLRPCServerEntry(server_name, port, uri); + + server_mut.lock(); + servers.insert(make_pair(app_name, sc)); + server_mut.unlock(); +} + +XMLRPCServerEntry* XMLRPC2DI::getServer(const string& app_name) { + vector scs; + server_mut.lock(); + for (multimap::iterator it= + servers.lower_bound(app_name); + it != servers.upper_bound(app_name); it++) { + if (it->second->is_active()) + scs.push_back(it->second); + } + server_mut.unlock(); + + DBG("found %d active connections for application %s\n", + scs.size(), app_name.c_str()); + if (scs.empty()) { + // no connections found + return NULL; + } + + // select one connection randomly + return scs[random() % scs.size()]; +} + +void XMLRPC2DI::sendRequest(const AmArg& args, AmArg& ret) { + string app_name = args.get(0).asCStr(); + string method = args.get(1).asCStr(); + AmArg& params = args.get(2); + + while (true) { + XMLRPCServerEntry* srv = getServer(app_name); + if (NULL == srv) { + ret.push(-1); + ret.push("no active connections"); + return; + } + XmlRpcClient c(srv->server.c_str(), srv->port, + srv->uri.empty()?NULL:srv->uri.c_str()); + + XmlRpcValue x_args, x_result; + XMLRPC2DIServer::amarg2xmlrpcval(params, x_args); + if (c.execute(method.c_str(), x_args, x_result) && !c.isFault()) { + DBG("successfully executed method %s on server %s:%d\n", + method.c_str(), srv->server.c_str(), srv->port); + ret.push(0); + ret.push("OK"); + ret.assertArray(3); + XMLRPC2DIServer::xmlrpcval2amarg(x_result, ret[2]); + return; + } else { + DBG("executing method %s failed on server %s:%d\n", + method.c_str(), srv->server.c_str(), srv->port); + srv->set_failed(); + } + } +} + +void XMLRPC2DI::invoke(const string& method, + const AmArg& args, AmArg& ret) { + + if(method == "newConnection"){ + args.assertArrayFmt("ssis"); // app, server, port, uri + newConnection(args, ret); + } else if(method == "sendRequest"){ + args.assertArrayFmt("ssa"); // app, method, args + sendRequest(args, ret); + } else if(method == "_list"){ + ret.push(AmArg("newConnection")); + ret.push(AmArg("sendRequest")); + } else + throw AmDynInvoke::NotImplemented(method); + +} + // XMLRPC server functions XMLRPC2DIServer::XMLRPC2DIServer(unsigned int port, @@ -96,7 +233,7 @@ XMLRPC2DIServer::XMLRPC2DIServer(unsigned int port, // register method 'set_loglevel' getloglevel_method(&s) { - DBG(" XMLRPC Server: enabled builtin method 'calls'\n"); + DBG("XMLRPC Server: enabled builtin method 'calls'\n"); DBG("XMLRPC Server: enabled builtin method 'get_loglevel'\n"); DBG("XMLRPC Server: enabled builtin method 'set_loglevel'\n"); @@ -256,9 +393,16 @@ void XMLRPC2DIServer::xmlrpcval2amarg(XmlRpcValue& v, AmArg& a, if (v.valid()) { for (int i=start_index; iA INT\n");*/ a.push(AmArg((int)v[i])); } break; + case XmlRpcValue::TypeDouble:{ /* DBG("X->A DBL\n");*/ a.push(AmArg((double)v[i])); } break; + case XmlRpcValue::TypeString:{ /* DBG("X->A STR\n");*/ a.push(AmArg(((string)v[i]).c_str())); } break; + case XmlRpcValue::TypeArray: { + // DBG("X->A ARR\n"); + a.push(AmArg()); + a[a.size()-1].assertArray(0); + AmArg arr; + xmlrpcval2amarg(v[i], a[a.size()-1], 0); + } break; // TODO: support more types (datetime, struct, ...) default: throw XmlRpcException("unsupported parameter type", 400); }; @@ -266,25 +410,39 @@ void XMLRPC2DIServer::xmlrpcval2amarg(XmlRpcValue& v, AmArg& a, } } -void XMLRPC2DIServer::amarg2xmlrpcval(AmArg& a, +void XMLRPC2DIServer::amarg2xmlrpcval(const AmArg& a, XmlRpcValue& result) { switch (a.getType()) { case AmArg::CStr: + // DBG("a->X CSTR\n"); result = string(a.asCStr()); break; - case AmArg::Int: + case AmArg::Int: + // DBG("a->X INT\n"); result=a.asInt(); break; case AmArg::Double: + // DBG("a->X DOUBLE\n"); result=a.asDouble(); break; case AmArg::Array: + // DBG("a->X ARRAY size %u\n", a.size()); result.setSize(a.size()); for (size_t i=0;iX STRUCT size %u\n", a.size()); + for (AmArg::ValueStruct::const_iterator it = + a.begin(); it != a.end(); it++) { + // duh... recursion... + amarg2xmlrpcval(it->second, result[it->first]); + } + break; + default: { WARN("unsupported return value type %d\n", a.getType()); } break; // TODO: do sth with the data here ? } diff --git a/apps/examples/xmlrpc2di/XMLRPC2DI.h b/apps/examples/xmlrpc2di/XMLRPC2DI.h index e897440a..5e08d1fb 100644 --- a/apps/examples/xmlrpc2di/XMLRPC2DI.h +++ b/apps/examples/xmlrpc2di/XMLRPC2DI.h @@ -33,16 +33,24 @@ using namespace XmlRpc; #include "AmThread.h" #include "AmApi.h" -#define DEF_XMLRPCSERVERMETHOD(cls_name, func_name) \ -class cls_name \ - : public XmlRpcServerMethod { \ -\ -public: \ - cls_name(XmlRpcServer* s) : \ - XmlRpcServerMethod(func_name, s) { } \ -\ +#include +using std::map; + +#include +using std::string; + +#include + +#define DEF_XMLRPCSERVERMETHOD(cls_name, func_name) \ + class cls_name \ + : public XmlRpcServerMethod { \ + \ + public: \ + cls_name(XmlRpcServer* s) : \ + XmlRpcServerMethod(func_name, s) { } \ + \ void execute(XmlRpcValue& params, XmlRpcValue& result); \ -} + } DEF_XMLRPCSERVERMETHOD(XMLRPC2DIServerCallsMethod, "calls"); @@ -93,19 +101,57 @@ class XMLRPC2DIServer : public AmThread { unsigned int start_index = 0); /** convert all args in a into result*/ - static void amarg2xmlrpcval(AmArg& a, XmlRpcValue& result); + static void amarg2xmlrpcval(const AmArg& a, XmlRpcValue& result); }; +class XMLRPCServerEntry { + bool active; + time_t last_try; + + public: + XMLRPCServerEntry(string s, int p, string u); + ~XMLRPCServerEntry(); + + bool is_active(); + void set_failed(); + + string server; + int port; + string uri; +}; + +class XMLRPC2DI +: public AmDynInvokeFactory, + public AmDynInvoke { -class XMLRPC2DI : public AmDynInvokeFactory { XMLRPC2DIServer* server; unsigned int XMLRPCPort; + + static XMLRPC2DI* _instance; + bool configured; + int load(); + + // app server + multimap servers; + AmMutex server_mut; + XMLRPCServerEntry* getServer(const string& app_name); + + void newConnection(const AmArg& args, AmArg& ret); + void sendRequest(const AmArg& args, AmArg& ret); public: - XMLRPC2DI(string mod_name); + XMLRPC2DI(const string& mod_name); ~XMLRPC2DI() { } int onLoad(); - AmDynInvoke* getInstance() { return NULL; } + // DI factory + AmDynInvoke* getInstance() { return instance(); } + + // DI API + static XMLRPC2DI* instance(); + void invoke(const string& method, + const AmArg& args, AmArg& ret); + + static unsigned int ServerRetryAfter; }; #endif