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