2 * Copyright (c) 2009 Nicira Networks.
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at:
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
24 #include "dynamic-string.h"
28 #include "poll-loop.h"
32 #define THIS_MODULE VLM_jsonrpc
36 struct stream *stream;
42 struct json_parser *parser;
43 struct jsonrpc_msg *received;
46 struct ovs_queue output;
50 /* Rate limit for error messages. */
51 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
53 static void jsonrpc_received(struct jsonrpc *);
54 static void jsonrpc_cleanup(struct jsonrpc *);
57 jsonrpc_open(struct stream *stream)
61 assert(stream != NULL);
63 rpc = xzalloc(sizeof *rpc);
64 rpc->name = xstrdup(stream_get_name(stream));
66 byteq_init(&rpc->input);
67 queue_init(&rpc->output);
73 jsonrpc_close(struct jsonrpc *rpc)
83 jsonrpc_run(struct jsonrpc *rpc)
89 while (!queue_is_empty(&rpc->output)) {
90 struct ofpbuf *buf = rpc->output.head;
93 retval = stream_send(rpc->stream, buf->data, buf->size);
95 rpc->backlog -= retval;
96 ofpbuf_pull(buf, retval);
98 ofpbuf_delete(queue_pop_head(&rpc->output));
101 if (retval != -EAGAIN) {
102 VLOG_WARN_RL(&rl, "%s: send error: %s",
103 rpc->name, strerror(-retval));
104 jsonrpc_error(rpc, -retval);
112 jsonrpc_wait(struct jsonrpc *rpc)
114 if (!rpc->status && !queue_is_empty(&rpc->output)) {
115 stream_send_wait(rpc->stream);
120 jsonrpc_get_status(const struct jsonrpc *rpc)
126 jsonrpc_get_backlog(const struct jsonrpc *rpc)
128 return rpc->status ? 0 : rpc->backlog;
132 jsonrpc_get_name(const struct jsonrpc *rpc)
138 jsonrpc_log_msg(const struct jsonrpc *rpc, const char *title,
139 const struct jsonrpc_msg *msg)
141 if (VLOG_IS_DBG_ENABLED()) {
142 struct ds s = DS_EMPTY_INITIALIZER;
144 ds_put_format(&s, ", method=\"%s\"", msg->method);
147 ds_put_cstr(&s, ", params=");
148 ds_put_and_free_cstr(&s, json_to_string(msg->params, 0));
151 ds_put_cstr(&s, ", result=");
152 ds_put_and_free_cstr(&s, json_to_string(msg->result, 0));
155 ds_put_cstr(&s, ", error=");
156 ds_put_and_free_cstr(&s, json_to_string(msg->error, 0));
159 ds_put_cstr(&s, ", id=");
160 ds_put_and_free_cstr(&s, json_to_string(msg->id, 0));
162 VLOG_DBG("%s: %s %s%s", rpc->name, title,
163 jsonrpc_msg_type_to_string(msg->type), ds_cstr(&s));
169 jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
177 jsonrpc_msg_destroy(msg);
181 jsonrpc_log_msg(rpc, "send", msg);
183 json = jsonrpc_msg_to_json(msg);
184 s = json_to_string(json, 0);
188 buf = xmalloc(sizeof *buf);
189 ofpbuf_use(buf, s, length);
191 queue_push_tail(&rpc->output, buf);
192 rpc->backlog += length;
194 if (rpc->output.n == 1) {
201 jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
208 while (!rpc->received) {
209 if (byteq_is_empty(&rpc->input)) {
213 chunk = byteq_headroom(&rpc->input);
214 retval = stream_recv(rpc->stream, byteq_head(&rpc->input), chunk);
216 if (retval == -EAGAIN) {
219 VLOG_WARN_RL(&rl, "%s: receive error: %s",
220 rpc->name, strerror(-retval));
221 jsonrpc_error(rpc, -retval);
224 } else if (retval == 0) {
225 VLOG_INFO_RL(&rl, "%s: connection closed", rpc->name);
226 jsonrpc_error(rpc, EOF);
229 byteq_advance_head(&rpc->input, retval);
234 rpc->parser = json_parser_create(0);
236 n = byteq_tailroom(&rpc->input);
237 used = json_parser_feed(rpc->parser,
238 (char *) byteq_tail(&rpc->input), n);
239 byteq_advance_tail(&rpc->input, used);
240 if (json_parser_is_done(rpc->parser)) {
241 jsonrpc_received(rpc);
249 *msgp = rpc->received;
250 rpc->received = NULL;
255 jsonrpc_recv_wait(struct jsonrpc *rpc)
257 if (rpc->status || rpc->received || !byteq_is_empty(&rpc->input)) {
258 poll_immediate_wake();
260 stream_recv_wait(rpc->stream);
265 jsonrpc_send_block(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
269 error = jsonrpc_send(rpc, msg);
274 while (!queue_is_empty(&rpc->output) && !rpc->status) {
283 jsonrpc_recv_block(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
286 int error = jsonrpc_recv(rpc, msgp);
287 if (error != EAGAIN) {
293 jsonrpc_recv_wait(rpc);
299 jsonrpc_transact_block(struct jsonrpc *rpc, struct jsonrpc_msg *request,
300 struct jsonrpc_msg **replyp)
302 struct jsonrpc_msg *reply = NULL;
306 id = json_clone(request->id);
307 error = jsonrpc_send_block(rpc, request);
310 error = jsonrpc_recv_block(rpc, &reply);
312 || (reply->type == JSONRPC_REPLY
313 && json_equal(id, reply->id))) {
316 jsonrpc_msg_destroy(reply);
319 *replyp = error ? NULL : reply;
325 jsonrpc_received(struct jsonrpc *rpc)
327 struct jsonrpc_msg *msg;
331 json = json_parser_finish(rpc->parser);
333 if (json->type == JSON_STRING) {
334 VLOG_WARN_RL(&rl, "%s: error parsing stream: %s",
335 rpc->name, json_string(json));
336 jsonrpc_error(rpc, EPROTO);
341 error = jsonrpc_msg_from_json(json, &msg);
343 VLOG_WARN_RL(&rl, "%s: received bad JSON-RPC message: %s",
346 jsonrpc_error(rpc, EPROTO);
350 jsonrpc_log_msg(rpc, "received", msg);
355 jsonrpc_error(struct jsonrpc *rpc, int error)
360 jsonrpc_cleanup(rpc);
365 jsonrpc_cleanup(struct jsonrpc *rpc)
367 stream_close(rpc->stream);
370 json_parser_abort(rpc->parser);
373 jsonrpc_msg_destroy(rpc->received);
374 rpc->received = NULL;
376 queue_clear(&rpc->output);
380 static struct jsonrpc_msg *
381 jsonrpc_create(enum jsonrpc_msg_type type, const char *method,
382 struct json *params, struct json *result, struct json *error,
385 struct jsonrpc_msg *msg = xmalloc(sizeof *msg);
387 msg->method = method ? xstrdup(method) : NULL;
388 msg->params = params;
389 msg->result = result;
396 jsonrpc_create_id(void)
398 static unsigned int id;
399 return json_integer_create(id++);
403 jsonrpc_create_request(const char *method, struct json *params)
405 return jsonrpc_create(JSONRPC_REQUEST, method, params, NULL, NULL,
406 jsonrpc_create_id());
410 jsonrpc_create_notify(const char *method, struct json *params)
412 return jsonrpc_create(JSONRPC_NOTIFY, method, params, NULL, NULL, NULL);
416 jsonrpc_create_reply(struct json *result, const struct json *id)
418 return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, result, NULL,
423 jsonrpc_create_error(struct json *error, const struct json *id)
425 return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, NULL, error,
430 jsonrpc_msg_type_to_string(enum jsonrpc_msg_type type)
433 case JSONRPC_REQUEST:
437 return "notification";
449 jsonrpc_msg_is_valid(const struct jsonrpc_msg *m)
451 const char *type_name;
452 unsigned int pattern;
454 if (m->params && m->params->type != JSON_ARRAY) {
455 return xstrdup("\"params\" must be JSON array");
459 case JSONRPC_REQUEST:
476 return xasprintf("invalid JSON-RPC message type %d", m->type);
479 type_name = jsonrpc_msg_type_to_string(m->type);
480 if ((m->method != NULL) != ((pattern & 0x10000) != 0)) {
481 return xasprintf("%s must%s have \"method\"",
482 type_name, (pattern & 0x10000) ? "" : " not");
485 if ((m->params != NULL) != ((pattern & 0x1000) != 0)) {
486 return xasprintf("%s must%s have \"params\"",
487 type_name, (pattern & 0x1000) ? "" : " not");
490 if ((m->result != NULL) != ((pattern & 0x100) != 0)) {
491 return xasprintf("%s must%s have \"result\"",
492 type_name, (pattern & 0x100) ? "" : " not");
495 if ((m->error != NULL) != ((pattern & 0x10) != 0)) {
496 return xasprintf("%s must%s have \"error\"",
497 type_name, (pattern & 0x10) ? "" : " not");
500 if ((m->id != NULL) != ((pattern & 0x1) != 0)) {
501 return xasprintf("%s must%s have \"id\"",
502 type_name, (pattern & 0x1) ? "" : " not");
509 jsonrpc_msg_destroy(struct jsonrpc_msg *m)
513 json_destroy(m->params);
514 json_destroy(m->result);
515 json_destroy(m->error);
522 null_from_json_null(struct json *json)
524 if (json && json->type == JSON_NULL) {
532 jsonrpc_msg_from_json(struct json *json, struct jsonrpc_msg **msgp)
534 struct json *method = NULL;
535 struct jsonrpc_msg *msg = NULL;
536 struct shash *object;
539 if (json->type != JSON_OBJECT) {
540 error = xstrdup("message is not a JSON object");
543 object = json_object(json);
545 method = shash_find_and_delete(object, "method");
546 if (method && method->type != JSON_STRING) {
547 error = xstrdup("method is not a JSON string");
551 msg = xzalloc(sizeof *msg);
552 msg->method = method ? xstrdup(method->u.string) : NULL;
553 msg->params = null_from_json_null(shash_find_and_delete(object, "params"));
554 msg->result = null_from_json_null(shash_find_and_delete(object, "result"));
555 msg->error = null_from_json_null(shash_find_and_delete(object, "error"));
556 msg->id = null_from_json_null(shash_find_and_delete(object, "id"));
557 msg->type = (msg->result ? JSONRPC_REPLY
558 : msg->error ? JSONRPC_ERROR
559 : msg->id ? JSONRPC_REQUEST
561 if (!shash_is_empty(object)) {
562 error = xasprintf("message has unexpected member \"%s\"",
563 shash_first(object)->name);
566 error = jsonrpc_msg_is_valid(msg);
572 json_destroy(method);
575 jsonrpc_msg_destroy(msg);
583 jsonrpc_msg_to_json(struct jsonrpc_msg *m)
585 struct json *json = json_object_create();
588 json_object_put(json, "method", json_string_create_nocopy(m->method));
592 json_object_put(json, "params", m->params);
596 json_object_put(json, "result", m->result);
597 } else if (m->type == JSONRPC_ERROR) {
598 json_object_put(json, "result", json_null_create());
602 json_object_put(json, "error", m->error);
603 } else if (m->type == JSONRPC_REPLY) {
604 json_object_put(json, "error", json_null_create());
608 json_object_put(json, "id", m->id);
609 } else if (m->type == JSONRPC_NOTIFY) {
610 json_object_put(json, "id", json_null_create());