lib/jsonrpc.c

   1 /*
   2  * Copyright (c) 2009, 2010, 2011 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)(rpc->name);
 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                 break;
 358             }
 359             if ((reply->type == JSONRPC_REPLY || reply->type == JSONRPC_ERROR)
 360                 && json_equal(id, reply->id)) {
 361                 break;
 362             }
 363             jsonrpc_msg_destroy(reply);
 364         }
 365     }
 366     *replyp = error ? NULL : reply;
 367     json_destroy(id);
 368     return error;
 369 }
 370
 371 static void
 372 jsonrpc_received(struct jsonrpc *rpc)
 373 {
 374     struct jsonrpc_msg *msg;
 375     struct json *json;
 376     char *error;
 377
 378     json = json_parser_finish(rpc->parser);
 379     rpc->parser = NULL;
 380     if (json->type == JSON_STRING) {
 381         VLOG_WARN_RL(&rl, "%s: error parsing stream: %s",
 382                      rpc->name, json_string(json));
 383         jsonrpc_error(rpc, EPROTO);
 384         json_destroy(json);
 385         return;
 386     }
 387
 388     error = jsonrpc_msg_from_json(json, &msg);
 389     if (error) {
 390         VLOG_WARN_RL(&rl, "%s: received bad JSON-RPC message: %s",
 391                      rpc->name, error);
 392         free(error);
 393         jsonrpc_error(rpc, EPROTO);
 394         return;
 395     }
 396
 397     jsonrpc_log_msg(rpc, "received", msg);
 398     rpc->received = msg;
 399 }
 400
 401 void
 402 jsonrpc_error(struct jsonrpc *rpc, int error)
 403 {
 404     assert(error);
 405     if (!rpc->status) {
 406         rpc->status = error;
 407         jsonrpc_cleanup(rpc);
 408     }
 409 }
 410
 411 static void
 412 jsonrpc_cleanup(struct jsonrpc *rpc)
 413 {
 414     stream_close(rpc->stream);
 415     rpc->stream = NULL;
 416
 417     json_parser_abort(rpc->parser);
 418     rpc->parser = NULL;
 419
 420     jsonrpc_msg_destroy(rpc->received);
 421     rpc->received = NULL;
 422
 423     ofpbuf_list_delete(&rpc->output);
 424     rpc->backlog = 0;
 425 }
 426 \f
 427 static struct jsonrpc_msg *
 428 jsonrpc_create(enum jsonrpc_msg_type type, const char *method,
 429                 struct json *params, struct json *result, struct json *error,
 430                 struct json *id)
 431 {
 432     struct jsonrpc_msg *msg = xmalloc(sizeof *msg);
 433     msg->type = type;
 434     msg->method = method ? xstrdup(method) : NULL;
 435     msg->params = params;
 436     msg->result = result;
 437     msg->error = error;
 438     msg->id = id;
 439     return msg;
 440 }
 441
 442 static struct json *
 443 jsonrpc_create_id(void)
 444 {
 445     static unsigned int id;
 446     return json_integer_create(id++);
 447 }
 448
 449 struct jsonrpc_msg *
 450 jsonrpc_create_request(const char *method, struct json *params,
 451                        struct json **idp)
 452 {
 453     struct json *id = jsonrpc_create_id();
 454     if (idp) {
 455         *idp = json_clone(id);
 456     }
 457     return jsonrpc_create(JSONRPC_REQUEST, method, params, NULL, NULL, id);
 458 }
 459
 460 struct jsonrpc_msg *
 461 jsonrpc_create_notify(const char *method, struct json *params)
 462 {
 463     return jsonrpc_create(JSONRPC_NOTIFY, method, params, NULL, NULL, NULL);
 464 }
 465
 466 struct jsonrpc_msg *
 467 jsonrpc_create_reply(struct json *result, const struct json *id)
 468 {
 469     return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, result, NULL,
 470                            json_clone(id));
 471 }
 472
 473 struct jsonrpc_msg *
 474 jsonrpc_create_error(struct json *error, const struct json *id)
 475 {
 476     return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, NULL, error,
 477                            json_clone(id));
 478 }
 479
 480 const char *
 481 jsonrpc_msg_type_to_string(enum jsonrpc_msg_type type)
 482 {
 483     switch (type) {
 484     case JSONRPC_REQUEST:
 485         return "request";
 486
 487     case JSONRPC_NOTIFY:
 488         return "notification";
 489
 490     case JSONRPC_REPLY:
 491         return "reply";
 492
 493     case JSONRPC_ERROR:
 494         return "error";
 495     }
 496     return "(null)";
 497 }
 498
 499 char *
 500 jsonrpc_msg_is_valid(const struct jsonrpc_msg *m)
 501 {
 502     const char *type_name;
 503     unsigned int pattern;
 504
 505     if (m->params && m->params->type != JSON_ARRAY) {
 506         return xstrdup("\"params\" must be JSON array");
 507     }
 508
 509     switch (m->type) {
 510     case JSONRPC_REQUEST:
 511         pattern = 0x11001;
 512         break;
 513
 514     case JSONRPC_NOTIFY:
 515         pattern = 0x11000;
 516         break;
 517
 518     case JSONRPC_REPLY:
 519         pattern = 0x00101;
 520         break;
 521
 522     case JSONRPC_ERROR:
 523         pattern = 0x00011;
 524         break;
 525
 526     default:
 527         return xasprintf("invalid JSON-RPC message type %d", m->type);
 528     }
 529
 530     type_name = jsonrpc_msg_type_to_string(m->type);
 531     if ((m->method != NULL) != ((pattern & 0x10000) != 0)) {
 532         return xasprintf("%s must%s have \"method\"",
 533                          type_name, (pattern & 0x10000) ? "" : " not");
 534
 535     }
 536     if ((m->params != NULL) != ((pattern & 0x1000) != 0)) {
 537         return xasprintf("%s must%s have \"params\"",
 538                          type_name, (pattern & 0x1000) ? "" : " not");
 539
 540     }
 541     if ((m->result != NULL) != ((pattern & 0x100) != 0)) {
 542         return xasprintf("%s must%s have \"result\"",
 543                          type_name, (pattern & 0x100) ? "" : " not");
 544
 545     }
 546     if ((m->error != NULL) != ((pattern & 0x10) != 0)) {
 547         return xasprintf("%s must%s have \"error\"",
 548                          type_name, (pattern & 0x10) ? "" : " not");
 549
 550     }
 551     if ((m->id != NULL) != ((pattern & 0x1) != 0)) {
 552         return xasprintf("%s must%s have \"id\"",
 553                          type_name, (pattern & 0x1) ? "" : " not");
 554
 555     }
 556     return NULL;
 557 }
 558
 559 void
 560 jsonrpc_msg_destroy(struct jsonrpc_msg *m)
 561 {
 562     if (m) {
 563         free(m->method);
 564         json_destroy(m->params);
 565         json_destroy(m->result);
 566         json_destroy(m->error);
 567         json_destroy(m->id);
 568         free(m);
 569     }
 570 }
 571
 572 static struct json *
 573 null_from_json_null(struct json *json)
 574 {
 575     if (json && json->type == JSON_NULL) {
 576         json_destroy(json);
 577         return NULL;
 578     }
 579     return json;
 580 }
 581
 582 char *
 583 jsonrpc_msg_from_json(struct json *json, struct jsonrpc_msg **msgp)
 584 {
 585     struct json *method = NULL;
 586     struct jsonrpc_msg *msg = NULL;
 587     struct shash *object;
 588     char *error;
 589
 590     if (json->type != JSON_OBJECT) {
 591         error = xstrdup("message is not a JSON object");
 592         goto exit;
 593     }
 594     object = json_object(json);
 595
 596     method = shash_find_and_delete(object, "method");
 597     if (method && method->type != JSON_STRING) {
 598         error = xstrdup("method is not a JSON string");
 599         goto exit;
 600     }
 601
 602     msg = xzalloc(sizeof *msg);
 603     msg->method = method ? xstrdup(method->u.string) : NULL;
 604     msg->params = null_from_json_null(shash_find_and_delete(object, "params"));
 605     msg->result = null_from_json_null(shash_find_and_delete(object, "result"));
 606     msg->error = null_from_json_null(shash_find_and_delete(object, "error"));
 607     msg->id = null_from_json_null(shash_find_and_delete(object, "id"));
 608     msg->type = (msg->result ? JSONRPC_REPLY
 609                  : msg->error ? JSONRPC_ERROR
 610                  : msg->id ? JSONRPC_REQUEST
 611                  : JSONRPC_NOTIFY);
 612     if (!shash_is_empty(object)) {
 613         error = xasprintf("message has unexpected member \"%s\"",
 614                           shash_first(object)->name);
 615         goto exit;
 616     }
 617     error = jsonrpc_msg_is_valid(msg);
 618     if (error) {
 619         goto exit;
 620     }
 621
 622 exit:
 623     json_destroy(method);
 624     json_destroy(json);
 625     if (error) {
 626         jsonrpc_msg_destroy(msg);
 627         msg = NULL;
 628     }
 629     *msgp = msg;
 630     return error;
 631 }
 632
 633 struct json *
 634 jsonrpc_msg_to_json(struct jsonrpc_msg *m)
 635 {
 636     struct json *json = json_object_create();
 637
 638     if (m->method) {
 639         json_object_put(json, "method", json_string_create_nocopy(m->method));
 640     }
 641
 642     if (m->params) {
 643         json_object_put(json, "params", m->params);
 644     }
 645
 646     if (m->result) {
 647         json_object_put(json, "result", m->result);
 648     } else if (m->type == JSONRPC_ERROR) {
 649         json_object_put(json, "result", json_null_create());
 650     }
 651
 652     if (m->error) {
 653         json_object_put(json, "error", m->error);
 654     } else if (m->type == JSONRPC_REPLY) {
 655         json_object_put(json, "error", json_null_create());
 656     }
 657
 658     if (m->id) {
 659         json_object_put(json, "id", m->id);
 660     } else if (m->type == JSONRPC_NOTIFY) {
 661         json_object_put(json, "id", json_null_create());
 662     }
 663
 664     free(m);
 665
 666     return json;
 667 }
 668 \f
 669 /* A JSON-RPC session with reconnection. */
 670
 671 struct jsonrpc_session {
 672     struct reconnect *reconnect;
 673     struct jsonrpc *rpc;
 674     struct stream *stream;
 675     struct pstream *pstream;
 676     unsigned int seqno;
 677 };
 678
 679 /* Creates and returns a jsonrpc_session to 'name', which should be a string
 680  * acceptable to stream_open() or pstream_open().
 681  *
 682  * If 'name' is an active connection method, e.g. "tcp:127.1.2.3", the new
 683  * jsonrpc_session connects and reconnects, with back-off, to 'name'.
 684  *
 685  * If 'name' is a passive connection method, e.g. "ptcp:", the new
 686  * jsonrpc_session listens for connections to 'name'.  It maintains at most one
 687  * connection at any given time.  Any new connection causes the previous one
 688  * (if any) to be dropped. */
 689 struct jsonrpc_session *
 690 jsonrpc_session_open(const char *name)
 691 {
 692     struct jsonrpc_session *s;
 693
 694     s = xmalloc(sizeof *s);
 695     s->reconnect = reconnect_create(time_msec());
 696     reconnect_set_name(s->reconnect, name);
 697     reconnect_enable(s->reconnect, time_msec());
 698     s->rpc = NULL;
 699     s->stream = NULL;
 700     s->pstream = NULL;
 701     s->seqno = 0;
 702
 703     if (!pstream_verify_name(name)) {
 704         reconnect_set_passive(s->reconnect, true, time_msec());
 705     }
 706
 707     return s;
 708 }
 709
 710 /* Creates and returns a jsonrpc_session that is initially connected to
 711  * 'jsonrpc'.  If the connection is dropped, it will not be reconnected.
 712  *
 713  * On the assumption that such connections are likely to be short-lived
 714  * (e.g. from ovs-vsctl), informational logging for them is suppressed. */
 715 struct jsonrpc_session *
 716 jsonrpc_session_open_unreliably(struct jsonrpc *jsonrpc)
 717 {
 718     struct jsonrpc_session *s;
 719
 720     s = xmalloc(sizeof *s);
 721     s->reconnect = reconnect_create(time_msec());
 722     reconnect_set_quiet(s->reconnect, true);
 723     reconnect_set_name(s->reconnect, jsonrpc_get_name(jsonrpc));
 724     reconnect_set_max_tries(s->reconnect, 0);
 725     reconnect_connected(s->reconnect, time_msec());
 726     s->rpc = jsonrpc;
 727     s->stream = NULL;
 728     s->pstream = NULL;
 729     s->seqno = 0;
 730
 731     return s;
 732 }
 733
 734 void
 735 jsonrpc_session_close(struct jsonrpc_session *s)
 736 {
 737     if (s) {
 738         jsonrpc_close(s->rpc);
 739         reconnect_destroy(s->reconnect);
 740         stream_close(s->stream);
 741         pstream_close(s->pstream);
 742         free(s);
 743     }
 744 }
 745
 746 static void
 747 jsonrpc_session_disconnect(struct jsonrpc_session *s)
 748 {
 749     if (s->rpc) {
 750         jsonrpc_error(s->rpc, EOF);
 751         jsonrpc_close(s->rpc);
 752         s->rpc = NULL;
 753         s->seqno++;
 754     } else if (s->stream) {
 755         stream_close(s->stream);
 756         s->stream = NULL;
 757         s->seqno++;
 758     }
 759 }
 760
 761 static void
 762 jsonrpc_session_connect(struct jsonrpc_session *s)
 763 {
 764     const char *name = reconnect_get_name(s->reconnect);
 765     int error;
 766
 767     jsonrpc_session_disconnect(s);
 768     if (!reconnect_is_passive(s->reconnect)) {
 769         error = jsonrpc_stream_open(name, &s->stream);
 770         if (!error) {
 771             reconnect_connecting(s->reconnect, time_msec());
 772         }
 773     } else {
 774         error = s->pstream ? 0 : jsonrpc_pstream_open(name, &s->pstream);
 775         if (!error) {
 776             reconnect_listening(s->reconnect, time_msec());
 777         }
 778     }
 779
 780     if (error) {
 781         reconnect_connect_failed(s->reconnect, time_msec(), error);
 782     }
 783     s->seqno++;
 784 }
 785
 786 void
 787 jsonrpc_session_run(struct jsonrpc_session *s)
 788 {
 789     if (s->pstream) {
 790         struct stream *stream;
 791         int error;
 792
 793         error = pstream_accept(s->pstream, &stream);
 794         if (!error) {
 795             if (s->rpc || s->stream) {
 796                 VLOG_INFO_RL(&rl,
 797                              "%s: new connection replacing active connection",
 798                              reconnect_get_name(s->reconnect));
 799                 jsonrpc_session_disconnect(s);
 800             }
 801             reconnect_connected(s->reconnect, time_msec());
 802             s->rpc = jsonrpc_open(stream);
 803         } else if (error != EAGAIN) {
 804             reconnect_listen_error(s->reconnect, time_msec(), error);
 805             pstream_close(s->pstream);
 806             s->pstream = NULL;
 807         }
 808     }
 809
 810     if (s->rpc) {
 811         int error;
 812
 813         jsonrpc_run(s->rpc);
 814         error = jsonrpc_get_status(s->rpc);
 815         if (error) {
 816             reconnect_disconnected(s->reconnect, time_msec(), error);
 817             jsonrpc_session_disconnect(s);
 818         }
 819     } else if (s->stream) {
 820         int error;
 821
 822         stream_run(s->stream);
 823         error = stream_connect(s->stream);
 824         if (!error) {
 825             reconnect_connected(s->reconnect, time_msec());
 826             s->rpc = jsonrpc_open(s->stream);
 827             s->stream = NULL;
 828         } else if (error != EAGAIN) {
 829             reconnect_connect_failed(s->reconnect, time_msec(), error);
 830             stream_close(s->stream);
 831             s->stream = NULL;
 832         }
 833     }
 834
 835     switch (reconnect_run(s->reconnect, time_msec())) {
 836     case RECONNECT_CONNECT:
 837         jsonrpc_session_connect(s);
 838         break;
 839
 840     case RECONNECT_DISCONNECT:
 841         reconnect_disconnected(s->reconnect, time_msec(), 0);
 842         jsonrpc_session_disconnect(s);
 843         break;
 844
 845     case RECONNECT_PROBE:
 846         if (s->rpc) {
 847             struct json *params;
 848             struct jsonrpc_msg *request;
 849
 850             params = json_array_create_empty();
 851             request = jsonrpc_create_request("echo", params, NULL);
 852             json_destroy(request->id);
 853             request->id = json_string_create("echo");
 854             jsonrpc_send(s->rpc, request);
 855         }
 856         break;
 857     }
 858 }
 859
 860 void
 861 jsonrpc_session_wait(struct jsonrpc_session *s)
 862 {
 863     if (s->rpc) {
 864         jsonrpc_wait(s->rpc);
 865     } else if (s->stream) {
 866         stream_run_wait(s->stream);
 867         stream_connect_wait(s->stream);
 868     }
 869     if (s->pstream) {
 870         pstream_wait(s->pstream);
 871     }
 872     reconnect_wait(s->reconnect, time_msec());
 873 }
 874
 875 size_t
 876 jsonrpc_session_get_backlog(const struct jsonrpc_session *s)
 877 {
 878     return s->rpc ? jsonrpc_get_backlog(s->rpc) : 0;
 879 }
 880
 881 /* Always returns a pointer to a valid C string, assuming 's' was initialized
 882  * correctly. */
 883 const char *
 884 jsonrpc_session_get_name(const struct jsonrpc_session *s)
 885 {
 886     return reconnect_get_name(s->reconnect);
 887 }
 888
 889 /* Always takes ownership of 'msg', regardless of success. */
 890 int
 891 jsonrpc_session_send(struct jsonrpc_session *s, struct jsonrpc_msg *msg)
 892 {
 893     if (s->rpc) {
 894         return jsonrpc_send(s->rpc, msg);
 895     } else {
 896         jsonrpc_msg_destroy(msg);
 897         return ENOTCONN;
 898     }
 899 }
 900
 901 struct jsonrpc_msg *
 902 jsonrpc_session_recv(struct jsonrpc_session *s)
 903 {
 904     if (s->rpc) {
 905         struct jsonrpc_msg *msg;
 906         jsonrpc_recv(s->rpc, &msg);
 907         if (msg) {
 908             reconnect_received(s->reconnect, time_msec());
 909             if (msg->type == JSONRPC_REQUEST && !strcmp(msg->method, "echo")) {
 910                 /* Echo request.  Send reply. */
 911                 struct jsonrpc_msg *reply;
 912
 913                 reply = jsonrpc_create_reply(json_clone(msg->params), msg->id);
 914                 jsonrpc_session_send(s, reply);
 915             } else if (msg->type == JSONRPC_REPLY
 916                        && msg->id && msg->id->type == JSON_STRING
 917                        && !strcmp(msg->id->u.string, "echo")) {
 918                 /* It's a reply to our echo request.  Suppress it. */
 919             } else {
 920                 return msg;
 921             }
 922             jsonrpc_msg_destroy(msg);
 923         }
 924     }
 925     return NULL;
 926 }
 927
 928 void
 929 jsonrpc_session_recv_wait(struct jsonrpc_session *s)
 930 {
 931     if (s->rpc) {
 932         jsonrpc_recv_wait(s->rpc);
 933     }
 934 }
 935
 936 bool
 937 jsonrpc_session_is_alive(const struct jsonrpc_session *s)
 938 {
 939     return s->rpc || s->stream || reconnect_get_max_tries(s->reconnect);
 940 }
 941
 942 bool
 943 jsonrpc_session_is_connected(const struct jsonrpc_session *s)
 944 {
 945     return s->rpc != NULL;
 946 }
 947
 948 unsigned int
 949 jsonrpc_session_get_seqno(const struct jsonrpc_session *s)
 950 {
 951     return s->seqno;
 952 }
 953
 954 int
 955 jsonrpc_session_get_status(const struct jsonrpc_session *s)
 956 {
 957     return s && s->rpc ? jsonrpc_get_status(s->rpc) : 0;
 958 }
 959
 960 void
 961 jsonrpc_session_get_reconnect_stats(const struct jsonrpc_session *s,
 962                                     struct reconnect_stats *stats)
 963 {
 964     reconnect_get_stats(s->reconnect, time_msec(), stats);
 965 }
 966
 967 void
 968 jsonrpc_session_force_reconnect(struct jsonrpc_session *s)
 969 {
 970     reconnect_force_reconnect(s->reconnect, time_msec());
 971 }
 972
 973 void
 974 jsonrpc_session_set_max_backoff(struct jsonrpc_session *s, int max_backoff)
 975 {
 976     reconnect_set_backoff(s->reconnect, 0, max_backoff);
 977 }
 978
 979 void
 980 jsonrpc_session_set_probe_interval(struct jsonrpc_session *s,
 981                                    int probe_interval)
 982 {
 983     reconnect_set_probe_interval(s->reconnect, probe_interval);
 984 }