X-Git-Url: https://pintos-os.org/cgi-bin/gitweb.cgi?a=blobdiff_plain;f=lib%2Fjsonrpc.c;h=bd019f7381694d3b0806315375ee7a2717ae9105;hb=c75d15113c56e30ea5a1ace777cd3105c7b750d3;hp=d298f0d5957a1385af00edc160ce1bcf2279ea00;hpb=d0632593feacc7942fa1650baf07e1e0b8dd1676;p=openvswitch diff --git a/lib/jsonrpc.c b/lib/jsonrpc.c index d298f0d5..bd019f73 100644 --- a/lib/jsonrpc.c +++ b/lib/jsonrpc.c @@ -21,12 +21,15 @@ #include #include "byteq.h" +#include "dynamic-string.h" #include "json.h" #include "list.h" #include "ofpbuf.h" #include "poll-loop.h" #include "queue.h" +#include "reconnect.h" #include "stream.h" +#include "timeval.h" #define THIS_MODULE VLM_jsonrpc #include "vlog.h" @@ -133,6 +136,37 @@ jsonrpc_get_name(const struct jsonrpc *rpc) return rpc->name; } +static void +jsonrpc_log_msg(const struct jsonrpc *rpc, const char *title, + const struct jsonrpc_msg *msg) +{ + if (VLOG_IS_DBG_ENABLED()) { + struct ds s = DS_EMPTY_INITIALIZER; + if (msg->method) { + ds_put_format(&s, ", method=\"%s\"", msg->method); + } + if (msg->params) { + ds_put_cstr(&s, ", params="); + ds_put_and_free_cstr(&s, json_to_string(msg->params, 0)); + } + if (msg->result) { + ds_put_cstr(&s, ", result="); + ds_put_and_free_cstr(&s, json_to_string(msg->result, 0)); + } + if (msg->error) { + ds_put_cstr(&s, ", error="); + ds_put_and_free_cstr(&s, json_to_string(msg->error, 0)); + } + if (msg->id) { + ds_put_cstr(&s, ", id="); + ds_put_and_free_cstr(&s, json_to_string(msg->id, 0)); + } + VLOG_DBG("%s: %s %s%s", rpc->name, title, + jsonrpc_msg_type_to_string(msg->type), ds_cstr(&s)); + ds_destroy(&s); + } +} + int jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg) { @@ -146,6 +180,8 @@ jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg) return rpc->status; } + jsonrpc_log_msg(rpc, "send", msg); + json = jsonrpc_msg_to_json(msg); s = json_to_string(json, 0); length = strlen(s); @@ -313,6 +349,7 @@ jsonrpc_received(struct jsonrpc *rpc) return; } + jsonrpc_log_msg(rpc, "received", msg); rpc->received = msg; } @@ -365,10 +402,14 @@ jsonrpc_create_id(void) } struct jsonrpc_msg * -jsonrpc_create_request(const char *method, struct json *params) +jsonrpc_create_request(const char *method, struct json *params, + struct json **idp) { - return jsonrpc_create(JSONRPC_REQUEST, method, params, NULL, NULL, - jsonrpc_create_id()); + struct json *id = jsonrpc_create_id(); + if (idp) { + *idp = json_clone(id); + } + return jsonrpc_create(JSONRPC_REQUEST, method, params, NULL, NULL, id); } struct jsonrpc_msg * @@ -579,3 +620,185 @@ jsonrpc_msg_to_json(struct jsonrpc_msg *m) return json; } + +/* A JSON-RPC session with reconnection. */ + +struct jsonrpc_session { + struct reconnect *reconnect; + struct jsonrpc *rpc; + struct stream *stream; + unsigned int seqno; +}; + +struct jsonrpc_session * +jsonrpc_session_open(const char *name) +{ + struct jsonrpc_session *s; + + s = xmalloc(sizeof *s); + s->reconnect = reconnect_create(time_msec()); + reconnect_set_name(s->reconnect, name); + reconnect_enable(s->reconnect, time_msec()); + s->rpc = NULL; + s->stream = NULL; + s->seqno = 0; + + return s; +} + +void +jsonrpc_session_close(struct jsonrpc_session *s) +{ + if (s) { + jsonrpc_close(s->rpc); + reconnect_destroy(s->reconnect); + free(s); + } +} + +static void +jsonrpc_session_disconnect(struct jsonrpc_session *s) +{ + reconnect_disconnected(s->reconnect, time_msec(), 0); + if (s->rpc) { + jsonrpc_error(s->rpc, EOF); + jsonrpc_close(s->rpc); + s->rpc = NULL; + s->seqno++; + } else if (s->stream) { + stream_close(s->stream); + s->stream = NULL; + s->seqno++; + } +} + +static void +jsonrpc_session_connect(struct jsonrpc_session *s) +{ + int error; + + jsonrpc_session_disconnect(s); + error = stream_open(reconnect_get_name(s->reconnect), &s->stream); + if (error) { + reconnect_connect_failed(s->reconnect, time_msec(), error); + } else { + reconnect_connecting(s->reconnect, time_msec()); + } + s->seqno++; +} + +void +jsonrpc_session_run(struct jsonrpc_session *s) +{ + if (s->rpc) { + int error; + + jsonrpc_run(s->rpc); + error = jsonrpc_get_status(s->rpc); + if (error) { + jsonrpc_session_disconnect(s); + } + } else if (s->stream) { + int error = stream_connect(s->stream); + if (!error) { + reconnect_connected(s->reconnect, time_msec()); + s->rpc = jsonrpc_open(s->stream); + s->stream = NULL; + } else if (error != EAGAIN) { + reconnect_connect_failed(s->reconnect, time_msec(), error); + stream_close(s->stream); + s->stream = NULL; + } + } + + switch (reconnect_run(s->reconnect, time_msec())) { + case RECONNECT_CONNECT: + jsonrpc_session_connect(s); + break; + + case RECONNECT_DISCONNECT: + jsonrpc_session_disconnect(s); + break; + + case RECONNECT_PROBE: + if (s->rpc) { + struct json *params; + struct jsonrpc_msg *request; + + params = json_array_create_empty(); + request = jsonrpc_create_request("echo", params, NULL); + json_destroy(request->id); + request->id = json_string_create("echo"); + jsonrpc_send(s->rpc, request); + } + break; + } +} + +void +jsonrpc_session_wait(struct jsonrpc_session *s) +{ + if (s->rpc) { + jsonrpc_wait(s->rpc); + } else if (s->stream) { + stream_connect_wait(s->stream); + } + reconnect_wait(s->reconnect, time_msec()); +} + +size_t +jsonrpc_session_get_backlog(const struct jsonrpc_session *s) +{ + return s->rpc ? jsonrpc_get_backlog(s->rpc) : 0; +} + +const char * +jsonrpc_session_get_name(const struct jsonrpc_session *s) +{ + return reconnect_get_name(s->reconnect); +} + +int +jsonrpc_session_send(struct jsonrpc_session *s, struct jsonrpc_msg *msg) +{ + return s->rpc ? jsonrpc_send(s->rpc, msg) : ENOTCONN; +} + +struct jsonrpc_msg * +jsonrpc_session_recv(struct jsonrpc_session *s) +{ + struct jsonrpc_msg *msg = NULL; + if (s->rpc) { + jsonrpc_recv(s->rpc, &msg); + if (msg) { + reconnect_received(s->reconnect, time_msec()); + } + } + return msg; +} + +void +jsonrpc_session_recv_wait(struct jsonrpc_session *s) +{ + if (s->rpc) { + jsonrpc_recv_wait(s->rpc); + } +} + +bool +jsonrpc_session_is_connected(const struct jsonrpc_session *s) +{ + return s->rpc != NULL; +} + +unsigned int +jsonrpc_session_get_seqno(const struct jsonrpc_session *s) +{ + return s->seqno; +} + +void +jsonrpc_session_force_reconnect(struct jsonrpc_session *s) +{ + reconnect_force_reconnect(s->reconnect, time_msec()); +}