sipwise
/
sems
mirror of https://github.com/sipwise/sems.git
1
0
Fork 0
Code Issues Releases Wiki Activity

jsonrpc: support for per request user data

Browse Source 
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.
sayer/1.4-spce2.6
Stefan Sayer 16 years ago
parent e73038f1fc
commit edd2fa1cc0
12 changed files with 69 additions and 34 deletions
Show Stats Download Patch File Download Diff File

11
apps/dsm/DSMCall.cpp
Unescape View File

@ -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, &params);
avar.erase(DSM_AVAR_JSONRPCRESPONEDATA);
avar.erase(DSM_AVAR_JSONRPCRESPONSEUDATA);
avar.erase(DSM_AVAR_JSONRPCRESPONSEDATA);
return;
}

3
apps/dsm/DSMSession.h
Unescape View File

@ -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"

9
apps/dsm/SystemDSM.cpp
Unescape View File

@ -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, &params);
avar.erase(DSM_AVAR_JSONRPCRESPONEDATA);
avar.erase(DSM_AVAR_JSONRPCRESPONSEUDATA);
avar.erase(DSM_AVAR_JSONRPCRESPONSEDATA);
return;
}

16
apps/jsonrpc/JsonRPC.cpp
Unescape View File

@ -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);
}

15
apps/jsonrpc/JsonRPCEvents.h
Unescape View File

@ -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;
}

17
apps/jsonrpc/JsonRPCServer.cpp
Unescape View File

@ -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<std::string, std::string>::iterator rep_recv_q =
peer->replyReceivers.find(id);
std::map<std::string, std::pair<std::string, AmArg > >::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);

3
apps/jsonrpc/JsonRPCServer.h
Unescape View File

@ -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);

4
apps/jsonrpc/RpcPeer.cpp
Unescape View File

@ -54,10 +54,10 @@ void JsonrpcPeerConnection::notifyDisconnect() {
post(requestReceiver,
new JsonRpcConnectionEvent(JsonRpcConnectionEvent::DISCONNECT, id));
for (std::map<std::string, std::string>::iterator it=
for (std::map<std::string, std::pair<std::string, AmArg > > ::iterator it=
replyReceivers.begin(); it != replyReceivers.end(); it++) {
AmEventDispatcher::instance()->
post(it->second,
post(it->second.first,
new JsonRpcConnectionEvent(JsonRpcConnectionEvent::DISCONNECT, id));
}
}

4
apps/jsonrpc/RpcPeer.h
Unescape View File

@ -31,6 +31,7 @@
#include <ev.h>
#include <stdlib.h>
#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<std::string, std::string> replyReceivers;
// req_id queue udata
std::map<std::string, std::pair<std::string, AmArg > > replyReceivers;
// if present, notifications will be sent
// to that event queue directly

12
apps/jsonrpc/RpcServerLoop.cpp
Unescape View File

@ -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);

6
apps/jsonrpc/RpcServerLoop.h
Unescape View File

@ -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();

3
apps/jsonrpc/RpcServerThread.cpp
Unescape View File

@ -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

Write Preview
Loading…
Cancel
Save