pintos-os.org Git - openvswitch/blob - 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             ds_put_and_free_cstr(&s, json_to_string(msg->params, 0));
 156         }
 157         if (msg->result) {
 158             ds_put_cstr(&s, ", result=");
 159             ds_put_and_free_cstr(&s, json_to_string(msg->result, 0));
 160         }
 161         if (msg->error) {
 162             ds_put_cstr(&s, ", error=");
 163             ds_put_and_free_cstr(&s, json_to_string(msg->error, 0));
 164         }
 165         if (msg->id) {
 166             ds_put_cstr(&s, ", id=");
 167             ds_put_and_free_cstr(&s, json_to_string(msg->id, 0));
 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 int
 176 jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
 177 {
 178     struct ofpbuf *buf;
 179     struct json *json;
 180     size_t length;
 181     char *s;
 182
 183     if (rpc->status) {
 184         jsonrpc_msg_destroy(msg);
 185         return rpc->status;
 186     }
 187
 188     jsonrpc_log_msg(rpc, "send", msg);
 189
 190     json = jsonrpc_msg_to_json(msg);
 191     s = json_to_string(json, 0);
 192     length = strlen(s);
 193     json_destroy(json);
 194
 195     buf = xmalloc(sizeof *buf);
 196     ofpbuf_use(buf, s, length);
 197     buf->size = length;
 198     queue_push_tail(&rpc->output, buf);
 199     rpc->backlog += length;
 200
 201     if (rpc->output.n == 1) {
 202         jsonrpc_run(rpc);
 203     }
 204     return rpc->status;
 205 }
 206
 207 int
 208 jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
 209 {
 210     *msgp = NULL;
 211     if (rpc->status) {
 212         return rpc->status;
 213     }
 214
 215     while (!rpc->received) {
 216         if (byteq_is_empty(&rpc->input)) {
 217             size_t chunk;
 218             int retval;
 219
 220             chunk = byteq_headroom(&rpc->input);
 221             retval = stream_recv(rpc->stream, byteq_head(&rpc->input), chunk);
 222             if (retval < 0) {
 223                 if (retval == -EAGAIN) {
 224                     return EAGAIN;
 225                 } else {
 226                     VLOG_WARN_RL(&rl, "%s: receive error: %s",
 227                                  rpc->name, strerror(-retval));
 228                     jsonrpc_error(rpc, -retval);
 229                     return rpc->status;
 230                 }
 231             } else if (retval == 0) {
 232                 VLOG_INFO_RL(&rl, "%s: connection closed", rpc->name);
 233                 jsonrpc_error(rpc, EOF);
 234                 return EOF;
 235             }
 236             byteq_advance_head(&rpc->input, retval);
 237         } else {
 238             size_t n, used;
 239
 240             if (!rpc->parser) {
 241                 rpc->parser = json_parser_create(0);
 242             }
 243             n = byteq_tailroom(&rpc->input);
 244             used = json_parser_feed(rpc->parser,
 245                                     (char *) byteq_tail(&rpc->input), n);
 246             byteq_advance_tail(&rpc->input, used);
 247             if (json_parser_is_done(rpc->parser)) {
 248                 jsonrpc_received(rpc);
 249                 if (rpc->status) {
 250                     return rpc->status;
 251                 }
 252             }
 253         }
 254     }
 255
 256     *msgp = rpc->received;
 257     rpc->received = NULL;
 258     return 0;
 259 }
 260
 261 void
 262 jsonrpc_recv_wait(struct jsonrpc *rpc)
 263 {
 264     if (rpc->status || rpc->received || !byteq_is_empty(&rpc->input)) {
 265         poll_immediate_wake();
 266     } else {
 267         stream_recv_wait(rpc->stream);
 268     }
 269 }
 270
 271 int
 272 jsonrpc_send_block(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
 273 {
 274     int error;
 275
 276     error = jsonrpc_send(rpc, msg);
 277     if (error) {
 278         return error;
 279     }
 280
 281     for (;;) {
 282         jsonrpc_run(rpc);
 283         if (queue_is_empty(&rpc->output) || rpc->status) {
 284             return rpc->status;
 285         }
 286         jsonrpc_wait(rpc);
 287         poll_block();
 288     }
 289 }
 290
 291 int
 292 jsonrpc_recv_block(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
 293 {
 294     for (;;) {
 295         int error = jsonrpc_recv(rpc, msgp);
 296         if (error != EAGAIN) {
 297             return error;
 298         }
 299
 300         jsonrpc_run(rpc);
 301         jsonrpc_wait(rpc);
 302         jsonrpc_recv_wait(rpc);
 303         poll_block();
 304     }
 305 }
 306
 307 int
 308 jsonrpc_transact_block(struct jsonrpc *rpc, struct jsonrpc_msg *request,
 309                        struct jsonrpc_msg **replyp)
 310 {
 311     struct jsonrpc_msg *reply = NULL;
 312     struct json *id;
 313     int error;
 314
 315     id = json_clone(request->id);
 316     error = jsonrpc_send_block(rpc, request);
 317     if (!error) {
 318         for (;;) {
 319             error = jsonrpc_recv_block(rpc, &reply);
 320             if (error
 321                 || (reply->type == JSONRPC_REPLY
 322                     && json_equal(id, reply->id))) {
 323                 break;
 324             }
 325             jsonrpc_msg_destroy(reply);
 326         }
 327     }
 328     *replyp = error ? NULL : reply;
 329     json_destroy(id);
 330     return error;
 331 }
 332
 333 static void
 334 jsonrpc_received(struct jsonrpc *rpc)
 335 {
 336     struct jsonrpc_msg *msg;
 337     struct json *json;
 338     char *error;
 339
 340     json = json_parser_finish(rpc->parser);
 341     rpc->parser = NULL;
 342     if (json->type == JSON_STRING) {
 343         VLOG_WARN_RL(&rl, "%s: error parsing stream: %s",
 344                      rpc->name, json_string(json));
 345         jsonrpc_error(rpc, EPROTO);
 346         json_destroy(json);
 347         return;
 348     }
 349
 350     error = jsonrpc_msg_from_json(json, &msg);
 351     if (error) {
 352         VLOG_WARN_RL(&rl, "%s: received bad JSON-RPC message: %s",
 353                      rpc->name, error);
 354         free(error);
 355         jsonrpc_error(rpc, EPROTO);
 356         return;
 357     }
 358
 359     jsonrpc_log_msg(rpc, "received", msg);
 360     rpc->received = msg;
 361 }
 362
 363 void
 364 jsonrpc_error(struct jsonrpc *rpc, int error)
 365 {
 366     assert(error);
 367     if (!rpc->status) {
 368         rpc->status = error;
 369         jsonrpc_cleanup(rpc);
 370     }
 371 }
 372
 373 static void
 374 jsonrpc_cleanup(struct jsonrpc *rpc)
 375 {
 376     stream_close(rpc->stream);
 377     rpc->stream = NULL;
 378
 379     json_parser_abort(rpc->parser);
 380     rpc->parser = NULL;
 381
 382     jsonrpc_msg_destroy(rpc->received);
 383     rpc->received = NULL;
 384
 385     queue_clear(&rpc->output);
 386     rpc->backlog = 0;
 387 }
 388 \f
 389 static struct jsonrpc_msg *
 390 jsonrpc_create(enum jsonrpc_msg_type type, const char *method,
 391                 struct json *params, struct json *result, struct json *error,
 392                 struct json *id)
 393 {
 394     struct jsonrpc_msg *msg = xmalloc(sizeof *msg);
 395     msg->type = type;
 396     msg->method = method ? xstrdup(method) : NULL;
 397     msg->params = params;
 398     msg->result = result;
 399     msg->error = error;
 400     msg->id = id;
 401     return msg;
 402 }
 403
 404 static struct json *
 405 jsonrpc_create_id(void)
 406 {
 407     static unsigned int id;
 408     return json_integer_create(id++);
 409 }
 410
 411 struct jsonrpc_msg *
 412 jsonrpc_create_request(const char *method, struct json *params,
 413                        struct json **idp)
 414 {
 415     struct json *id = jsonrpc_create_id();
 416     if (idp) {
 417         *idp = json_clone(id);
 418     }
 419     return jsonrpc_create(JSONRPC_REQUEST, method, params, NULL, NULL, id);
 420 }
 421
 422 struct jsonrpc_msg *
 423 jsonrpc_create_notify(const char *method, struct json *params)
 424 {
 425     return jsonrpc_create(JSONRPC_NOTIFY, method, params, NULL, NULL, NULL);
 426 }
 427
 428 struct jsonrpc_msg *
 429 jsonrpc_create_reply(struct json *result, const struct json *id)
 430 {
 431     return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, result, NULL,
 432                            json_clone(id));
 433 }
 434
 435 struct jsonrpc_msg *
 436 jsonrpc_create_error(struct json *error, const struct json *id)
 437 {
 438     return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, NULL, error,
 439                            json_clone(id));
 440 }
 441
 442 const char *
 443 jsonrpc_msg_type_to_string(enum jsonrpc_msg_type type)
 444 {
 445     switch (type) {
 446     case JSONRPC_REQUEST:
 447         return "request";
 448
 449     case JSONRPC_NOTIFY:
 450         return "notification";
 451
 452     case JSONRPC_REPLY:
 453         return "reply";
 454
 455     case JSONRPC_ERROR:
 456         return "error";
 457     }
 458     return "(null)";
 459 }
 460
 461 char *
 462 jsonrpc_msg_is_valid(const struct jsonrpc_msg *m)
 463 {
 464     const char *type_name;
 465     unsigned int pattern;
 466
 467     if (m->params && m->params->type != JSON_ARRAY) {
 468         return xstrdup("\"params\" must be JSON array");
 469     }
 470
 471     switch (m->type) {
 472     case JSONRPC_REQUEST:
 473         pattern = 0x11001;
 474         break;
 475
 476     case JSONRPC_NOTIFY:
 477         pattern = 0x11000;
 478         break;
 479
 480     case JSONRPC_REPLY:
 481         pattern = 0x00101;
 482         break;
 483
 484     case JSONRPC_ERROR:
 485         pattern = 0x00011;
 486         break;
 487
 488     default:
 489         return xasprintf("invalid JSON-RPC message type %d", m->type);
 490     }
 491
 492     type_name = jsonrpc_msg_type_to_string(m->type);
 493     if ((m->method != NULL) != ((pattern & 0x10000) != 0)) {
 494         return xasprintf("%s must%s have \"method\"",
 495                          type_name, (pattern & 0x10000) ? "" : " not");
 496
 497     }
 498     if ((m->params != NULL) != ((pattern & 0x1000) != 0)) {
 499         return xasprintf("%s must%s have \"params\"",
 500                          type_name, (pattern & 0x1000) ? "" : " not");
 501
 502     }
 503     if ((m->result != NULL) != ((pattern & 0x100) != 0)) {
 504         return xasprintf("%s must%s have \"result\"",
 505                          type_name, (pattern & 0x100) ? "" : " not");
 506
 507     }
 508     if ((m->error != NULL) != ((pattern & 0x10) != 0)) {
 509         return xasprintf("%s must%s have \"error\"",
 510                          type_name, (pattern & 0x10) ? "" : " not");
 511
 512     }
 513     if ((m->id != NULL) != ((pattern & 0x1) != 0)) {
 514         return xasprintf("%s must%s have \"id\"",
 515                          type_name, (pattern & 0x1) ? "" : " not");
 516
 517     }
 518     return NULL;
 519 }
 520
 521 void
 522 jsonrpc_msg_destroy(struct jsonrpc_msg *m)
 523 {
 524     if (m) {
 525         free(m->method);
 526         json_destroy(m->params);
 527         json_destroy(m->result);
 528         json_destroy(m->error);
 529         json_destroy(m->id);
 530         free(m);
 531     }
 532 }
 533
 534 static struct json *
 535 null_from_json_null(struct json *json)
 536 {
 537     if (json && json->type == JSON_NULL) {
 538         json_destroy(json);
 539         return NULL;
 540     }
 541     return json;
 542 }
 543
 544 char *
 545 jsonrpc_msg_from_json(struct json *json, struct jsonrpc_msg **msgp)
 546 {
 547     struct json *method = NULL;
 548     struct jsonrpc_msg *msg = NULL;
 549     struct shash *object;
 550     char *error;
 551
 552     if (json->type != JSON_OBJECT) {
 553         error = xstrdup("message is not a JSON object");
 554         goto exit;
 555     }
 556     object = json_object(json);
 557
 558     method = shash_find_and_delete(object, "method");
 559     if (method && method->type != JSON_STRING) {
 560         error = xstrdup("method is not a JSON string");
 561         goto exit;
 562     }
 563
 564     msg = xzalloc(sizeof *msg);
 565     msg->method = method ? xstrdup(method->u.string) : NULL;
 566     msg->params = null_from_json_null(shash_find_and_delete(object, "params"));
 567     msg->result = null_from_json_null(shash_find_and_delete(object, "result"));
 568     msg->error = null_from_json_null(shash_find_and_delete(object, "error"));
 569     msg->id = null_from_json_null(shash_find_and_delete(object, "id"));
 570     msg->type = (msg->result ? JSONRPC_REPLY
 571                  : msg->error ? JSONRPC_ERROR
 572                  : msg->id ? JSONRPC_REQUEST
 573                  : JSONRPC_NOTIFY);
 574     if (!shash_is_empty(object)) {
 575         error = xasprintf("message has unexpected member \"%s\"",
 576                           shash_first(object)->name);
 577         goto exit;
 578     }
 579     error = jsonrpc_msg_is_valid(msg);
 580     if (error) {
 581         goto exit;
 582     }
 583
 584 exit:
 585     json_destroy(method);
 586     json_destroy(json);
 587     if (error) {
 588         jsonrpc_msg_destroy(msg);
 589         msg = NULL;
 590     }
 591     *msgp = msg;
 592     return error;
 593 }
 594
 595 struct json *
 596 jsonrpc_msg_to_json(struct jsonrpc_msg *m)
 597 {
 598     struct json *json = json_object_create();
 599
 600     if (m->method) {
 601         json_object_put(json, "method", json_string_create_nocopy(m->method));
 602     }
 603
 604     if (m->params) {
 605         json_object_put(json, "params", m->params);
 606     }
 607
 608     if (m->result) {
 609         json_object_put(json, "result", m->result);
 610     } else if (m->type == JSONRPC_ERROR) {
 611         json_object_put(json, "result", json_null_create());
 612     }
 613
 614     if (m->error) {
 615         json_object_put(json, "error", m->error);
 616     } else if (m->type == JSONRPC_REPLY) {
 617         json_object_put(json, "error", json_null_create());
 618     }
 619
 620     if (m->id) {
 621         json_object_put(json, "id", m->id);
 622     } else if (m->type == JSONRPC_NOTIFY) {
 623         json_object_put(json, "id", json_null_create());
 624     }
 625
 626     free(m);
 627
 628     return json;
 629 }
 630 \f
 631 /* A JSON-RPC session with reconnection. */
 632
 633 struct jsonrpc_session {
 634     struct reconnect *reconnect;
 635     struct jsonrpc *rpc;
 636     struct stream *stream;
 637     unsigned int seqno;
 638 };
 639
 640 /* Creates and returns a jsonrpc_session that connects and reconnects, with
 641  * back-off, to 'name', which should be a string acceptable to
 642  * stream_open(). */
 643 struct jsonrpc_session *
 644 jsonrpc_session_open(const char *name)
 645 {
 646     struct jsonrpc_session *s;
 647
 648     s = xmalloc(sizeof *s);
 649     s->reconnect = reconnect_create(time_msec());
 650     reconnect_set_name(s->reconnect, name);
 651     reconnect_enable(s->reconnect, time_msec());
 652     s->rpc = NULL;
 653     s->stream = NULL;
 654     s->seqno = 0;
 655
 656     return s;
 657 }
 658
 659 /* Creates and returns a jsonrpc_session that is initially connected to
 660  * 'jsonrpc'.  If the connection is dropped, it will not be reconnected. */
 661 struct jsonrpc_session *
 662 jsonrpc_session_open_unreliably(struct jsonrpc *jsonrpc)
 663 {
 664     struct jsonrpc_session *s;
 665
 666     s = xmalloc(sizeof *s);
 667     s->reconnect = reconnect_create(time_msec());
 668     reconnect_set_name(s->reconnect, jsonrpc_get_name(jsonrpc));
 669     reconnect_set_max_tries(s->reconnect, 0);
 670     reconnect_connected(s->reconnect, time_msec());
 671     s->rpc = jsonrpc;
 672     s->stream = NULL;
 673     s->seqno = 0;
 674
 675     return s;
 676 }
 677
 678 void
 679 jsonrpc_session_close(struct jsonrpc_session *s)
 680 {
 681     if (s) {
 682         jsonrpc_close(s->rpc);
 683         reconnect_destroy(s->reconnect);
 684         free(s);
 685     }
 686 }
 687
 688 static void
 689 jsonrpc_session_disconnect(struct jsonrpc_session *s)
 690 {
 691     if (s->rpc) {
 692         jsonrpc_error(s->rpc, EOF);
 693         jsonrpc_close(s->rpc);
 694         s->rpc = NULL;
 695         s->seqno++;
 696     } else if (s->stream) {
 697         stream_close(s->stream);
 698         s->stream = NULL;
 699         s->seqno++;
 700     }
 701 }
 702
 703 static void
 704 jsonrpc_session_connect(struct jsonrpc_session *s)
 705 {
 706     int error;
 707
 708     jsonrpc_session_disconnect(s);
 709     error = stream_open(reconnect_get_name(s->reconnect), &s->stream);
 710     if (error) {
 711         reconnect_connect_failed(s->reconnect, time_msec(), error);
 712     } else {
 713         reconnect_connecting(s->reconnect, time_msec());
 714     }
 715     s->seqno++;
 716 }
 717
 718 void
 719 jsonrpc_session_run(struct jsonrpc_session *s)
 720 {
 721     if (s->rpc) {
 722         int error;
 723
 724         jsonrpc_run(s->rpc);
 725         error = jsonrpc_get_status(s->rpc);
 726         if (error) {
 727             reconnect_disconnected(s->reconnect, time_msec(), 0);
 728             jsonrpc_session_disconnect(s);
 729         }
 730     } else if (s->stream) {
 731         int error;
 732
 733         stream_run(s->stream);
 734         error = stream_connect(s->stream);
 735         if (!error) {
 736             reconnect_connected(s->reconnect, time_msec());
 737             s->rpc = jsonrpc_open(s->stream);
 738             s->stream = NULL;
 739         } else if (error != EAGAIN) {
 740             reconnect_connect_failed(s->reconnect, time_msec(), error);
 741             stream_close(s->stream);
 742             s->stream = NULL;
 743         }
 744     }
 745
 746     switch (reconnect_run(s->reconnect, time_msec())) {
 747     case RECONNECT_CONNECT:
 748         jsonrpc_session_connect(s);
 749         break;
 750
 751     case RECONNECT_DISCONNECT:
 752         reconnect_disconnected(s->reconnect, time_msec(), 0);
 753         jsonrpc_session_disconnect(s);
 754         break;
 755
 756     case RECONNECT_PROBE:
 757         if (s->rpc) {
 758             struct json *params;
 759             struct jsonrpc_msg *request;
 760
 761             params = json_array_create_empty();
 762             request = jsonrpc_create_request("echo", params, NULL);
 763             json_destroy(request->id);
 764             request->id = json_string_create("echo");
 765             jsonrpc_send(s->rpc, request);
 766         }
 767         break;
 768     }
 769 }
 770
 771 void
 772 jsonrpc_session_wait(struct jsonrpc_session *s)
 773 {
 774     if (s->rpc) {
 775         jsonrpc_wait(s->rpc);
 776     } else if (s->stream) {
 777         stream_run_wait(s->stream);
 778         stream_connect_wait(s->stream);
 779     }
 780     reconnect_wait(s->reconnect, time_msec());
 781 }
 782
 783 size_t
 784 jsonrpc_session_get_backlog(const struct jsonrpc_session *s)
 785 {
 786     return s->rpc ? jsonrpc_get_backlog(s->rpc) : 0;
 787 }
 788
 789 const char *
 790 jsonrpc_session_get_name(const struct jsonrpc_session *s)
 791 {
 792     return reconnect_get_name(s->reconnect);
 793 }
 794
 795 int
 796 jsonrpc_session_send(struct jsonrpc_session *s, struct jsonrpc_msg *msg)
 797 {
 798     return s->rpc ? jsonrpc_send(s->rpc, msg) : ENOTCONN;
 799 }
 800
 801 struct jsonrpc_msg *
 802 jsonrpc_session_recv(struct jsonrpc_session *s)
 803 {
 804     if (s->rpc) {
 805         struct jsonrpc_msg *msg;
 806         jsonrpc_recv(s->rpc, &msg);
 807         if (msg) {
 808             reconnect_received(s->reconnect, time_msec());
 809             if (msg->type == JSONRPC_REQUEST && !strcmp(msg->method, "echo")) {
 810                 /* Echo request.  Send reply. */
 811                 struct jsonrpc_msg *reply;
 812
 813                 reply = jsonrpc_create_reply(json_clone(msg->params), msg->id);
 814                 jsonrpc_session_send(s, reply);
 815             } else if (msg->type == JSONRPC_REPLY
 816                 && msg->id && msg->id->type == JSON_STRING
 817                 && !strcmp(msg->id->u.string, "echo")) {
 818                 /* It's a reply to our echo request.  Suppress it. */
 819             } else {
 820                 return msg;
 821             }
 822             jsonrpc_msg_destroy(msg);
 823         }
 824     }
 825     return NULL;
 826 }
 827
 828 void
 829 jsonrpc_session_recv_wait(struct jsonrpc_session *s)
 830 {
 831     if (s->rpc) {
 832         jsonrpc_recv_wait(s->rpc);
 833     }
 834 }
 835
 836 bool
 837 jsonrpc_session_is_alive(const struct jsonrpc_session *s)
 838 {
 839     return s->rpc || s->stream || reconnect_get_max_tries(s->reconnect);
 840 }
 841
 842 bool
 843 jsonrpc_session_is_connected(const struct jsonrpc_session *s)
 844 {
 845     return s->rpc != NULL;
 846 }
 847
 848 unsigned int
 849 jsonrpc_session_get_seqno(const struct jsonrpc_session *s)
 850 {
 851     return s->seqno;
 852 }
 853
 854 void
 855 jsonrpc_session_force_reconnect(struct jsonrpc_session *s)
 856 {
 857     reconnect_force_reconnect(s->reconnect, time_msec());
 858 }