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.
27 #include "poll-loop.h"
31 #define THIS_MODULE VLM_jsonrpc
35 struct stream *stream;
41 struct json_parser *parser;
42 struct jsonrpc_msg *received;
45 struct ovs_queue output;
49 /* Rate limit for error messages. */
50 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
52 static void jsonrpc_received(struct jsonrpc *);
53 static void jsonrpc_cleanup(struct jsonrpc *);
56 jsonrpc_open(struct stream *stream)
60 assert(stream != NULL);
62 rpc = xzalloc(sizeof *rpc);
63 rpc->name = xstrdup(stream_get_name(stream));
65 byteq_init(&rpc->input);
66 queue_init(&rpc->output);
72 jsonrpc_close(struct jsonrpc *rpc)
82 jsonrpc_run(struct jsonrpc *rpc)
88 while (!queue_is_empty(&rpc->output)) {
89 struct ofpbuf *buf = rpc->output.head;
92 retval = stream_send(rpc->stream, buf->data, buf->size);
94 rpc->backlog -= retval;
95 ofpbuf_pull(buf, retval);
97 ofpbuf_delete(queue_pop_head(&rpc->output));
100 if (retval != -EAGAIN) {
101 VLOG_WARN_RL(&rl, "%s: send error: %s",
102 rpc->name, strerror(-retval));
103 jsonrpc_error(rpc, -retval);
111 jsonrpc_wait(struct jsonrpc *rpc)
113 if (!rpc->status && !queue_is_empty(&rpc->output)) {
114 stream_send_wait(rpc->stream);
119 jsonrpc_get_status(const struct jsonrpc *rpc)
125 jsonrpc_get_backlog(const struct jsonrpc *rpc)
127 return rpc->status ? 0 : rpc->backlog;
131 jsonrpc_get_name(const struct jsonrpc *rpc)
137 jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
145 jsonrpc_msg_destroy(msg);
149 json = jsonrpc_msg_to_json(msg);
150 s = json_to_string(json, 0);
154 buf = xmalloc(sizeof *buf);
155 ofpbuf_use(buf, s, length);
157 queue_push_tail(&rpc->output, buf);
158 rpc->backlog += length;
160 if (rpc->output.n == 1) {
167 jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
174 while (!rpc->received) {
175 if (byteq_is_empty(&rpc->input)) {
179 chunk = byteq_headroom(&rpc->input);
180 retval = stream_recv(rpc->stream, byteq_head(&rpc->input), chunk);
182 if (retval == -EAGAIN) {
185 VLOG_WARN_RL(&rl, "%s: receive error: %s",
186 rpc->name, strerror(-retval));
187 jsonrpc_error(rpc, -retval);
190 } else if (retval == 0) {
191 VLOG_INFO_RL(&rl, "%s: connection closed", rpc->name);
192 jsonrpc_error(rpc, EOF);
195 byteq_advance_head(&rpc->input, retval);
200 rpc->parser = json_parser_create(0);
202 n = byteq_tailroom(&rpc->input);
203 used = json_parser_feed(rpc->parser,
204 (char *) byteq_tail(&rpc->input), n);
205 byteq_advance_tail(&rpc->input, used);
206 if (json_parser_is_done(rpc->parser)) {
207 jsonrpc_received(rpc);
215 *msgp = rpc->received;
216 rpc->received = NULL;
221 jsonrpc_recv_wait(struct jsonrpc *rpc)
223 if (rpc->status || rpc->received || !byteq_is_empty(&rpc->input)) {
224 poll_immediate_wake();
226 stream_recv_wait(rpc->stream);
231 jsonrpc_send_block(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
235 error = jsonrpc_send(rpc, msg);
240 while (!queue_is_empty(&rpc->output) && !rpc->status) {
249 jsonrpc_recv_block(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
252 int error = jsonrpc_recv(rpc, msgp);
253 if (error != EAGAIN) {
259 jsonrpc_recv_wait(rpc);
265 jsonrpc_received(struct jsonrpc *rpc)
267 struct jsonrpc_msg *msg;
271 json = json_parser_finish(rpc->parser);
273 if (json->type == JSON_STRING) {
274 VLOG_WARN_RL(&rl, "%s: error parsing stream: %s",
275 rpc->name, json_string(json));
276 jsonrpc_error(rpc, EPROTO);
281 error = jsonrpc_msg_from_json(json, &msg);
283 VLOG_WARN_RL(&rl, "%s: received bad JSON-RPC message: %s",
286 jsonrpc_error(rpc, EPROTO);
294 jsonrpc_error(struct jsonrpc *rpc, int error)
299 jsonrpc_cleanup(rpc);
304 jsonrpc_cleanup(struct jsonrpc *rpc)
306 stream_close(rpc->stream);
309 json_parser_abort(rpc->parser);
312 jsonrpc_msg_destroy(rpc->received);
313 rpc->received = NULL;
315 queue_clear(&rpc->output);
319 static struct jsonrpc_msg *
320 jsonrpc_create(enum jsonrpc_msg_type type, const char *method,
321 struct json *params, struct json *result, struct json *error,
324 struct jsonrpc_msg *msg = xmalloc(sizeof *msg);
326 msg->method = method ? xstrdup(method) : NULL;
327 msg->params = params;
328 msg->result = result;
335 jsonrpc_create_id(void)
337 static unsigned int id;
338 return json_integer_create(id++);
342 jsonrpc_create_request(const char *method, struct json *params)
344 return jsonrpc_create(JSONRPC_REQUEST, method, params, NULL, NULL,
345 jsonrpc_create_id());
349 jsonrpc_create_notify(const char *method, struct json *params)
351 return jsonrpc_create(JSONRPC_NOTIFY, method, params, NULL, NULL, NULL);
355 jsonrpc_create_reply(struct json *result, const struct json *id)
357 return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, result, NULL,
362 jsonrpc_create_error(struct json *error, const struct json *id)
364 return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, NULL, error,
369 jsonrpc_msg_type_to_string(enum jsonrpc_msg_type type)
372 case JSONRPC_REQUEST:
376 return "notification";
388 jsonrpc_msg_is_valid(const struct jsonrpc_msg *m)
390 const char *type_name;
391 unsigned int pattern;
393 if (m->params && m->params->type != JSON_ARRAY) {
394 return xstrdup("\"params\" must be JSON array");
398 case JSONRPC_REQUEST:
415 return xasprintf("invalid JSON-RPC message type %d", m->type);
418 type_name = jsonrpc_msg_type_to_string(m->type);
419 if ((m->method != NULL) != ((pattern & 0x10000) != 0)) {
420 return xasprintf("%s must%s have \"method\"",
421 type_name, (pattern & 0x10000) ? "" : " not");
424 if ((m->params != NULL) != ((pattern & 0x1000) != 0)) {
425 return xasprintf("%s must%s have \"params\"",
426 type_name, (pattern & 0x1000) ? "" : " not");
429 if ((m->result != NULL) != ((pattern & 0x100) != 0)) {
430 return xasprintf("%s must%s have \"result\"",
431 type_name, (pattern & 0x100) ? "" : " not");
434 if ((m->error != NULL) != ((pattern & 0x10) != 0)) {
435 return xasprintf("%s must%s have \"error\"",
436 type_name, (pattern & 0x10) ? "" : " not");
439 if ((m->id != NULL) != ((pattern & 0x1) != 0)) {
440 return xasprintf("%s must%s have \"id\"",
441 type_name, (pattern & 0x1) ? "" : " not");
448 jsonrpc_msg_destroy(struct jsonrpc_msg *m)
452 json_destroy(m->params);
453 json_destroy(m->result);
454 json_destroy(m->error);
461 null_from_json_null(struct json *json)
463 if (json && json->type == JSON_NULL) {
471 jsonrpc_msg_from_json(struct json *json, struct jsonrpc_msg **msgp)
473 struct json *method = NULL;
474 struct jsonrpc_msg *msg = NULL;
475 struct shash *object;
478 if (json->type != JSON_OBJECT) {
479 error = xstrdup("message is not a JSON object");
482 object = json_object(json);
484 method = shash_find_and_delete(object, "method");
485 if (method && method->type != JSON_STRING) {
486 error = xstrdup("method is not a JSON string");
490 msg = xzalloc(sizeof *msg);
491 msg->method = method ? xstrdup(method->u.string) : NULL;
492 msg->params = null_from_json_null(shash_find_and_delete(object, "params"));
493 msg->result = null_from_json_null(shash_find_and_delete(object, "result"));
494 msg->error = null_from_json_null(shash_find_and_delete(object, "error"));
495 msg->id = null_from_json_null(shash_find_and_delete(object, "id"));
496 msg->type = (msg->result ? JSONRPC_REPLY
497 : msg->error ? JSONRPC_ERROR
498 : msg->id ? JSONRPC_REQUEST
500 if (!shash_is_empty(object)) {
501 error = xasprintf("message has unexpected member \"%s\"",
502 shash_first(object)->name);
505 error = jsonrpc_msg_is_valid(msg);
511 json_destroy(method);
514 jsonrpc_msg_destroy(msg);
522 jsonrpc_msg_to_json(struct jsonrpc_msg *m)
524 struct json *json = json_object_create();
527 json_object_put(json, "method", json_string_create_nocopy(m->method));
531 json_object_put(json, "params", m->params);
535 json_object_put(json, "result", m->result);
536 } else if (m->type == JSONRPC_ERROR) {
537 json_object_put(json, "result", json_null_create());
541 json_object_put(json, "error", m->error);
542 } else if (m->type == JSONRPC_REPLY) {
543 json_object_put(json, "error", json_null_create());
547 json_object_put(json, "id", m->id);
548 } else if (m->type == JSONRPC_NOTIFY) {
549 json_object_put(json, "id", json_null_create());