ovsdb-idl: Improve check in ovsdb_idl_row_is_orphan().
[openvswitch] / lib / jsonrpc.c
1 /*
2  * Copyright (c) 2009, 2010 Nicira Networks.
3  *
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:
7  *
8  *     http://www.apache.org/licenses/LICENSE-2.0
9  *
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.
15  */
16
17 #include <config.h>
18
19 #include "jsonrpc.h"
20
21 #include <assert.h>
22 #include <errno.h>
23
24 #include "byteq.h"
25 #include "dynamic-string.h"
26 #include "json.h"
27 #include "list.h"
28 #include "ofpbuf.h"
29 #include "poll-loop.h"
30 #include "queue.h"
31 #include "reconnect.h"
32 #include "stream.h"
33 #include "timeval.h"
34
35 #define THIS_MODULE VLM_jsonrpc
36 #include "vlog.h"
37 \f
38 struct jsonrpc {
39     struct stream *stream;
40     char *name;
41     int status;
42
43     /* Input. */
44     struct byteq input;
45     struct json_parser *parser;
46     struct jsonrpc_msg *received;
47
48     /* Output. */
49     struct ovs_queue output;
50     size_t backlog;
51 };
52
53 /* Rate limit for error messages. */
54 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
55
56 static void jsonrpc_received(struct jsonrpc *);
57 static void jsonrpc_cleanup(struct jsonrpc *);
58
59 struct jsonrpc *
60 jsonrpc_open(struct stream *stream)
61 {
62     struct jsonrpc *rpc;
63
64     assert(stream != NULL);
65
66     rpc = xzalloc(sizeof *rpc);
67     rpc->name = xstrdup(stream_get_name(stream));
68     rpc->stream = stream;
69     byteq_init(&rpc->input);
70     queue_init(&rpc->output);
71
72     return rpc;
73 }
74
75 void
76 jsonrpc_close(struct jsonrpc *rpc)
77 {
78     if (rpc) {
79         jsonrpc_cleanup(rpc);
80         free(rpc->name);
81         free(rpc);
82     }
83 }
84
85 void
86 jsonrpc_run(struct jsonrpc *rpc)
87 {
88     if (rpc->status) {
89         return;
90     }
91
92     stream_run(rpc->stream);
93     while (!queue_is_empty(&rpc->output)) {
94         struct ofpbuf *buf = rpc->output.head;
95         int retval;
96
97         retval = stream_send(rpc->stream, buf->data, buf->size);
98         if (retval >= 0) {
99             rpc->backlog -= retval;
100             ofpbuf_pull(buf, retval);
101             if (!buf->size) {
102                 ofpbuf_delete(queue_pop_head(&rpc->output));
103             }
104         } else {
105             if (retval != -EAGAIN) {
106                 VLOG_WARN_RL(&rl, "%s: send error: %s",
107                              rpc->name, strerror(-retval));
108                 jsonrpc_error(rpc, -retval);
109             }
110             break;
111         }
112     }
113 }
114
115 void
116 jsonrpc_wait(struct jsonrpc *rpc)
117 {
118     if (!rpc->status) {
119         stream_run_wait(rpc->stream);
120         if (!queue_is_empty(&rpc->output)) {
121             stream_send_wait(rpc->stream);
122         }
123     }
124 }
125
126 int
127 jsonrpc_get_status(const struct jsonrpc *rpc)
128 {
129     return rpc->status;
130 }
131
132 size_t
133 jsonrpc_get_backlog(const struct jsonrpc *rpc)
134 {
135     return rpc->status ? 0 : rpc->backlog;
136 }
137
138 const char *
139 jsonrpc_get_name(const struct jsonrpc *rpc)
140 {
141     return rpc->name;
142 }
143
144 static void
145 jsonrpc_log_msg(const struct jsonrpc *rpc, const char *title,
146                 const struct jsonrpc_msg *msg)
147 {
148     if (VLOG_IS_DBG_ENABLED()) {
149         struct ds s = DS_EMPTY_INITIALIZER;
150         if (msg->method) {
151             ds_put_format(&s, ", method=\"%s\"", msg->method);
152         }
153         if (msg->params) {
154             ds_put_cstr(&s, ", params=");
155             json_to_ds(msg->params, 0, &s);
156         }
157         if (msg->result) {
158             ds_put_cstr(&s, ", result=");
159             json_to_ds(msg->result, 0, &s);
160         }
161         if (msg->error) {
162             ds_put_cstr(&s, ", error=");
163             json_to_ds(msg->error, 0, &s);
164         }
165         if (msg->id) {
166             ds_put_cstr(&s, ", id=");
167             json_to_ds(msg->id, 0, &s);
168         }
169         VLOG_DBG("%s: %s %s%s", rpc->name, title,
170                  jsonrpc_msg_type_to_string(msg->type), ds_cstr(&s));
171         ds_destroy(&s);
172     }
173 }
174
175 /* Always takes ownership of 'msg', regardless of success. */
176 int
177 jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
178 {
179     struct ofpbuf *buf;
180     struct json *json;
181     size_t length;
182     char *s;
183
184     if (rpc->status) {
185         jsonrpc_msg_destroy(msg);
186         return rpc->status;
187     }
188
189     jsonrpc_log_msg(rpc, "send", msg);
190
191     json = jsonrpc_msg_to_json(msg);
192     s = json_to_string(json, 0);
193     length = strlen(s);
194     json_destroy(json);
195
196     buf = xmalloc(sizeof *buf);
197     ofpbuf_use(buf, s, length);
198     buf->size = length;
199     queue_push_tail(&rpc->output, buf);
200     rpc->backlog += length;
201
202     if (rpc->output.n == 1) {
203         jsonrpc_run(rpc);
204     }
205     return rpc->status;
206 }
207
208 int
209 jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
210 {
211     *msgp = NULL;
212     if (rpc->status) {
213         return rpc->status;
214     }
215
216     while (!rpc->received) {
217         if (byteq_is_empty(&rpc->input)) {
218             size_t chunk;
219             int retval;
220
221             chunk = byteq_headroom(&rpc->input);
222             retval = stream_recv(rpc->stream, byteq_head(&rpc->input), chunk);
223             if (retval < 0) {
224                 if (retval == -EAGAIN) {
225                     return EAGAIN;
226                 } else {
227                     VLOG_WARN_RL(&rl, "%s: receive error: %s",
228                                  rpc->name, strerror(-retval));
229                     jsonrpc_error(rpc, -retval);
230                     return rpc->status;
231                 }
232             } else if (retval == 0) {
233                 VLOG_INFO_RL(&rl, "%s: connection closed", rpc->name);
234                 jsonrpc_error(rpc, EOF);
235                 return EOF;
236             }
237             byteq_advance_head(&rpc->input, retval);
238         } else {
239             size_t n, used;
240
241             if (!rpc->parser) {
242                 rpc->parser = json_parser_create(0);
243             }
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);
250                 if (rpc->status) {
251                     return rpc->status;
252                 }
253             }
254         }
255     }
256
257     *msgp = rpc->received;
258     rpc->received = NULL;
259     return 0;
260 }
261
262 void
263 jsonrpc_recv_wait(struct jsonrpc *rpc)
264 {
265     if (rpc->status || rpc->received || !byteq_is_empty(&rpc->input)) {
266         poll_immediate_wake();
267     } else {
268         stream_recv_wait(rpc->stream);
269     }
270 }
271
272 /* Always takes ownership of 'msg', regardless of success. */
273 int
274 jsonrpc_send_block(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
275 {
276     int error;
277
278     error = jsonrpc_send(rpc, msg);
279     if (error) {
280         return error;
281     }
282
283     for (;;) {
284         jsonrpc_run(rpc);
285         if (queue_is_empty(&rpc->output) || rpc->status) {
286             return rpc->status;
287         }
288         jsonrpc_wait(rpc);
289         poll_block();
290     }
291 }
292
293 int
294 jsonrpc_recv_block(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
295 {
296     for (;;) {
297         int error = jsonrpc_recv(rpc, msgp);
298         if (error != EAGAIN) {
299             return error;
300         }
301
302         jsonrpc_run(rpc);
303         jsonrpc_wait(rpc);
304         jsonrpc_recv_wait(rpc);
305         poll_block();
306     }
307 }
308
309 /* Always takes ownership of 'request', regardless of success. */
310 int
311 jsonrpc_transact_block(struct jsonrpc *rpc, struct jsonrpc_msg *request,
312                        struct jsonrpc_msg **replyp)
313 {
314     struct jsonrpc_msg *reply = NULL;
315     struct json *id;
316     int error;
317
318     id = json_clone(request->id);
319     error = jsonrpc_send_block(rpc, request);
320     if (!error) {
321         for (;;) {
322             error = jsonrpc_recv_block(rpc, &reply);
323             if (error
324                 || (reply->type == JSONRPC_REPLY
325                     && json_equal(id, reply->id))) {
326                 break;
327             }
328             jsonrpc_msg_destroy(reply);
329         }
330     }
331     *replyp = error ? NULL : reply;
332     json_destroy(id);
333     return error;
334 }
335
336 static void
337 jsonrpc_received(struct jsonrpc *rpc)
338 {
339     struct jsonrpc_msg *msg;
340     struct json *json;
341     char *error;
342
343     json = json_parser_finish(rpc->parser);
344     rpc->parser = NULL;
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);
349         json_destroy(json);
350         return;
351     }
352
353     error = jsonrpc_msg_from_json(json, &msg);
354     if (error) {
355         VLOG_WARN_RL(&rl, "%s: received bad JSON-RPC message: %s",
356                      rpc->name, error);
357         free(error);
358         jsonrpc_error(rpc, EPROTO);
359         return;
360     }
361
362     jsonrpc_log_msg(rpc, "received", msg);
363     rpc->received = msg;
364 }
365
366 void
367 jsonrpc_error(struct jsonrpc *rpc, int error)
368 {
369     assert(error);
370     if (!rpc->status) {
371         rpc->status = error;
372         jsonrpc_cleanup(rpc);
373     }
374 }
375
376 static void
377 jsonrpc_cleanup(struct jsonrpc *rpc)
378 {
379     stream_close(rpc->stream);
380     rpc->stream = NULL;
381
382     json_parser_abort(rpc->parser);
383     rpc->parser = NULL;
384
385     jsonrpc_msg_destroy(rpc->received);
386     rpc->received = NULL;
387
388     queue_clear(&rpc->output);
389     rpc->backlog = 0;
390 }
391 \f
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,
395                 struct json *id)
396 {
397     struct jsonrpc_msg *msg = xmalloc(sizeof *msg);
398     msg->type = type;
399     msg->method = method ? xstrdup(method) : NULL;
400     msg->params = params;
401     msg->result = result;
402     msg->error = error;
403     msg->id = id;
404     return msg;
405 }
406
407 static struct json *
408 jsonrpc_create_id(void)
409 {
410     static unsigned int id;
411     return json_integer_create(id++);
412 }
413
414 struct jsonrpc_msg *
415 jsonrpc_create_request(const char *method, struct json *params,
416                        struct json **idp)
417 {
418     struct json *id = jsonrpc_create_id();
419     if (idp) {
420         *idp = json_clone(id);
421     }
422     return jsonrpc_create(JSONRPC_REQUEST, method, params, NULL, NULL, id);
423 }
424
425 struct jsonrpc_msg *
426 jsonrpc_create_notify(const char *method, struct json *params)
427 {
428     return jsonrpc_create(JSONRPC_NOTIFY, method, params, NULL, NULL, NULL);
429 }
430
431 struct jsonrpc_msg *
432 jsonrpc_create_reply(struct json *result, const struct json *id)
433 {
434     return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, result, NULL,
435                            json_clone(id));
436 }
437
438 struct jsonrpc_msg *
439 jsonrpc_create_error(struct json *error, const struct json *id)
440 {
441     return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, NULL, error,
442                            json_clone(id));
443 }
444
445 const char *
446 jsonrpc_msg_type_to_string(enum jsonrpc_msg_type type)
447 {
448     switch (type) {
449     case JSONRPC_REQUEST:
450         return "request";
451
452     case JSONRPC_NOTIFY:
453         return "notification";
454
455     case JSONRPC_REPLY:
456         return "reply";
457
458     case JSONRPC_ERROR:
459         return "error";
460     }
461     return "(null)";
462 }
463
464 char *
465 jsonrpc_msg_is_valid(const struct jsonrpc_msg *m)
466 {
467     const char *type_name;
468     unsigned int pattern;
469
470     if (m->params && m->params->type != JSON_ARRAY) {
471         return xstrdup("\"params\" must be JSON array");
472     }
473
474     switch (m->type) {
475     case JSONRPC_REQUEST:
476         pattern = 0x11001;
477         break;
478
479     case JSONRPC_NOTIFY:
480         pattern = 0x11000;
481         break;
482
483     case JSONRPC_REPLY:
484         pattern = 0x00101;
485         break;
486
487     case JSONRPC_ERROR:
488         pattern = 0x00011;
489         break;
490
491     default:
492         return xasprintf("invalid JSON-RPC message type %d", m->type);
493     }
494
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");
499
500     }
501     if ((m->params != NULL) != ((pattern & 0x1000) != 0)) {
502         return xasprintf("%s must%s have \"params\"",
503                          type_name, (pattern & 0x1000) ? "" : " not");
504
505     }
506     if ((m->result != NULL) != ((pattern & 0x100) != 0)) {
507         return xasprintf("%s must%s have \"result\"",
508                          type_name, (pattern & 0x100) ? "" : " not");
509
510     }
511     if ((m->error != NULL) != ((pattern & 0x10) != 0)) {
512         return xasprintf("%s must%s have \"error\"",
513                          type_name, (pattern & 0x10) ? "" : " not");
514
515     }
516     if ((m->id != NULL) != ((pattern & 0x1) != 0)) {
517         return xasprintf("%s must%s have \"id\"",
518                          type_name, (pattern & 0x1) ? "" : " not");
519
520     }
521     return NULL;
522 }
523
524 void
525 jsonrpc_msg_destroy(struct jsonrpc_msg *m)
526 {
527     if (m) {
528         free(m->method);
529         json_destroy(m->params);
530         json_destroy(m->result);
531         json_destroy(m->error);
532         json_destroy(m->id);
533         free(m);
534     }
535 }
536
537 static struct json *
538 null_from_json_null(struct json *json)
539 {
540     if (json && json->type == JSON_NULL) {
541         json_destroy(json);
542         return NULL;
543     }
544     return json;
545 }
546
547 char *
548 jsonrpc_msg_from_json(struct json *json, struct jsonrpc_msg **msgp)
549 {
550     struct json *method = NULL;
551     struct jsonrpc_msg *msg = NULL;
552     struct shash *object;
553     char *error;
554
555     if (json->type != JSON_OBJECT) {
556         error = xstrdup("message is not a JSON object");
557         goto exit;
558     }
559     object = json_object(json);
560
561     method = shash_find_and_delete(object, "method");
562     if (method && method->type != JSON_STRING) {
563         error = xstrdup("method is not a JSON string");
564         goto exit;
565     }
566
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
576                  : JSONRPC_NOTIFY);
577     if (!shash_is_empty(object)) {
578         error = xasprintf("message has unexpected member \"%s\"",
579                           shash_first(object)->name);
580         goto exit;
581     }
582     error = jsonrpc_msg_is_valid(msg);
583     if (error) {
584         goto exit;
585     }
586
587 exit:
588     json_destroy(method);
589     json_destroy(json);
590     if (error) {
591         jsonrpc_msg_destroy(msg);
592         msg = NULL;
593     }
594     *msgp = msg;
595     return error;
596 }
597
598 struct json *
599 jsonrpc_msg_to_json(struct jsonrpc_msg *m)
600 {
601     struct json *json = json_object_create();
602
603     if (m->method) {
604         json_object_put(json, "method", json_string_create_nocopy(m->method));
605     }
606
607     if (m->params) {
608         json_object_put(json, "params", m->params);
609     }
610
611     if (m->result) {
612         json_object_put(json, "result", m->result);
613     } else if (m->type == JSONRPC_ERROR) {
614         json_object_put(json, "result", json_null_create());
615     }
616
617     if (m->error) {
618         json_object_put(json, "error", m->error);
619     } else if (m->type == JSONRPC_REPLY) {
620         json_object_put(json, "error", json_null_create());
621     }
622
623     if (m->id) {
624         json_object_put(json, "id", m->id);
625     } else if (m->type == JSONRPC_NOTIFY) {
626         json_object_put(json, "id", json_null_create());
627     }
628
629     free(m);
630
631     return json;
632 }
633 \f
634 /* A JSON-RPC session with reconnection. */
635
636 struct jsonrpc_session {
637     struct reconnect *reconnect;
638     struct jsonrpc *rpc;
639     struct stream *stream;
640     unsigned int seqno;
641 };
642
643 /* Creates and returns a jsonrpc_session that connects and reconnects, with
644  * back-off, to 'name', which should be a string acceptable to
645  * stream_open(). */
646 struct jsonrpc_session *
647 jsonrpc_session_open(const char *name)
648 {
649     struct jsonrpc_session *s;
650
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());
655     s->rpc = NULL;
656     s->stream = NULL;
657     s->seqno = 0;
658
659     return s;
660 }
661
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)
666 {
667     struct jsonrpc_session *s;
668
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());
674     s->rpc = jsonrpc;
675     s->stream = NULL;
676     s->seqno = 0;
677
678     return s;
679 }
680
681 void
682 jsonrpc_session_close(struct jsonrpc_session *s)
683 {
684     if (s) {
685         jsonrpc_close(s->rpc);
686         reconnect_destroy(s->reconnect);
687         free(s);
688     }
689 }
690
691 static void
692 jsonrpc_session_disconnect(struct jsonrpc_session *s)
693 {
694     if (s->rpc) {
695         jsonrpc_error(s->rpc, EOF);
696         jsonrpc_close(s->rpc);
697         s->rpc = NULL;
698         s->seqno++;
699     } else if (s->stream) {
700         stream_close(s->stream);
701         s->stream = NULL;
702         s->seqno++;
703     }
704 }
705
706 static void
707 jsonrpc_session_connect(struct jsonrpc_session *s)
708 {
709     int error;
710
711     jsonrpc_session_disconnect(s);
712     error = stream_open(reconnect_get_name(s->reconnect), &s->stream);
713     if (error) {
714         reconnect_connect_failed(s->reconnect, time_msec(), error);
715     } else {
716         reconnect_connecting(s->reconnect, time_msec());
717     }
718     s->seqno++;
719 }
720
721 void
722 jsonrpc_session_run(struct jsonrpc_session *s)
723 {
724     if (s->rpc) {
725         int error;
726
727         jsonrpc_run(s->rpc);
728         error = jsonrpc_get_status(s->rpc);
729         if (error) {
730             reconnect_disconnected(s->reconnect, time_msec(), 0);
731             jsonrpc_session_disconnect(s);
732         }
733     } else if (s->stream) {
734         int error;
735
736         stream_run(s->stream);
737         error = stream_connect(s->stream);
738         if (!error) {
739             reconnect_connected(s->reconnect, time_msec());
740             s->rpc = jsonrpc_open(s->stream);
741             s->stream = NULL;
742         } else if (error != EAGAIN) {
743             reconnect_connect_failed(s->reconnect, time_msec(), error);
744             stream_close(s->stream);
745             s->stream = NULL;
746         }
747     }
748
749     switch (reconnect_run(s->reconnect, time_msec())) {
750     case RECONNECT_CONNECT:
751         jsonrpc_session_connect(s);
752         break;
753
754     case RECONNECT_DISCONNECT:
755         reconnect_disconnected(s->reconnect, time_msec(), 0);
756         jsonrpc_session_disconnect(s);
757         break;
758
759     case RECONNECT_PROBE:
760         if (s->rpc) {
761             struct json *params;
762             struct jsonrpc_msg *request;
763
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);
769         }
770         break;
771     }
772 }
773
774 void
775 jsonrpc_session_wait(struct jsonrpc_session *s)
776 {
777     if (s->rpc) {
778         jsonrpc_wait(s->rpc);
779     } else if (s->stream) {
780         stream_run_wait(s->stream);
781         stream_connect_wait(s->stream);
782     }
783     reconnect_wait(s->reconnect, time_msec());
784 }
785
786 size_t
787 jsonrpc_session_get_backlog(const struct jsonrpc_session *s)
788 {
789     return s->rpc ? jsonrpc_get_backlog(s->rpc) : 0;
790 }
791
792 const char *
793 jsonrpc_session_get_name(const struct jsonrpc_session *s)
794 {
795     return reconnect_get_name(s->reconnect);
796 }
797
798 /* Always takes ownership of 'msg', regardless of success. */
799 int
800 jsonrpc_session_send(struct jsonrpc_session *s, struct jsonrpc_msg *msg)
801 {
802     if (s->rpc) {
803         return jsonrpc_send(s->rpc, msg);
804     } else {
805         jsonrpc_msg_destroy(msg);
806         return ENOTCONN;
807     }
808 }
809
810 struct jsonrpc_msg *
811 jsonrpc_session_recv(struct jsonrpc_session *s)
812 {
813     if (s->rpc) {
814         struct jsonrpc_msg *msg;
815         jsonrpc_recv(s->rpc, &msg);
816         if (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;
821
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. */
828             } else {
829                 return msg;
830             }
831             jsonrpc_msg_destroy(msg);
832         }
833     }
834     return NULL;
835 }
836
837 void
838 jsonrpc_session_recv_wait(struct jsonrpc_session *s)
839 {
840     if (s->rpc) {
841         jsonrpc_recv_wait(s->rpc);
842     }
843 }
844
845 bool
846 jsonrpc_session_is_alive(const struct jsonrpc_session *s)
847 {
848     return s->rpc || s->stream || reconnect_get_max_tries(s->reconnect);
849 }
850
851 bool
852 jsonrpc_session_is_connected(const struct jsonrpc_session *s)
853 {
854     return s->rpc != NULL;
855 }
856
857 unsigned int
858 jsonrpc_session_get_seqno(const struct jsonrpc_session *s)
859 {
860     return s->seqno;
861 }
862
863 void
864 jsonrpc_session_force_reconnect(struct jsonrpc_session *s)
865 {
866     reconnect_force_reconnect(s->reconnect, time_msec());
867 }