From edd2fa1cc0a0dfcb2088cfc72d183c81d17b8c7c Mon Sep 17 00:00:00 2001 From: Stefan Sayer Date: Mon, 12 Jul 2010 11:46:25 +0200 Subject: [PATCH] jsonrpc: support for per request user data Due to the multiplexing of requests from several sources on to json-rpc connections, when Jsonrpc requests are sent, the application can not know the request ID which is to be used. A new AmArg parameter to jsonrpc sendMessage and execRpc functions can provide user data, which is copied into the response event, and thus can be matched to the request in the application. --- apps/dsm/DSMCall.cpp | 11 ++++++++--- apps/dsm/DSMSession.h | 3 ++- apps/dsm/SystemDSM.cpp | 9 +++++++-- apps/jsonrpc/JsonRPC.cpp | 16 +++++++++++++--- apps/jsonrpc/JsonRPCEvents.h | 15 +++++++++------ apps/jsonrpc/JsonRPCServer.cpp | 17 +++++++++++------ apps/jsonrpc/JsonRPCServer.h | 3 ++- apps/jsonrpc/RpcPeer.cpp | 4 ++-- apps/jsonrpc/RpcPeer.h | 4 +++- apps/jsonrpc/RpcServerLoop.cpp | 12 ++++++------ apps/jsonrpc/RpcServerLoop.h | 6 ++++-- apps/jsonrpc/RpcServerThread.cpp | 3 ++- 12 files changed, 69 insertions(+), 34 deletions(-) diff --git a/apps/dsm/DSMCall.cpp b/apps/dsm/DSMCall.cpp index 14e4b76c..16dcb3b7 100644 --- a/apps/dsm/DSMCall.cpp +++ b/apps/dsm/DSMCall.cpp @@ -385,12 +385,17 @@ void DSMCall::process(AmEvent* event) // decode result for easy use from script varPrintArg(resp_ev->response.data, params, resp_ev->response.is_error ? "error": "result"); - // save reference to full parameters - avar[DSM_AVAR_JSONRPCRESPONEDATA] = AmArg(&resp_ev->response.data); + // decode udata for easy use from script + varPrintArg(resp_ev->udata, params, "udata"); + + // save reference to full parameters as avar + avar[DSM_AVAR_JSONRPCRESPONSEDATA] = AmArg(&resp_ev->response.data); + avar[DSM_AVAR_JSONRPCRESPONSEUDATA] = AmArg(&resp_ev->udata); engine.runEvent(this, this, DSMCondition::JsonRpcResponse, ¶ms); - avar.erase(DSM_AVAR_JSONRPCRESPONEDATA); + avar.erase(DSM_AVAR_JSONRPCRESPONSEUDATA); + avar.erase(DSM_AVAR_JSONRPCRESPONSEDATA); return; } diff --git a/apps/dsm/DSMSession.h b/apps/dsm/DSMSession.h index f396461d..65c3ee9c 100644 --- a/apps/dsm/DSMSession.h +++ b/apps/dsm/DSMSession.h @@ -63,7 +63,8 @@ using std::map; #define DSM_AVAR_REPLY "reply" #define DSM_AVAR_JSONRPCREQUESTDATA "JsonRpcRequestParameters" -#define DSM_AVAR_JSONRPCRESPONEDATA "JsonRpcResponseParameters" +#define DSM_AVAR_JSONRPCRESPONSEDATA "JsonRpcResponseParameters" +#define DSM_AVAR_JSONRPCRESPONSEUDATA "JsonRpcResponseUData" #define DSM_ERRNO_FILE "file" #define DSM_ERRNO_UNKNOWN_ARG "arg" diff --git a/apps/dsm/SystemDSM.cpp b/apps/dsm/SystemDSM.cpp index 775c40a5..ae2dc08b 100644 --- a/apps/dsm/SystemDSM.cpp +++ b/apps/dsm/SystemDSM.cpp @@ -88,12 +88,17 @@ void SystemDSM::process(AmEvent* event) { // decode result for easy use from script varPrintArg(resp_ev->response.data, params, resp_ev->response.is_error ? "error": "result"); + // decode udata for easy use from script + varPrintArg(resp_ev->udata, params, "udata"); + // save reference to full parameters - avar[DSM_AVAR_JSONRPCRESPONEDATA] = AmArg(&resp_ev->response.data); + avar[DSM_AVAR_JSONRPCRESPONSEDATA] = AmArg(&resp_ev->response.data); + avar[DSM_AVAR_JSONRPCRESPONSEUDATA] = AmArg(&resp_ev->udata); engine.runEvent(&dummy_session, this, DSMCondition::JsonRpcResponse, ¶ms); - avar.erase(DSM_AVAR_JSONRPCRESPONEDATA); + avar.erase(DSM_AVAR_JSONRPCRESPONSEUDATA); + avar.erase(DSM_AVAR_JSONRPCRESPONSEDATA); return; } diff --git a/apps/jsonrpc/JsonRPC.cpp b/apps/jsonrpc/JsonRPC.cpp index 123e3133..0eec937f 100644 --- a/apps/jsonrpc/JsonRPC.cpp +++ b/apps/jsonrpc/JsonRPC.cpp @@ -123,7 +123,12 @@ void JsonRPCServerModule::execRpc(const AmArg& args, AmArg& ret) { AmArg& params = none_params; if (args.size()>7) params = args.get(7); - + + AmArg u_none_params; + AmArg& udata = u_none_params; + if (args.size()>8) + udata = args.get(8); + JsonRPCServerLoop::execRpc(// evq_link, notification_link, request_link args.get(0).asCStr(), args.get(1).asCStr(), args.get(2).asCStr(), @@ -132,7 +137,7 @@ void JsonRPCServerModule::execRpc(const AmArg& args, AmArg& ret) { // host, port, method args.get(4).asCStr(), args.get(5).asInt(), args.get(6).asCStr(), - params, ret); + params, udata, ret); } void JsonRPCServerModule::sendMessage(const AmArg& args, AmArg& ret) { @@ -140,10 +145,15 @@ void JsonRPCServerModule::sendMessage(const AmArg& args, AmArg& ret) { AmArg& params = none_params; if (args.size()>5) params = args.get(5); + AmArg u_none_params; + AmArg& udata = u_none_params; + if (args.size()>6) + udata = args.get(6); + JsonRPCServerLoop::sendMessage(args.get(0).asCStr(), // conn_id, args.get(1).asInt(), // type, (0 == reply) args.get(2).asCStr(), // method, args.get(3).asCStr(), // id args.get(4).asCStr(), // reply_sink - params, ret); + params, udata, ret); } diff --git a/apps/jsonrpc/JsonRPCEvents.h b/apps/jsonrpc/JsonRPCEvents.h index baac783b..5f606404 100644 --- a/apps/jsonrpc/JsonRPCEvents.h +++ b/apps/jsonrpc/JsonRPCEvents.h @@ -52,7 +52,7 @@ struct JsonRpcResponse { AmArg data; bool is_error; - JsonRpcResponse(bool is_error, string id, AmArg data) + JsonRpcResponse(bool is_error, string id, const AmArg& data) : is_error(is_error), id(id), data(data) { } JsonRpcResponse(bool is_error, string id) : is_error(is_error), id(id) { } @@ -63,9 +63,10 @@ struct JsonRpcResponse { struct JsonRpcResponseEvent : public JsonRpcEvent { JsonRpcResponse response; + AmArg udata; - JsonRpcResponseEvent(bool is_error, string id, AmArg data) - : response(is_error, id, data) + JsonRpcResponseEvent(bool is_error, string id, const AmArg& data, const AmArg& udata) + : response(is_error, id, data), udata(udata) { } JsonRpcResponseEvent(bool is_error, string id) : response(is_error, id) @@ -148,23 +149,25 @@ struct JsonServerSendMessageEvent AmArg params; string reply_link; bool is_error; + AmArg udata; JsonServerSendMessageEvent(const string& connection_id, bool is_reply, const string& method, const string& id, - AmArg& params, + const AmArg& params, + const AmArg& udata = AmArg(), const string& reply_link = "") : JsonServerEvent(connection_id, SendMessage), is_reply(is_reply), reply_link(reply_link), - method(method), id(id), params(params) { } + method(method), id(id), params(params), udata(udata) { } JsonServerSendMessageEvent(const JsonServerSendMessageEvent& e, JsonrpcNetstringsConnection* conn) : JsonServerEvent(conn, SendMessage), is_reply(e.is_reply),reply_link(e.reply_link), method(e.method), id(e.id), params(e.params), - is_error(e.is_error) { + is_error(e.is_error), udata(e.udata) { connection_id = e.connection_id; } diff --git a/apps/jsonrpc/JsonRPCServer.cpp b/apps/jsonrpc/JsonRPCServer.cpp index 715248f0..904f25ee 100644 --- a/apps/jsonrpc/JsonRPCServer.cpp +++ b/apps/jsonrpc/JsonRPCServer.cpp @@ -39,6 +39,7 @@ int JsonRpcServer::createRequest(const string& evq_link, const string& method, AmArg& params, JsonrpcNetstringsConnection* peer, + const AmArg& udata, bool is_notification) { AmArg rpc_params; rpc_params["jsonrpc"] = "2.0"; @@ -50,9 +51,10 @@ int JsonRpcServer::createRequest(const string& evq_link, const string& method, rpc_params["id"] = req_id; if (!evq_link.empty()) - peer->replyReceivers[req_id] = evq_link; + peer->replyReceivers[req_id] = make_pair(evq_link, udata); DBG("registering reply sink '%s' for id %s\n", evq_link.c_str(), req_id.c_str()); + } string rpc_params_json = arg2json(rpc_params); @@ -122,8 +124,8 @@ int JsonRpcServer::processMessage(char* msgbuf, unsigned int* msg_size, } string id = rpc_params["id"].asCStr(); - std::map::iterator rep_recv_q = - peer->replyReceivers.find(id); + std::map >::iterator + rep_recv_q = peer->replyReceivers.find(id); if (rep_recv_q == peer->replyReceivers.end()) { DBG("received reply for unknown request"); *msg_size = 0; @@ -134,20 +136,23 @@ int JsonRpcServer::processMessage(char* msgbuf, unsigned int* msg_size, } return 0; } + const AmArg& udata = rep_recv_q->second.second; + const string& event_queue_id = rep_recv_q->second.first; + JsonRpcResponseEvent* resp_ev = NULL; if (rpc_params.hasMember("result")) { - resp_ev = new JsonRpcResponseEvent(false, id, rpc_params["result"]); + resp_ev = new JsonRpcResponseEvent(false, id, rpc_params["result"], udata); } else { if (!rpc_params.hasMember("error")) { INFO("protocol error: reply does not have error nor result!\n"); return -2; } - resp_ev = new JsonRpcResponseEvent(true, id, rpc_params["error"]); + resp_ev = new JsonRpcResponseEvent(true, id, rpc_params["error"], udata); } resp_ev->connection_id = peer->id; bool posted = AmEventDispatcher::instance()-> - post(rep_recv_q->second, resp_ev); + post(event_queue_id, resp_ev); if (!posted) { DBG("receiver event queue does not exist (any more)\n"); peer->replyReceivers.erase(rep_recv_q); diff --git a/apps/jsonrpc/JsonRPCServer.h b/apps/jsonrpc/JsonRPCServer.h index 039c08b6..8860e137 100644 --- a/apps/jsonrpc/JsonRPCServer.h +++ b/apps/jsonrpc/JsonRPCServer.h @@ -55,7 +55,8 @@ class JsonRpcServer { JsonrpcPeerConnection* peer); static int createRequest(const string& evq_link, const string& method, AmArg& params, - JsonrpcNetstringsConnection* peer, bool is_notification = false); + JsonrpcNetstringsConnection* peer, const AmArg& udata, + bool is_notification = false); static int createReply(JsonrpcNetstringsConnection* peer, const string& id, AmArg& result, bool is_error); diff --git a/apps/jsonrpc/RpcPeer.cpp b/apps/jsonrpc/RpcPeer.cpp index 6b823f32..f7496257 100644 --- a/apps/jsonrpc/RpcPeer.cpp +++ b/apps/jsonrpc/RpcPeer.cpp @@ -54,10 +54,10 @@ void JsonrpcPeerConnection::notifyDisconnect() { post(requestReceiver, new JsonRpcConnectionEvent(JsonRpcConnectionEvent::DISCONNECT, id)); - for (std::map::iterator it= + for (std::map > ::iterator it= replyReceivers.begin(); it != replyReceivers.end(); it++) { AmEventDispatcher::instance()-> - post(it->second, + post(it->second.first, new JsonRpcConnectionEvent(JsonRpcConnectionEvent::DISCONNECT, id)); } } diff --git a/apps/jsonrpc/RpcPeer.h b/apps/jsonrpc/RpcPeer.h index 360fba4a..4734356e 100644 --- a/apps/jsonrpc/RpcPeer.h +++ b/apps/jsonrpc/RpcPeer.h @@ -31,6 +31,7 @@ #include #include #include "log.h" +#include "AmArg.h" #define MAX_RPC_MSG_SIZE 20*1024*1024 // 20k #define MAX_NS_LEN_SIZE 10 @@ -44,7 +45,8 @@ struct JsonrpcPeerConnection { // event queue keys that should receive the reply // to requests sent on that connection - std::map replyReceivers; + // req_id queue udata + std::map > replyReceivers; // if present, notifications will be sent // to that event queue directly diff --git a/apps/jsonrpc/RpcServerLoop.cpp b/apps/jsonrpc/RpcServerLoop.cpp index 5676eee3..30c3e52d 100644 --- a/apps/jsonrpc/RpcServerLoop.cpp +++ b/apps/jsonrpc/RpcServerLoop.cpp @@ -367,7 +367,8 @@ void JsonRPCServerLoop::execRpc(const string& evq_link, int flags, const string& host, int port, const string& method, - AmArg& params, + const AmArg& params, + const AmArg& udata, AmArg& ret) { string connection_id = newConnectionId(); JsonrpcNetstringsConnection* peer = new JsonrpcNetstringsConnection(connection_id); @@ -397,9 +398,7 @@ void JsonRPCServerLoop::execRpc(const string& evq_link, DBG("dispatching JsonServerSendMessageEvent\n"); JsonServerSendMessageEvent* send_message_event = new JsonServerSendMessageEvent(connection_id, false, method, "1" /* id - not empty */, - params); - - send_message_event->reply_link = evq_link; + params, udata, evq_link); JsonRPCServerLoop::dispatchServerEvent(send_message_event); @@ -416,7 +415,8 @@ void JsonRPCServerLoop::sendMessage(const string& connection_id, const string& method, const string& id, const string& reply_sink, - AmArg& params, + const AmArg& params, + const AmArg& udata, AmArg& ret) { // check for presence of connection // (connection might still be removed until we really @@ -430,7 +430,7 @@ void JsonRPCServerLoop::sendMessage(const string& connection_id, JsonServerSendMessageEvent* ev = new JsonServerSendMessageEvent(connection_id, msg_type != JSONRPC_MSG_REQUEST, - method, id, params, reply_sink); + method, id, params, udata, reply_sink); ev->is_error = msg_type == JSONRPC_MSG_ERROR; instance()->postEvent(ev); diff --git a/apps/jsonrpc/RpcServerLoop.h b/apps/jsonrpc/RpcServerLoop.h index e4382778..e56144f0 100644 --- a/apps/jsonrpc/RpcServerLoop.h +++ b/apps/jsonrpc/RpcServerLoop.h @@ -72,7 +72,8 @@ class JsonRPCServerLoop int flags, const string& host, int port, const string& method, - AmArg& params, + const AmArg& params, + const AmArg& udata, AmArg& ret); static void sendMessage(const string& connection_id, @@ -80,7 +81,8 @@ class JsonRPCServerLoop const string& method, const string& id, const string& reply_sink, - AmArg& params, + const AmArg& params, + const AmArg& udata, AmArg& ret); void run(); void on_stop(); diff --git a/apps/jsonrpc/RpcServerThread.cpp b/apps/jsonrpc/RpcServerThread.cpp index 3225a3e3..f4cd7979 100644 --- a/apps/jsonrpc/RpcServerThread.cpp +++ b/apps/jsonrpc/RpcServerThread.cpp @@ -89,7 +89,8 @@ void RpcServerThread::process(AmEvent* event) { if (!snd_msg_ev->is_reply) { if (JsonRpcServer::createRequest(snd_msg_ev->reply_link, snd_msg_ev->method, - snd_msg_ev->params, connection, + snd_msg_ev->params, connection, + snd_msg_ev->udata, snd_msg_ev->id.empty())) { ERROR("creating request\n"); // give back connection into server loop