]>
git.proxmox.com Git - mirror_ovs.git/blob - lib/jsonrpc.c
2 * Copyright (c) 2009, 2010 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.
25 #include "dynamic-string.h"
29 #include "poll-loop.h"
31 #include "reconnect.h"
35 #define THIS_MODULE VLM_jsonrpc
39 struct stream
*stream
;
45 struct json_parser
*parser
;
46 struct jsonrpc_msg
*received
;
49 struct ovs_queue output
;
53 /* Rate limit for error messages. */
54 static struct vlog_rate_limit rl
= VLOG_RATE_LIMIT_INIT(5, 5);
56 static void jsonrpc_received(struct jsonrpc
*);
57 static void jsonrpc_cleanup(struct jsonrpc
*);
60 jsonrpc_open(struct stream
*stream
)
64 assert(stream
!= NULL
);
66 rpc
= xzalloc(sizeof *rpc
);
67 rpc
->name
= xstrdup(stream_get_name(stream
));
69 byteq_init(&rpc
->input
);
70 queue_init(&rpc
->output
);
76 jsonrpc_close(struct jsonrpc
*rpc
)
86 jsonrpc_run(struct jsonrpc
*rpc
)
92 stream_run(rpc
->stream
);
93 while (!queue_is_empty(&rpc
->output
)) {
94 struct ofpbuf
*buf
= rpc
->output
.head
;
97 retval
= stream_send(rpc
->stream
, buf
->data
, buf
->size
);
99 rpc
->backlog
-= retval
;
100 ofpbuf_pull(buf
, retval
);
102 ofpbuf_delete(queue_pop_head(&rpc
->output
));
105 if (retval
!= -EAGAIN
) {
106 VLOG_WARN_RL(&rl
, "%s: send error: %s",
107 rpc
->name
, strerror(-retval
));
108 jsonrpc_error(rpc
, -retval
);
116 jsonrpc_wait(struct jsonrpc
*rpc
)
119 stream_run_wait(rpc
->stream
);
120 if (!queue_is_empty(&rpc
->output
)) {
121 stream_send_wait(rpc
->stream
);
127 jsonrpc_get_status(const struct jsonrpc
*rpc
)
133 jsonrpc_get_backlog(const struct jsonrpc
*rpc
)
135 return rpc
->status
? 0 : rpc
->backlog
;
139 jsonrpc_get_name(const struct jsonrpc
*rpc
)
145 jsonrpc_log_msg(const struct jsonrpc
*rpc
, const char *title
,
146 const struct jsonrpc_msg
*msg
)
148 if (VLOG_IS_DBG_ENABLED()) {
149 struct ds s
= DS_EMPTY_INITIALIZER
;
151 ds_put_format(&s
, ", method=\"%s\"", msg
->method
);
154 ds_put_cstr(&s
, ", params=");
155 json_to_ds(msg
->params
, 0, &s
);
158 ds_put_cstr(&s
, ", result=");
159 json_to_ds(msg
->result
, 0, &s
);
162 ds_put_cstr(&s
, ", error=");
163 json_to_ds(msg
->error
, 0, &s
);
166 ds_put_cstr(&s
, ", id=");
167 json_to_ds(msg
->id
, 0, &s
);
169 VLOG_DBG("%s: %s %s%s", rpc
->name
, title
,
170 jsonrpc_msg_type_to_string(msg
->type
), ds_cstr(&s
));
175 /* Always takes ownership of 'msg', regardless of success. */
177 jsonrpc_send(struct jsonrpc
*rpc
, struct jsonrpc_msg
*msg
)
185 jsonrpc_msg_destroy(msg
);
189 jsonrpc_log_msg(rpc
, "send", msg
);
191 json
= jsonrpc_msg_to_json(msg
);
192 s
= json_to_string(json
, 0);
196 buf
= xmalloc(sizeof *buf
);
197 ofpbuf_use(buf
, s
, length
);
199 queue_push_tail(&rpc
->output
, buf
);
200 rpc
->backlog
+= length
;
202 if (rpc
->output
.n
== 1) {
209 jsonrpc_recv(struct jsonrpc
*rpc
, struct jsonrpc_msg
**msgp
)
216 while (!rpc
->received
) {
217 if (byteq_is_empty(&rpc
->input
)) {
221 chunk
= byteq_headroom(&rpc
->input
);
222 retval
= stream_recv(rpc
->stream
, byteq_head(&rpc
->input
), chunk
);
224 if (retval
== -EAGAIN
) {
227 VLOG_WARN_RL(&rl
, "%s: receive error: %s",
228 rpc
->name
, strerror(-retval
));
229 jsonrpc_error(rpc
, -retval
);
232 } else if (retval
== 0) {
233 VLOG_INFO_RL(&rl
, "%s: connection closed", rpc
->name
);
234 jsonrpc_error(rpc
, EOF
);
237 byteq_advance_head(&rpc
->input
, retval
);
242 rpc
->parser
= json_parser_create(0);
244 n
= byteq_tailroom(&rpc
->input
);
245 used
= json_parser_feed(rpc
->parser
,
246 (char *) byteq_tail(&rpc
->input
), n
);
247 byteq_advance_tail(&rpc
->input
, used
);
248 if (json_parser_is_done(rpc
->parser
)) {
249 jsonrpc_received(rpc
);
257 *msgp
= rpc
->received
;
258 rpc
->received
= NULL
;
263 jsonrpc_recv_wait(struct jsonrpc
*rpc
)
265 if (rpc
->status
|| rpc
->received
|| !byteq_is_empty(&rpc
->input
)) {
266 poll_immediate_wake();
268 stream_recv_wait(rpc
->stream
);
272 /* Always takes ownership of 'msg', regardless of success. */
274 jsonrpc_send_block(struct jsonrpc
*rpc
, struct jsonrpc_msg
*msg
)
278 error
= jsonrpc_send(rpc
, msg
);
285 if (queue_is_empty(&rpc
->output
) || rpc
->status
) {
294 jsonrpc_recv_block(struct jsonrpc
*rpc
, struct jsonrpc_msg
**msgp
)
297 int error
= jsonrpc_recv(rpc
, msgp
);
298 if (error
!= EAGAIN
) {
304 jsonrpc_recv_wait(rpc
);
309 /* Always takes ownership of 'request', regardless of success. */
311 jsonrpc_transact_block(struct jsonrpc
*rpc
, struct jsonrpc_msg
*request
,
312 struct jsonrpc_msg
**replyp
)
314 struct jsonrpc_msg
*reply
= NULL
;
318 id
= json_clone(request
->id
);
319 error
= jsonrpc_send_block(rpc
, request
);
322 error
= jsonrpc_recv_block(rpc
, &reply
);
324 || (reply
->type
== JSONRPC_REPLY
325 && json_equal(id
, reply
->id
))) {
328 jsonrpc_msg_destroy(reply
);
331 *replyp
= error
? NULL
: reply
;
337 jsonrpc_received(struct jsonrpc
*rpc
)
339 struct jsonrpc_msg
*msg
;
343 json
= json_parser_finish(rpc
->parser
);
345 if (json
->type
== JSON_STRING
) {
346 VLOG_WARN_RL(&rl
, "%s: error parsing stream: %s",
347 rpc
->name
, json_string(json
));
348 jsonrpc_error(rpc
, EPROTO
);
353 error
= jsonrpc_msg_from_json(json
, &msg
);
355 VLOG_WARN_RL(&rl
, "%s: received bad JSON-RPC message: %s",
358 jsonrpc_error(rpc
, EPROTO
);
362 jsonrpc_log_msg(rpc
, "received", msg
);
367 jsonrpc_error(struct jsonrpc
*rpc
, int error
)
372 jsonrpc_cleanup(rpc
);
377 jsonrpc_cleanup(struct jsonrpc
*rpc
)
379 stream_close(rpc
->stream
);
382 json_parser_abort(rpc
->parser
);
385 jsonrpc_msg_destroy(rpc
->received
);
386 rpc
->received
= NULL
;
388 queue_clear(&rpc
->output
);
392 static struct jsonrpc_msg
*
393 jsonrpc_create(enum jsonrpc_msg_type type
, const char *method
,
394 struct json
*params
, struct json
*result
, struct json
*error
,
397 struct jsonrpc_msg
*msg
= xmalloc(sizeof *msg
);
399 msg
->method
= method
? xstrdup(method
) : NULL
;
400 msg
->params
= params
;
401 msg
->result
= result
;
408 jsonrpc_create_id(void)
410 static unsigned int id
;
411 return json_integer_create(id
++);
415 jsonrpc_create_request(const char *method
, struct json
*params
,
418 struct json
*id
= jsonrpc_create_id();
420 *idp
= json_clone(id
);
422 return jsonrpc_create(JSONRPC_REQUEST
, method
, params
, NULL
, NULL
, id
);
426 jsonrpc_create_notify(const char *method
, struct json
*params
)
428 return jsonrpc_create(JSONRPC_NOTIFY
, method
, params
, NULL
, NULL
, NULL
);
432 jsonrpc_create_reply(struct json
*result
, const struct json
*id
)
434 return jsonrpc_create(JSONRPC_REPLY
, NULL
, NULL
, result
, NULL
,
439 jsonrpc_create_error(struct json
*error
, const struct json
*id
)
441 return jsonrpc_create(JSONRPC_REPLY
, NULL
, NULL
, NULL
, error
,
446 jsonrpc_msg_type_to_string(enum jsonrpc_msg_type type
)
449 case JSONRPC_REQUEST
:
453 return "notification";
465 jsonrpc_msg_is_valid(const struct jsonrpc_msg
*m
)
467 const char *type_name
;
468 unsigned int pattern
;
470 if (m
->params
&& m
->params
->type
!= JSON_ARRAY
) {
471 return xstrdup("\"params\" must be JSON array");
475 case JSONRPC_REQUEST
:
492 return xasprintf("invalid JSON-RPC message type %d", m
->type
);
495 type_name
= jsonrpc_msg_type_to_string(m
->type
);
496 if ((m
->method
!= NULL
) != ((pattern
& 0x10000) != 0)) {
497 return xasprintf("%s must%s have \"method\"",
498 type_name
, (pattern
& 0x10000) ? "" : " not");
501 if ((m
->params
!= NULL
) != ((pattern
& 0x1000) != 0)) {
502 return xasprintf("%s must%s have \"params\"",
503 type_name
, (pattern
& 0x1000) ? "" : " not");
506 if ((m
->result
!= NULL
) != ((pattern
& 0x100) != 0)) {
507 return xasprintf("%s must%s have \"result\"",
508 type_name
, (pattern
& 0x100) ? "" : " not");
511 if ((m
->error
!= NULL
) != ((pattern
& 0x10) != 0)) {
512 return xasprintf("%s must%s have \"error\"",
513 type_name
, (pattern
& 0x10) ? "" : " not");
516 if ((m
->id
!= NULL
) != ((pattern
& 0x1) != 0)) {
517 return xasprintf("%s must%s have \"id\"",
518 type_name
, (pattern
& 0x1) ? "" : " not");
525 jsonrpc_msg_destroy(struct jsonrpc_msg
*m
)
529 json_destroy(m
->params
);
530 json_destroy(m
->result
);
531 json_destroy(m
->error
);
538 null_from_json_null(struct json
*json
)
540 if (json
&& json
->type
== JSON_NULL
) {
548 jsonrpc_msg_from_json(struct json
*json
, struct jsonrpc_msg
**msgp
)
550 struct json
*method
= NULL
;
551 struct jsonrpc_msg
*msg
= NULL
;
552 struct shash
*object
;
555 if (json
->type
!= JSON_OBJECT
) {
556 error
= xstrdup("message is not a JSON object");
559 object
= json_object(json
);
561 method
= shash_find_and_delete(object
, "method");
562 if (method
&& method
->type
!= JSON_STRING
) {
563 error
= xstrdup("method is not a JSON string");
567 msg
= xzalloc(sizeof *msg
);
568 msg
->method
= method
? xstrdup(method
->u
.string
) : NULL
;
569 msg
->params
= null_from_json_null(shash_find_and_delete(object
, "params"));
570 msg
->result
= null_from_json_null(shash_find_and_delete(object
, "result"));
571 msg
->error
= null_from_json_null(shash_find_and_delete(object
, "error"));
572 msg
->id
= null_from_json_null(shash_find_and_delete(object
, "id"));
573 msg
->type
= (msg
->result
? JSONRPC_REPLY
574 : msg
->error
? JSONRPC_ERROR
575 : msg
->id
? JSONRPC_REQUEST
577 if (!shash_is_empty(object
)) {
578 error
= xasprintf("message has unexpected member \"%s\"",
579 shash_first(object
)->name
);
582 error
= jsonrpc_msg_is_valid(msg
);
588 json_destroy(method
);
591 jsonrpc_msg_destroy(msg
);
599 jsonrpc_msg_to_json(struct jsonrpc_msg
*m
)
601 struct json
*json
= json_object_create();
604 json_object_put(json
, "method", json_string_create_nocopy(m
->method
));
608 json_object_put(json
, "params", m
->params
);
612 json_object_put(json
, "result", m
->result
);
613 } else if (m
->type
== JSONRPC_ERROR
) {
614 json_object_put(json
, "result", json_null_create());
618 json_object_put(json
, "error", m
->error
);
619 } else if (m
->type
== JSONRPC_REPLY
) {
620 json_object_put(json
, "error", json_null_create());
624 json_object_put(json
, "id", m
->id
);
625 } else if (m
->type
== JSONRPC_NOTIFY
) {
626 json_object_put(json
, "id", json_null_create());
634 /* A JSON-RPC session with reconnection. */
636 struct jsonrpc_session
{
637 struct reconnect
*reconnect
;
639 struct stream
*stream
;
643 /* Creates and returns a jsonrpc_session that connects and reconnects, with
644 * back-off, to 'name', which should be a string acceptable to
646 struct jsonrpc_session
*
647 jsonrpc_session_open(const char *name
)
649 struct jsonrpc_session
*s
;
651 s
= xmalloc(sizeof *s
);
652 s
->reconnect
= reconnect_create(time_msec());
653 reconnect_set_name(s
->reconnect
, name
);
654 reconnect_enable(s
->reconnect
, time_msec());
662 /* Creates and returns a jsonrpc_session that is initially connected to
663 * 'jsonrpc'. If the connection is dropped, it will not be reconnected. */
664 struct jsonrpc_session
*
665 jsonrpc_session_open_unreliably(struct jsonrpc
*jsonrpc
)
667 struct jsonrpc_session
*s
;
669 s
= xmalloc(sizeof *s
);
670 s
->reconnect
= reconnect_create(time_msec());
671 reconnect_set_name(s
->reconnect
, jsonrpc_get_name(jsonrpc
));
672 reconnect_set_max_tries(s
->reconnect
, 0);
673 reconnect_connected(s
->reconnect
, time_msec());
682 jsonrpc_session_close(struct jsonrpc_session
*s
)
685 jsonrpc_close(s
->rpc
);
686 reconnect_destroy(s
->reconnect
);
692 jsonrpc_session_disconnect(struct jsonrpc_session
*s
)
695 jsonrpc_error(s
->rpc
, EOF
);
696 jsonrpc_close(s
->rpc
);
699 } else if (s
->stream
) {
700 stream_close(s
->stream
);
707 jsonrpc_session_connect(struct jsonrpc_session
*s
)
711 jsonrpc_session_disconnect(s
);
712 error
= stream_open(reconnect_get_name(s
->reconnect
), &s
->stream
);
714 reconnect_connect_failed(s
->reconnect
, time_msec(), error
);
716 reconnect_connecting(s
->reconnect
, time_msec());
722 jsonrpc_session_run(struct jsonrpc_session
*s
)
728 error
= jsonrpc_get_status(s
->rpc
);
730 reconnect_disconnected(s
->reconnect
, time_msec(), 0);
731 jsonrpc_session_disconnect(s
);
733 } else if (s
->stream
) {
736 stream_run(s
->stream
);
737 error
= stream_connect(s
->stream
);
739 reconnect_connected(s
->reconnect
, time_msec());
740 s
->rpc
= jsonrpc_open(s
->stream
);
742 } else if (error
!= EAGAIN
) {
743 reconnect_connect_failed(s
->reconnect
, time_msec(), error
);
744 stream_close(s
->stream
);
749 switch (reconnect_run(s
->reconnect
, time_msec())) {
750 case RECONNECT_CONNECT
:
751 jsonrpc_session_connect(s
);
754 case RECONNECT_DISCONNECT
:
755 reconnect_disconnected(s
->reconnect
, time_msec(), 0);
756 jsonrpc_session_disconnect(s
);
759 case RECONNECT_PROBE
:
762 struct jsonrpc_msg
*request
;
764 params
= json_array_create_empty();
765 request
= jsonrpc_create_request("echo", params
, NULL
);
766 json_destroy(request
->id
);
767 request
->id
= json_string_create("echo");
768 jsonrpc_send(s
->rpc
, request
);
775 jsonrpc_session_wait(struct jsonrpc_session
*s
)
778 jsonrpc_wait(s
->rpc
);
779 } else if (s
->stream
) {
780 stream_run_wait(s
->stream
);
781 stream_connect_wait(s
->stream
);
783 reconnect_wait(s
->reconnect
, time_msec());
787 jsonrpc_session_get_backlog(const struct jsonrpc_session
*s
)
789 return s
->rpc
? jsonrpc_get_backlog(s
->rpc
) : 0;
793 jsonrpc_session_get_name(const struct jsonrpc_session
*s
)
795 return reconnect_get_name(s
->reconnect
);
798 /* Always takes ownership of 'msg', regardless of success. */
800 jsonrpc_session_send(struct jsonrpc_session
*s
, struct jsonrpc_msg
*msg
)
803 return jsonrpc_send(s
->rpc
, msg
);
805 jsonrpc_msg_destroy(msg
);
811 jsonrpc_session_recv(struct jsonrpc_session
*s
)
814 struct jsonrpc_msg
*msg
;
815 jsonrpc_recv(s
->rpc
, &msg
);
817 reconnect_received(s
->reconnect
, time_msec());
818 if (msg
->type
== JSONRPC_REQUEST
&& !strcmp(msg
->method
, "echo")) {
819 /* Echo request. Send reply. */
820 struct jsonrpc_msg
*reply
;
822 reply
= jsonrpc_create_reply(json_clone(msg
->params
), msg
->id
);
823 jsonrpc_session_send(s
, reply
);
824 } else if (msg
->type
== JSONRPC_REPLY
825 && msg
->id
&& msg
->id
->type
== JSON_STRING
826 && !strcmp(msg
->id
->u
.string
, "echo")) {
827 /* It's a reply to our echo request. Suppress it. */
831 jsonrpc_msg_destroy(msg
);
838 jsonrpc_session_recv_wait(struct jsonrpc_session
*s
)
841 jsonrpc_recv_wait(s
->rpc
);
846 jsonrpc_session_is_alive(const struct jsonrpc_session
*s
)
848 return s
->rpc
|| s
->stream
|| reconnect_get_max_tries(s
->reconnect
);
852 jsonrpc_session_is_connected(const struct jsonrpc_session
*s
)
854 return s
->rpc
!= NULL
;
858 jsonrpc_session_get_seqno(const struct jsonrpc_session
*s
)
864 jsonrpc_session_force_reconnect(struct jsonrpc_session
*s
)
866 reconnect_force_reconnect(s
->reconnect
, time_msec());