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