ofproto: Move hton_ofp_phy_port() to ofp-util.
[sliver-openvswitch.git] / 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 "fatal-signal.h"
27 #include "json.h"
28 #include "list.h"
29 #include "ofpbuf.h"
30 #include "poll-loop.h"
31 #include "reconnect.h"
32 #include "stream.h"
33 #include "timeval.h"
34 #include "vlog.h"
35
36 VLOG_DEFINE_THIS_MODULE(jsonrpc);
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 list output;         /* Contains "struct ofpbuf"s. */
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 /* This is just the same as stream_open() except that it uses the default
60  * JSONRPC ports if none is specified. */
61 int
62 jsonrpc_stream_open(const char *name, struct stream **streamp)
63 {
64     return stream_open_with_default_ports(name, JSONRPC_TCP_PORT,
65                                           JSONRPC_SSL_PORT, streamp);
66 }
67
68 /* This is just the same as pstream_open() except that it uses the default
69  * JSONRPC ports if none is specified. */
70 int
71 jsonrpc_pstream_open(const char *name, struct pstream **pstreamp)
72 {
73     return pstream_open_with_default_ports(name, JSONRPC_TCP_PORT,
74                                            JSONRPC_SSL_PORT, pstreamp);
75 }
76
77 struct jsonrpc *
78 jsonrpc_open(struct stream *stream)
79 {
80     struct jsonrpc *rpc;
81
82     assert(stream != NULL);
83
84     rpc = xzalloc(sizeof *rpc);
85     rpc->name = xstrdup(stream_get_name(stream));
86     rpc->stream = stream;
87     byteq_init(&rpc->input);
88     list_init(&rpc->output);
89
90     return rpc;
91 }
92
93 void
94 jsonrpc_close(struct jsonrpc *rpc)
95 {
96     if (rpc) {
97         jsonrpc_cleanup(rpc);
98         free(rpc->name);
99         free(rpc);
100     }
101 }
102
103 void
104 jsonrpc_run(struct jsonrpc *rpc)
105 {
106     if (rpc->status) {
107         return;
108     }
109
110     stream_run(rpc->stream);
111     while (!list_is_empty(&rpc->output)) {
112         struct ofpbuf *buf = ofpbuf_from_list(rpc->output.next);
113         int retval;
114
115         retval = stream_send(rpc->stream, buf->data, buf->size);
116         if (retval >= 0) {
117             rpc->backlog -= retval;
118             ofpbuf_pull(buf, retval);
119             if (!buf->size) {
120                 list_remove(&buf->list_node);
121                 ofpbuf_delete(buf);
122             }
123         } else {
124             if (retval != -EAGAIN) {
125                 VLOG_WARN_RL(&rl, "%s: send error: %s",
126                              rpc->name, strerror(-retval));
127                 jsonrpc_error(rpc, -retval);
128             }
129             break;
130         }
131     }
132 }
133
134 void
135 jsonrpc_wait(struct jsonrpc *rpc)
136 {
137     if (!rpc->status) {
138         stream_run_wait(rpc->stream);
139         if (!list_is_empty(&rpc->output)) {
140             stream_send_wait(rpc->stream);
141         }
142     }
143 }
144
145 /*
146  * Possible status values:
147  * - 0: no error yet
148  * - >0: errno value
149  * - EOF: end of file (remote end closed connection; not necessarily an error)
150  */
151 int
152 jsonrpc_get_status(const struct jsonrpc *rpc)
153 {
154     return rpc->status;
155 }
156
157 size_t
158 jsonrpc_get_backlog(const struct jsonrpc *rpc)
159 {
160     return rpc->status ? 0 : rpc->backlog;
161 }
162
163 const char *
164 jsonrpc_get_name(const struct jsonrpc *rpc)
165 {
166     return rpc->name;
167 }
168
169 static void
170 jsonrpc_log_msg(const struct jsonrpc *rpc, const char *title,
171                 const struct jsonrpc_msg *msg)
172 {
173     if (VLOG_IS_DBG_ENABLED()) {
174         struct ds s = DS_EMPTY_INITIALIZER;
175         if (msg->method) {
176             ds_put_format(&s, ", method=\"%s\"", msg->method);
177         }
178         if (msg->params) {
179             ds_put_cstr(&s, ", params=");
180             json_to_ds(msg->params, 0, &s);
181         }
182         if (msg->result) {
183             ds_put_cstr(&s, ", result=");
184             json_to_ds(msg->result, 0, &s);
185         }
186         if (msg->error) {
187             ds_put_cstr(&s, ", error=");
188             json_to_ds(msg->error, 0, &s);
189         }
190         if (msg->id) {
191             ds_put_cstr(&s, ", id=");
192             json_to_ds(msg->id, 0, &s);
193         }
194         VLOG_DBG("%s: %s %s%s", rpc->name, title,
195                  jsonrpc_msg_type_to_string(msg->type), ds_cstr(&s));
196         ds_destroy(&s);
197     }
198 }
199
200 /* Always takes ownership of 'msg', regardless of success. */
201 int
202 jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
203 {
204     struct ofpbuf *buf;
205     struct json *json;
206     size_t length;
207     char *s;
208
209     if (rpc->status) {
210         jsonrpc_msg_destroy(msg);
211         return rpc->status;
212     }
213
214     jsonrpc_log_msg(rpc, "send", msg);
215
216     json = jsonrpc_msg_to_json(msg);
217     s = json_to_string(json, 0);
218     length = strlen(s);
219     json_destroy(json);
220
221     buf = xmalloc(sizeof *buf);
222     ofpbuf_use(buf, s, length);
223     buf->size = length;
224     list_push_back(&rpc->output, &buf->list_node);
225     rpc->backlog += length;
226
227     if (rpc->backlog == length) {
228         jsonrpc_run(rpc);
229     }
230     return rpc->status;
231 }
232
233 int
234 jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
235 {
236     *msgp = NULL;
237     if (rpc->status) {
238         return rpc->status;
239     }
240
241     while (!rpc->received) {
242         if (byteq_is_empty(&rpc->input)) {
243             size_t chunk;
244             int retval;
245
246             chunk = byteq_headroom(&rpc->input);
247             retval = stream_recv(rpc->stream, byteq_head(&rpc->input), chunk);
248             if (retval < 0) {
249                 if (retval == -EAGAIN) {
250                     return EAGAIN;
251                 } else {
252                     VLOG_WARN_RL(&rl, "%s: receive error: %s",
253                                  rpc->name, strerror(-retval));
254                     jsonrpc_error(rpc, -retval);
255                     return rpc->status;
256                 }
257             } else if (retval == 0) {
258                 jsonrpc_error(rpc, EOF);
259                 return EOF;
260             }
261             byteq_advance_head(&rpc->input, retval);
262         } else {
263             size_t n, used;
264
265             if (!rpc->parser) {
266                 rpc->parser = json_parser_create(0);
267             }
268             n = byteq_tailroom(&rpc->input);
269             used = json_parser_feed(rpc->parser,
270                                     (char *) byteq_tail(&rpc->input), n);
271             byteq_advance_tail(&rpc->input, used);
272             if (json_parser_is_done(rpc->parser)) {
273                 jsonrpc_received(rpc);
274                 if (rpc->status) {
275                     const struct byteq *q = &rpc->input;
276                     if (q->head <= BYTEQ_SIZE) {
277                         stream_report_content(q->buffer, q->head,
278                                               STREAM_JSONRPC,
279                                               THIS_MODULE, rpc->name);
280                     }
281                     return rpc->status;
282                 }
283             }
284         }
285     }
286
287     *msgp = rpc->received;
288     rpc->received = NULL;
289     return 0;
290 }
291
292 void
293 jsonrpc_recv_wait(struct jsonrpc *rpc)
294 {
295     if (rpc->status || rpc->received || !byteq_is_empty(&rpc->input)) {
296         poll_immediate_wake();
297     } else {
298         stream_recv_wait(rpc->stream);
299     }
300 }
301
302 /* Always takes ownership of 'msg', regardless of success. */
303 int
304 jsonrpc_send_block(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
305 {
306     int error;
307
308     fatal_signal_run();
309
310     error = jsonrpc_send(rpc, msg);
311     if (error) {
312         return error;
313     }
314
315     for (;;) {
316         jsonrpc_run(rpc);
317         if (list_is_empty(&rpc->output) || rpc->status) {
318             return rpc->status;
319         }
320         jsonrpc_wait(rpc);
321         poll_block();
322     }
323 }
324
325 int
326 jsonrpc_recv_block(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
327 {
328     for (;;) {
329         int error = jsonrpc_recv(rpc, msgp);
330         if (error != EAGAIN) {
331             fatal_signal_run();
332             return error;
333         }
334
335         jsonrpc_run(rpc);
336         jsonrpc_wait(rpc);
337         jsonrpc_recv_wait(rpc);
338         poll_block();
339     }
340 }
341
342 /* Always takes ownership of 'request', regardless of success. */
343 int
344 jsonrpc_transact_block(struct jsonrpc *rpc, struct jsonrpc_msg *request,
345                        struct jsonrpc_msg **replyp)
346 {
347     struct jsonrpc_msg *reply = NULL;
348     struct json *id;
349     int error;
350
351     id = json_clone(request->id);
352     error = jsonrpc_send_block(rpc, request);
353     if (!error) {
354         for (;;) {
355             error = jsonrpc_recv_block(rpc, &reply);
356             if (error
357                 || (reply->type == JSONRPC_REPLY
358                     && json_equal(id, reply->id))) {
359                 break;
360             }
361             jsonrpc_msg_destroy(reply);
362         }
363     }
364     *replyp = error ? NULL : reply;
365     json_destroy(id);
366     return error;
367 }
368
369 static void
370 jsonrpc_received(struct jsonrpc *rpc)
371 {
372     struct jsonrpc_msg *msg;
373     struct json *json;
374     char *error;
375
376     json = json_parser_finish(rpc->parser);
377     rpc->parser = NULL;
378     if (json->type == JSON_STRING) {
379         VLOG_WARN_RL(&rl, "%s: error parsing stream: %s",
380                      rpc->name, json_string(json));
381         jsonrpc_error(rpc, EPROTO);
382         json_destroy(json);
383         return;
384     }
385
386     error = jsonrpc_msg_from_json(json, &msg);
387     if (error) {
388         VLOG_WARN_RL(&rl, "%s: received bad JSON-RPC message: %s",
389                      rpc->name, error);
390         free(error);
391         jsonrpc_error(rpc, EPROTO);
392         return;
393     }
394
395     jsonrpc_log_msg(rpc, "received", msg);
396     rpc->received = msg;
397 }
398
399 void
400 jsonrpc_error(struct jsonrpc *rpc, int error)
401 {
402     assert(error);
403     if (!rpc->status) {
404         rpc->status = error;
405         jsonrpc_cleanup(rpc);
406     }
407 }
408
409 static void
410 jsonrpc_cleanup(struct jsonrpc *rpc)
411 {
412     stream_close(rpc->stream);
413     rpc->stream = NULL;
414
415     json_parser_abort(rpc->parser);
416     rpc->parser = NULL;
417
418     jsonrpc_msg_destroy(rpc->received);
419     rpc->received = NULL;
420
421     ofpbuf_list_delete(&rpc->output);
422     rpc->backlog = 0;
423 }
424 \f
425 static struct jsonrpc_msg *
426 jsonrpc_create(enum jsonrpc_msg_type type, const char *method,
427                 struct json *params, struct json *result, struct json *error,
428                 struct json *id)
429 {
430     struct jsonrpc_msg *msg = xmalloc(sizeof *msg);
431     msg->type = type;
432     msg->method = method ? xstrdup(method) : NULL;
433     msg->params = params;
434     msg->result = result;
435     msg->error = error;
436     msg->id = id;
437     return msg;
438 }
439
440 static struct json *
441 jsonrpc_create_id(void)
442 {
443     static unsigned int id;
444     return json_integer_create(id++);
445 }
446
447 struct jsonrpc_msg *
448 jsonrpc_create_request(const char *method, struct json *params,
449                        struct json **idp)
450 {
451     struct json *id = jsonrpc_create_id();
452     if (idp) {
453         *idp = json_clone(id);
454     }
455     return jsonrpc_create(JSONRPC_REQUEST, method, params, NULL, NULL, id);
456 }
457
458 struct jsonrpc_msg *
459 jsonrpc_create_notify(const char *method, struct json *params)
460 {
461     return jsonrpc_create(JSONRPC_NOTIFY, method, params, NULL, NULL, NULL);
462 }
463
464 struct jsonrpc_msg *
465 jsonrpc_create_reply(struct json *result, const struct json *id)
466 {
467     return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, result, NULL,
468                            json_clone(id));
469 }
470
471 struct jsonrpc_msg *
472 jsonrpc_create_error(struct json *error, const struct json *id)
473 {
474     return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, NULL, error,
475                            json_clone(id));
476 }
477
478 const char *
479 jsonrpc_msg_type_to_string(enum jsonrpc_msg_type type)
480 {
481     switch (type) {
482     case JSONRPC_REQUEST:
483         return "request";
484
485     case JSONRPC_NOTIFY:
486         return "notification";
487
488     case JSONRPC_REPLY:
489         return "reply";
490
491     case JSONRPC_ERROR:
492         return "error";
493     }
494     return "(null)";
495 }
496
497 char *
498 jsonrpc_msg_is_valid(const struct jsonrpc_msg *m)
499 {
500     const char *type_name;
501     unsigned int pattern;
502
503     if (m->params && m->params->type != JSON_ARRAY) {
504         return xstrdup("\"params\" must be JSON array");
505     }
506
507     switch (m->type) {
508     case JSONRPC_REQUEST:
509         pattern = 0x11001;
510         break;
511
512     case JSONRPC_NOTIFY:
513         pattern = 0x11000;
514         break;
515
516     case JSONRPC_REPLY:
517         pattern = 0x00101;
518         break;
519
520     case JSONRPC_ERROR:
521         pattern = 0x00011;
522         break;
523
524     default:
525         return xasprintf("invalid JSON-RPC message type %d", m->type);
526     }
527
528     type_name = jsonrpc_msg_type_to_string(m->type);
529     if ((m->method != NULL) != ((pattern & 0x10000) != 0)) {
530         return xasprintf("%s must%s have \"method\"",
531                          type_name, (pattern & 0x10000) ? "" : " not");
532
533     }
534     if ((m->params != NULL) != ((pattern & 0x1000) != 0)) {
535         return xasprintf("%s must%s have \"params\"",
536                          type_name, (pattern & 0x1000) ? "" : " not");
537
538     }
539     if ((m->result != NULL) != ((pattern & 0x100) != 0)) {
540         return xasprintf("%s must%s have \"result\"",
541                          type_name, (pattern & 0x100) ? "" : " not");
542
543     }
544     if ((m->error != NULL) != ((pattern & 0x10) != 0)) {
545         return xasprintf("%s must%s have \"error\"",
546                          type_name, (pattern & 0x10) ? "" : " not");
547
548     }
549     if ((m->id != NULL) != ((pattern & 0x1) != 0)) {
550         return xasprintf("%s must%s have \"id\"",
551                          type_name, (pattern & 0x1) ? "" : " not");
552
553     }
554     return NULL;
555 }
556
557 void
558 jsonrpc_msg_destroy(struct jsonrpc_msg *m)
559 {
560     if (m) {
561         free(m->method);
562         json_destroy(m->params);
563         json_destroy(m->result);
564         json_destroy(m->error);
565         json_destroy(m->id);
566         free(m);
567     }
568 }
569
570 static struct json *
571 null_from_json_null(struct json *json)
572 {
573     if (json && json->type == JSON_NULL) {
574         json_destroy(json);
575         return NULL;
576     }
577     return json;
578 }
579
580 char *
581 jsonrpc_msg_from_json(struct json *json, struct jsonrpc_msg **msgp)
582 {
583     struct json *method = NULL;
584     struct jsonrpc_msg *msg = NULL;
585     struct shash *object;
586     char *error;
587
588     if (json->type != JSON_OBJECT) {
589         error = xstrdup("message is not a JSON object");
590         goto exit;
591     }
592     object = json_object(json);
593
594     method = shash_find_and_delete(object, "method");
595     if (method && method->type != JSON_STRING) {
596         error = xstrdup("method is not a JSON string");
597         goto exit;
598     }
599
600     msg = xzalloc(sizeof *msg);
601     msg->method = method ? xstrdup(method->u.string) : NULL;
602     msg->params = null_from_json_null(shash_find_and_delete(object, "params"));
603     msg->result = null_from_json_null(shash_find_and_delete(object, "result"));
604     msg->error = null_from_json_null(shash_find_and_delete(object, "error"));
605     msg->id = null_from_json_null(shash_find_and_delete(object, "id"));
606     msg->type = (msg->result ? JSONRPC_REPLY
607                  : msg->error ? JSONRPC_ERROR
608                  : msg->id ? JSONRPC_REQUEST
609                  : JSONRPC_NOTIFY);
610     if (!shash_is_empty(object)) {
611         error = xasprintf("message has unexpected member \"%s\"",
612                           shash_first(object)->name);
613         goto exit;
614     }
615     error = jsonrpc_msg_is_valid(msg);
616     if (error) {
617         goto exit;
618     }
619
620 exit:
621     json_destroy(method);
622     json_destroy(json);
623     if (error) {
624         jsonrpc_msg_destroy(msg);
625         msg = NULL;
626     }
627     *msgp = msg;
628     return error;
629 }
630
631 struct json *
632 jsonrpc_msg_to_json(struct jsonrpc_msg *m)
633 {
634     struct json *json = json_object_create();
635
636     if (m->method) {
637         json_object_put(json, "method", json_string_create_nocopy(m->method));
638     }
639
640     if (m->params) {
641         json_object_put(json, "params", m->params);
642     }
643
644     if (m->result) {
645         json_object_put(json, "result", m->result);
646     } else if (m->type == JSONRPC_ERROR) {
647         json_object_put(json, "result", json_null_create());
648     }
649
650     if (m->error) {
651         json_object_put(json, "error", m->error);
652     } else if (m->type == JSONRPC_REPLY) {
653         json_object_put(json, "error", json_null_create());
654     }
655
656     if (m->id) {
657         json_object_put(json, "id", m->id);
658     } else if (m->type == JSONRPC_NOTIFY) {
659         json_object_put(json, "id", json_null_create());
660     }
661
662     free(m);
663
664     return json;
665 }
666 \f
667 /* A JSON-RPC session with reconnection. */
668
669 struct jsonrpc_session {
670     struct reconnect *reconnect;
671     struct jsonrpc *rpc;
672     struct stream *stream;
673     struct pstream *pstream;
674     unsigned int seqno;
675 };
676
677 /* Creates and returns a jsonrpc_session to 'name', which should be a string
678  * acceptable to stream_open() or pstream_open().
679  *
680  * If 'name' is an active connection method, e.g. "tcp:127.1.2.3", the new
681  * jsonrpc_session connects and reconnects, with back-off, to 'name'.
682  *
683  * If 'name' is a passive connection method, e.g. "ptcp:", the new
684  * jsonrpc_session listens for connections to 'name'.  It maintains at most one
685  * connection at any given time.  Any new connection causes the previous one
686  * (if any) to be dropped. */
687 struct jsonrpc_session *
688 jsonrpc_session_open(const char *name)
689 {
690     struct jsonrpc_session *s;
691
692     s = xmalloc(sizeof *s);
693     s->reconnect = reconnect_create(time_msec());
694     reconnect_set_name(s->reconnect, name);
695     reconnect_enable(s->reconnect, time_msec());
696     s->rpc = NULL;
697     s->stream = NULL;
698     s->pstream = NULL;
699     s->seqno = 0;
700
701     if (!pstream_verify_name(name)) {
702         reconnect_set_passive(s->reconnect, true, time_msec());
703     }
704
705     return s;
706 }
707
708 /* Creates and returns a jsonrpc_session that is initially connected to
709  * 'jsonrpc'.  If the connection is dropped, it will not be reconnected.
710  *
711  * On the assumption that such connections are likely to be short-lived
712  * (e.g. from ovs-vsctl), informational logging for them is suppressed. */
713 struct jsonrpc_session *
714 jsonrpc_session_open_unreliably(struct jsonrpc *jsonrpc)
715 {
716     struct jsonrpc_session *s;
717
718     s = xmalloc(sizeof *s);
719     s->reconnect = reconnect_create(time_msec());
720     reconnect_set_quiet(s->reconnect, true);
721     reconnect_set_name(s->reconnect, jsonrpc_get_name(jsonrpc));
722     reconnect_set_max_tries(s->reconnect, 0);
723     reconnect_connected(s->reconnect, time_msec());
724     s->rpc = jsonrpc;
725     s->stream = NULL;
726     s->pstream = NULL;
727     s->seqno = 0;
728
729     return s;
730 }
731
732 void
733 jsonrpc_session_close(struct jsonrpc_session *s)
734 {
735     if (s) {
736         jsonrpc_close(s->rpc);
737         reconnect_destroy(s->reconnect);
738         stream_close(s->stream);
739         pstream_close(s->pstream);
740         free(s);
741     }
742 }
743
744 static void
745 jsonrpc_session_disconnect(struct jsonrpc_session *s)
746 {
747     if (s->rpc) {
748         jsonrpc_error(s->rpc, EOF);
749         jsonrpc_close(s->rpc);
750         s->rpc = NULL;
751         s->seqno++;
752     } else if (s->stream) {
753         stream_close(s->stream);
754         s->stream = NULL;
755         s->seqno++;
756     }
757 }
758
759 static void
760 jsonrpc_session_connect(struct jsonrpc_session *s)
761 {
762     const char *name = reconnect_get_name(s->reconnect);
763     int error;
764
765     jsonrpc_session_disconnect(s);
766     if (!reconnect_is_passive(s->reconnect)) {
767         error = jsonrpc_stream_open(name, &s->stream);
768         if (!error) {
769             reconnect_connecting(s->reconnect, time_msec());
770         }
771     } else {
772         error = s->pstream ? 0 : jsonrpc_pstream_open(name, &s->pstream);
773         if (!error) {
774             reconnect_listening(s->reconnect, time_msec());
775         }
776     }
777
778     if (error) {
779         reconnect_connect_failed(s->reconnect, time_msec(), error);
780     }
781     s->seqno++;
782 }
783
784 void
785 jsonrpc_session_run(struct jsonrpc_session *s)
786 {
787     if (s->pstream) {
788         struct stream *stream;
789         int error;
790
791         error = pstream_accept(s->pstream, &stream);
792         if (!error) {
793             if (s->rpc || s->stream) {
794                 VLOG_INFO_RL(&rl,
795                              "%s: new connection replacing active connection",
796                              reconnect_get_name(s->reconnect));
797                 jsonrpc_session_disconnect(s);
798             }
799             reconnect_connected(s->reconnect, time_msec());
800             s->rpc = jsonrpc_open(stream);
801         } else if (error != EAGAIN) {
802             reconnect_listen_error(s->reconnect, time_msec(), error);
803             pstream_close(s->pstream);
804             s->pstream = NULL;
805         }
806     }
807
808     if (s->rpc) {
809         int error;
810
811         jsonrpc_run(s->rpc);
812         error = jsonrpc_get_status(s->rpc);
813         if (error) {
814             reconnect_disconnected(s->reconnect, time_msec(), error);
815             jsonrpc_session_disconnect(s);
816         }
817     } else if (s->stream) {
818         int error;
819
820         stream_run(s->stream);
821         error = stream_connect(s->stream);
822         if (!error) {
823             reconnect_connected(s->reconnect, time_msec());
824             s->rpc = jsonrpc_open(s->stream);
825             s->stream = NULL;
826         } else if (error != EAGAIN) {
827             reconnect_connect_failed(s->reconnect, time_msec(), error);
828             stream_close(s->stream);
829             s->stream = NULL;
830         }
831     }
832
833     switch (reconnect_run(s->reconnect, time_msec())) {
834     case RECONNECT_CONNECT:
835         jsonrpc_session_connect(s);
836         break;
837
838     case RECONNECT_DISCONNECT:
839         reconnect_disconnected(s->reconnect, time_msec(), 0);
840         jsonrpc_session_disconnect(s);
841         break;
842
843     case RECONNECT_PROBE:
844         if (s->rpc) {
845             struct json *params;
846             struct jsonrpc_msg *request;
847
848             params = json_array_create_empty();
849             request = jsonrpc_create_request("echo", params, NULL);
850             json_destroy(request->id);
851             request->id = json_string_create("echo");
852             jsonrpc_send(s->rpc, request);
853         }
854         break;
855     }
856 }
857
858 void
859 jsonrpc_session_wait(struct jsonrpc_session *s)
860 {
861     if (s->rpc) {
862         jsonrpc_wait(s->rpc);
863     } else if (s->stream) {
864         stream_run_wait(s->stream);
865         stream_connect_wait(s->stream);
866     }
867     if (s->pstream) {
868         pstream_wait(s->pstream);
869     }
870     reconnect_wait(s->reconnect, time_msec());
871 }
872
873 size_t
874 jsonrpc_session_get_backlog(const struct jsonrpc_session *s)
875 {
876     return s->rpc ? jsonrpc_get_backlog(s->rpc) : 0;
877 }
878
879 /* Always returns a pointer to a valid C string, assuming 's' was initialized
880  * correctly. */
881 const char *
882 jsonrpc_session_get_name(const struct jsonrpc_session *s)
883 {
884     return reconnect_get_name(s->reconnect);
885 }
886
887 /* Always takes ownership of 'msg', regardless of success. */
888 int
889 jsonrpc_session_send(struct jsonrpc_session *s, struct jsonrpc_msg *msg)
890 {
891     if (s->rpc) {
892         return jsonrpc_send(s->rpc, msg);
893     } else {
894         jsonrpc_msg_destroy(msg);
895         return ENOTCONN;
896     }
897 }
898
899 struct jsonrpc_msg *
900 jsonrpc_session_recv(struct jsonrpc_session *s)
901 {
902     if (s->rpc) {
903         struct jsonrpc_msg *msg;
904         jsonrpc_recv(s->rpc, &msg);
905         if (msg) {
906             reconnect_received(s->reconnect, time_msec());
907             if (msg->type == JSONRPC_REQUEST && !strcmp(msg->method, "echo")) {
908                 /* Echo request.  Send reply. */
909                 struct jsonrpc_msg *reply;
910
911                 reply = jsonrpc_create_reply(json_clone(msg->params), msg->id);
912                 jsonrpc_session_send(s, reply);
913             } else if (msg->type == JSONRPC_REPLY
914                        && msg->id && msg->id->type == JSON_STRING
915                        && !strcmp(msg->id->u.string, "echo")) {
916                 /* It's a reply to our echo request.  Suppress it. */
917             } else {
918                 return msg;
919             }
920             jsonrpc_msg_destroy(msg);
921         }
922     }
923     return NULL;
924 }
925
926 void
927 jsonrpc_session_recv_wait(struct jsonrpc_session *s)
928 {
929     if (s->rpc) {
930         jsonrpc_recv_wait(s->rpc);
931     }
932 }
933
934 bool
935 jsonrpc_session_is_alive(const struct jsonrpc_session *s)
936 {
937     return s->rpc || s->stream || reconnect_get_max_tries(s->reconnect);
938 }
939
940 bool
941 jsonrpc_session_is_connected(const struct jsonrpc_session *s)
942 {
943     return s->rpc != NULL;
944 }
945
946 unsigned int
947 jsonrpc_session_get_seqno(const struct jsonrpc_session *s)
948 {
949     return s->seqno;
950 }
951
952 int
953 jsonrpc_session_get_status(const struct jsonrpc_session *s)
954 {
955     return s && s->rpc ? jsonrpc_get_status(s->rpc) : 0;
956 }
957
958 void
959 jsonrpc_session_get_reconnect_stats(const struct jsonrpc_session *s,
960                                     struct reconnect_stats *stats)
961 {
962     reconnect_get_stats(s->reconnect, time_msec(), stats);
963 }
964
965 void
966 jsonrpc_session_force_reconnect(struct jsonrpc_session *s)
967 {
968     reconnect_force_reconnect(s->reconnect, time_msec());
969 }
970
971 void
972 jsonrpc_session_set_max_backoff(struct jsonrpc_session *s, int max_backoff)
973 {
974     reconnect_set_backoff(s->reconnect, 0, max_backoff);
975 }
976
977 void
978 jsonrpc_session_set_probe_interval(struct jsonrpc_session *s,
979                                    int probe_interval)
980 {
981     reconnect_set_probe_interval(s->reconnect, probe_interval);
982 }