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_transact_block(struct jsonrpc *rpc, struct jsonrpc_msg *request,
266 struct jsonrpc_msg **replyp)
268 struct jsonrpc_msg *reply = NULL;
272 id = json_clone(request->id);
273 error = jsonrpc_send_block(rpc, request);
276 error = jsonrpc_recv_block(rpc, &reply);
278 || (reply->type == JSONRPC_REPLY
279 && json_equal(id, reply->id))) {
282 jsonrpc_msg_destroy(reply);
285 *replyp = error ? NULL : reply;
291 jsonrpc_received(struct jsonrpc *rpc)
293 struct jsonrpc_msg *msg;
297 json = json_parser_finish(rpc->parser);
299 if (json->type == JSON_STRING) {
300 VLOG_WARN_RL(&rl, "%s: error parsing stream: %s",
301 rpc->name, json_string(json));
302 jsonrpc_error(rpc, EPROTO);
307 error = jsonrpc_msg_from_json(json, &msg);
309 VLOG_WARN_RL(&rl, "%s: received bad JSON-RPC message: %s",
312 jsonrpc_error(rpc, EPROTO);
320 jsonrpc_error(struct jsonrpc *rpc, int error)
325 jsonrpc_cleanup(rpc);
330 jsonrpc_cleanup(struct jsonrpc *rpc)
332 stream_close(rpc->stream);
335 json_parser_abort(rpc->parser);
338 jsonrpc_msg_destroy(rpc->received);
339 rpc->received = NULL;
341 queue_clear(&rpc->output);
345 static struct jsonrpc_msg *
346 jsonrpc_create(enum jsonrpc_msg_type type, const char *method,
347 struct json *params, struct json *result, struct json *error,
350 struct jsonrpc_msg *msg = xmalloc(sizeof *msg);
352 msg->method = method ? xstrdup(method) : NULL;
353 msg->params = params;
354 msg->result = result;
361 jsonrpc_create_id(void)
363 static unsigned int id;
364 return json_integer_create(id++);
368 jsonrpc_create_request(const char *method, struct json *params)
370 return jsonrpc_create(JSONRPC_REQUEST, method, params, NULL, NULL,
371 jsonrpc_create_id());
375 jsonrpc_create_notify(const char *method, struct json *params)
377 return jsonrpc_create(JSONRPC_NOTIFY, method, params, NULL, NULL, NULL);
381 jsonrpc_create_reply(struct json *result, const struct json *id)
383 return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, result, NULL,
388 jsonrpc_create_error(struct json *error, const struct json *id)
390 return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, NULL, error,
395 jsonrpc_msg_type_to_string(enum jsonrpc_msg_type type)
398 case JSONRPC_REQUEST:
402 return "notification";
414 jsonrpc_msg_is_valid(const struct jsonrpc_msg *m)
416 const char *type_name;
417 unsigned int pattern;
419 if (m->params && m->params->type != JSON_ARRAY) {
420 return xstrdup("\"params\" must be JSON array");
424 case JSONRPC_REQUEST:
441 return xasprintf("invalid JSON-RPC message type %d", m->type);
444 type_name = jsonrpc_msg_type_to_string(m->type);
445 if ((m->method != NULL) != ((pattern & 0x10000) != 0)) {
446 return xasprintf("%s must%s have \"method\"",
447 type_name, (pattern & 0x10000) ? "" : " not");
450 if ((m->params != NULL) != ((pattern & 0x1000) != 0)) {
451 return xasprintf("%s must%s have \"params\"",
452 type_name, (pattern & 0x1000) ? "" : " not");
455 if ((m->result != NULL) != ((pattern & 0x100) != 0)) {
456 return xasprintf("%s must%s have \"result\"",
457 type_name, (pattern & 0x100) ? "" : " not");
460 if ((m->error != NULL) != ((pattern & 0x10) != 0)) {
461 return xasprintf("%s must%s have \"error\"",
462 type_name, (pattern & 0x10) ? "" : " not");
465 if ((m->id != NULL) != ((pattern & 0x1) != 0)) {
466 return xasprintf("%s must%s have \"id\"",
467 type_name, (pattern & 0x1) ? "" : " not");
474 jsonrpc_msg_destroy(struct jsonrpc_msg *m)
478 json_destroy(m->params);
479 json_destroy(m->result);
480 json_destroy(m->error);
487 null_from_json_null(struct json *json)
489 if (json && json->type == JSON_NULL) {
497 jsonrpc_msg_from_json(struct json *json, struct jsonrpc_msg **msgp)
499 struct json *method = NULL;
500 struct jsonrpc_msg *msg = NULL;
501 struct shash *object;
504 if (json->type != JSON_OBJECT) {
505 error = xstrdup("message is not a JSON object");
508 object = json_object(json);
510 method = shash_find_and_delete(object, "method");
511 if (method && method->type != JSON_STRING) {
512 error = xstrdup("method is not a JSON string");
516 msg = xzalloc(sizeof *msg);
517 msg->method = method ? xstrdup(method->u.string) : NULL;
518 msg->params = null_from_json_null(shash_find_and_delete(object, "params"));
519 msg->result = null_from_json_null(shash_find_and_delete(object, "result"));
520 msg->error = null_from_json_null(shash_find_and_delete(object, "error"));
521 msg->id = null_from_json_null(shash_find_and_delete(object, "id"));
522 msg->type = (msg->result ? JSONRPC_REPLY
523 : msg->error ? JSONRPC_ERROR
524 : msg->id ? JSONRPC_REQUEST
526 if (!shash_is_empty(object)) {
527 error = xasprintf("message has unexpected member \"%s\"",
528 shash_first(object)->name);
531 error = jsonrpc_msg_is_valid(msg);
537 json_destroy(method);
540 jsonrpc_msg_destroy(msg);
548 jsonrpc_msg_to_json(struct jsonrpc_msg *m)
550 struct json *json = json_object_create();
553 json_object_put(json, "method", json_string_create_nocopy(m->method));
557 json_object_put(json, "params", m->params);
561 json_object_put(json, "result", m->result);
562 } else if (m->type == JSONRPC_ERROR) {
563 json_object_put(json, "result", json_null_create());
567 json_object_put(json, "error", m->error);
568 } else if (m->type == JSONRPC_REPLY) {
569 json_object_put(json, "error", json_null_create());
573 json_object_put(json, "id", m->id);
574 } else if (m->type == JSONRPC_NOTIFY) {
575 json_object_put(json, "id", json_null_create());