ovsdb/jsonrpc-server.c

   1 /* Copyright (c) 2009 Nicira Networks
   2  *
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include <config.h>
  17
  18 #include "jsonrpc-server.h"
  19
  20 #include <errno.h>
  21
  22 #include "column.h"
  23 #include "json.h"
  24 #include "jsonrpc.h"
  25 #include "ovsdb-error.h"
  26 #include "ovsdb-parser.h"
  27 #include "ovsdb.h"
  28 #include "reconnect.h"
  29 #include "row.h"
  30 #include "stream.h"
  31 #include "table.h"
  32 #include "timeval.h"
  33 #include "transaction.h"
  34 #include "trigger.h"
  35
  36 #define THIS_MODULE VLM_ovsdb_jsonrpc_server
  37 #include "vlog.h"
  38
  39 struct ovsdb_jsonrpc_session;
  40
  41 /* Sessions. */
  42 static void ovsdb_jsonrpc_session_create_active(struct ovsdb_jsonrpc_server *,
  43                                                 const char *name);
  44 static void ovsdb_jsonrpc_session_create_passive(struct ovsdb_jsonrpc_server *,
  45                                                  struct stream *);
  46 static void ovsdb_jsonrpc_session_run_all(struct ovsdb_jsonrpc_server *);
  47 static void ovsdb_jsonrpc_session_wait_all(struct ovsdb_jsonrpc_server *);
  48
  49 /* Triggers. */
  50 static void ovsdb_jsonrpc_trigger_create(struct ovsdb_jsonrpc_session *,
  51                                          struct json *id, struct json *params);
  52 static struct ovsdb_jsonrpc_trigger *ovsdb_jsonrpc_trigger_find(
  53     struct ovsdb_jsonrpc_session *, const struct json *id, size_t hash);
  54 static void ovsdb_jsonrpc_trigger_complete(struct ovsdb_jsonrpc_trigger *);
  55 static void ovsdb_jsonrpc_trigger_complete_all(struct ovsdb_jsonrpc_session *);
  56 static void ovsdb_jsonrpc_trigger_complete_done(
  57     struct ovsdb_jsonrpc_session *);
  58
  59 /* Monitors. */
  60 static struct json *ovsdb_jsonrpc_monitor_create(
  61     struct ovsdb_jsonrpc_session *, struct json *params);
  62 static struct jsonrpc_msg *ovsdb_jsonrpc_monitor_cancel(
  63     struct ovsdb_jsonrpc_session *,
  64     struct json_array *params,
  65     const struct json *request_id);
  66 static void ovsdb_jsonrpc_monitor_remove_all(struct ovsdb_jsonrpc_session *);
  67 \f
  68 /* JSON-RPC database server. */
  69
  70 struct ovsdb_jsonrpc_server {
  71     struct ovsdb *db;
  72
  73     struct list sessions;       /* List of "struct ovsdb_jsonrpc_session"s. */
  74     unsigned int n_sessions, max_sessions;
  75     unsigned int max_triggers;
  76
  77     struct pstream **listeners;
  78     size_t n_listeners, allocated_listeners;
  79 };
  80
  81 struct ovsdb_jsonrpc_server *
  82 ovsdb_jsonrpc_server_create(struct ovsdb *db)
  83 {
  84     struct ovsdb_jsonrpc_server *server = xzalloc(sizeof *server);
  85     server->db = db;
  86     server->max_sessions = 64;
  87     server->max_triggers = 64;
  88     list_init(&server->sessions);
  89     return server;
  90 }
  91
  92 int
  93 ovsdb_jsonrpc_server_listen(struct ovsdb_jsonrpc_server *svr, const char *name)
  94 {
  95     struct pstream *pstream;
  96     int error;
  97
  98     error = pstream_open(name, &pstream);
  99     if (error) {
 100         return error;
 101     }
 102
 103     if (svr->n_listeners >= svr->allocated_listeners) {
 104         svr->listeners = x2nrealloc(svr->listeners, &svr->allocated_listeners,
 105                                     sizeof *svr->listeners);
 106     }
 107     svr->listeners[svr->n_listeners++] = pstream;
 108     return 0;
 109 }
 110
 111 void
 112 ovsdb_jsonrpc_server_connect(struct ovsdb_jsonrpc_server *svr,
 113                              const char *name)
 114 {
 115     ovsdb_jsonrpc_session_create_active(svr, name);
 116 }
 117
 118 void
 119 ovsdb_jsonrpc_server_run(struct ovsdb_jsonrpc_server *svr)
 120 {
 121     size_t i;
 122
 123     /* Accept new connections. */
 124     for (i = 0; i < svr->n_listeners && svr->n_sessions < svr->max_sessions;) {
 125         struct pstream *listener = svr->listeners[i];
 126         struct stream *stream;
 127         int error;
 128
 129         error = pstream_accept(listener, &stream);
 130         if (!error) {
 131             ovsdb_jsonrpc_session_create_passive(svr, stream);
 132         } else if (error == EAGAIN) {
 133             i++;
 134         } else if (error) {
 135             VLOG_WARN("%s: accept failed: %s",
 136                       pstream_get_name(listener), strerror(error));
 137             pstream_close(listener);
 138             svr->listeners[i] = svr->listeners[--svr->n_listeners];
 139         }
 140     }
 141
 142     /* Handle each session. */
 143     ovsdb_jsonrpc_session_run_all(svr);
 144 }
 145
 146 void
 147 ovsdb_jsonrpc_server_wait(struct ovsdb_jsonrpc_server *svr)
 148 {
 149     if (svr->n_sessions < svr->max_sessions) {
 150         size_t i;
 151
 152         for (i = 0; i < svr->n_listeners; i++) {
 153             pstream_wait(svr->listeners[i]);
 154         }
 155     }
 156
 157     ovsdb_jsonrpc_session_wait_all(svr);
 158 }
 159 \f
 160 /* JSON-RPC database server session. */
 161
 162 struct ovsdb_jsonrpc_session {
 163     struct ovsdb_jsonrpc_server *server;
 164     struct list node;           /* Element in server's sessions list. */
 165
 166     /* Triggers. */
 167     struct hmap triggers;       /* Hmap of "struct ovsdb_jsonrpc_trigger"s. */
 168     struct list completions;    /* Completed triggers. */
 169
 170     /* Monitors. */
 171     struct hmap monitors;       /* Hmap of "struct ovsdb_jsonrpc_monitor"s. */
 172
 173     /* Connecting and reconnecting. */
 174     struct reconnect *reconnect; /* For back-off. */
 175     bool active;                /* Active or passive connection? */
 176     struct jsonrpc *rpc;
 177     struct stream *stream;      /* Only if active == false and rpc == NULL. */
 178 };
 179
 180 static void ovsdb_jsonrpc_session_close(struct ovsdb_jsonrpc_session *);
 181 static void ovsdb_jsonrpc_session_disconnect(struct ovsdb_jsonrpc_session *s);
 182 static int ovsdb_jsonrpc_session_run(struct ovsdb_jsonrpc_session *);
 183 static void ovsdb_jsonrpc_session_wait(struct ovsdb_jsonrpc_session *);
 184 static void ovsdb_jsonrpc_session_got_request(struct ovsdb_jsonrpc_session *,
 185                                              struct jsonrpc_msg *);
 186 static void ovsdb_jsonrpc_session_got_notify(struct ovsdb_jsonrpc_session *,
 187                                              struct jsonrpc_msg *);
 188
 189 static struct ovsdb_jsonrpc_session *
 190 ovsdb_jsonrpc_session_create(struct ovsdb_jsonrpc_server *svr,
 191                              const char *name, bool active)
 192 {
 193     struct ovsdb_jsonrpc_session *s;
 194
 195     s = xzalloc(sizeof *s);
 196     s->server = svr;
 197     list_push_back(&svr->sessions, &s->node);
 198     hmap_init(&s->triggers);
 199     hmap_init(&s->monitors);
 200     list_init(&s->completions);
 201     s->reconnect = reconnect_create(time_msec());
 202     reconnect_set_name(s->reconnect, name);
 203     reconnect_enable(s->reconnect, time_msec());
 204     s->active = active;
 205
 206     svr->n_sessions++;
 207
 208     return s;
 209 }
 210
 211 static void
 212 ovsdb_jsonrpc_session_create_active(struct ovsdb_jsonrpc_server *svr,
 213                                     const char *name)
 214 {
 215     ovsdb_jsonrpc_session_create(svr, name, true);
 216 }
 217
 218 static void
 219 ovsdb_jsonrpc_session_create_passive(struct ovsdb_jsonrpc_server *svr,
 220                                      struct stream *stream)
 221 {
 222     struct ovsdb_jsonrpc_session *s;
 223
 224     s = ovsdb_jsonrpc_session_create(svr, stream_get_name(stream), false);
 225     reconnect_connected(s->reconnect, time_msec());
 226     s->rpc = jsonrpc_open(stream);
 227 }
 228
 229 static void
 230 ovsdb_jsonrpc_session_close(struct ovsdb_jsonrpc_session *s)
 231 {
 232     ovsdb_jsonrpc_session_disconnect(s);
 233     list_remove(&s->node);
 234     s->server->n_sessions--;
 235 }
 236
 237 static void
 238 ovsdb_jsonrpc_session_disconnect(struct ovsdb_jsonrpc_session *s)
 239 {
 240     reconnect_disconnected(s->reconnect, time_msec(), 0);
 241     if (s->rpc) {
 242         jsonrpc_error(s->rpc, EOF);
 243         ovsdb_jsonrpc_trigger_complete_all(s);
 244         ovsdb_jsonrpc_monitor_remove_all(s);
 245         jsonrpc_close(s->rpc);
 246         s->rpc = NULL;
 247     } else if (s->stream) {
 248         stream_close(s->stream);
 249         s->stream = NULL;
 250     }
 251 }
 252
 253 static void
 254 ovsdb_jsonrpc_session_connect(struct ovsdb_jsonrpc_session *s)
 255 {
 256     ovsdb_jsonrpc_session_disconnect(s);
 257     if (s->active) {
 258         int error = stream_open(reconnect_get_name(s->reconnect), &s->stream);
 259         if (error) {
 260             reconnect_connect_failed(s->reconnect, time_msec(), error);
 261         } else {
 262             reconnect_connecting(s->reconnect, time_msec());
 263         }
 264     }
 265 }
 266
 267 static int
 268 ovsdb_jsonrpc_session_run(struct ovsdb_jsonrpc_session *s)
 269 {
 270     if (s->rpc) {
 271         struct jsonrpc_msg *msg;
 272         int error;
 273
 274         jsonrpc_run(s->rpc);
 275
 276         ovsdb_jsonrpc_trigger_complete_done(s);
 277
 278         if (!jsonrpc_get_backlog(s->rpc) && !jsonrpc_recv(s->rpc, &msg)) {
 279             reconnect_received(s->reconnect, time_msec());
 280             if (msg->type == JSONRPC_REQUEST) {
 281                 ovsdb_jsonrpc_session_got_request(s, msg);
 282             } else if (msg->type == JSONRPC_NOTIFY) {
 283                 ovsdb_jsonrpc_session_got_notify(s, msg);
 284             } else if (msg->type == JSONRPC_REPLY
 285                        && msg->id && msg->id->type == JSON_STRING
 286                        && !strcmp(msg->id->u.string, "echo")) {
 287                 /* It's a reply to our echo request.  Ignore it. */
 288             } else {
 289                 VLOG_WARN("%s: received unexpected %s message",
 290                           jsonrpc_get_name(s->rpc),
 291                           jsonrpc_msg_type_to_string(msg->type));
 292                 jsonrpc_error(s->rpc, EPROTO);
 293                 jsonrpc_msg_destroy(msg);
 294             }
 295         }
 296
 297         error = jsonrpc_get_status(s->rpc);
 298         if (error) {
 299             if (s->active) {
 300                 ovsdb_jsonrpc_session_disconnect(s);
 301             } else {
 302                 return error;
 303             }
 304         }
 305     } else if (s->stream) {
 306         int error = stream_connect(s->stream);
 307         if (!error) {
 308             reconnect_connected(s->reconnect, time_msec());
 309             s->rpc = jsonrpc_open(s->stream);
 310             s->stream = NULL;
 311         } else if (error != EAGAIN) {
 312             reconnect_connect_failed(s->reconnect, time_msec(), error);
 313             stream_close(s->stream);
 314             s->stream = NULL;
 315         }
 316     }
 317
 318     switch (reconnect_run(s->reconnect, time_msec())) {
 319     case RECONNECT_CONNECT:
 320         ovsdb_jsonrpc_session_connect(s);
 321         break;
 322
 323     case RECONNECT_DISCONNECT:
 324         ovsdb_jsonrpc_session_disconnect(s);
 325         break;
 326
 327     case RECONNECT_PROBE:
 328         if (s->rpc) {
 329             struct json *params;
 330             struct jsonrpc_msg *request;
 331
 332             params = json_array_create_empty();
 333             request = jsonrpc_create_request("echo", params);
 334             json_destroy(request->id);
 335             request->id = json_string_create("echo");
 336             jsonrpc_send(s->rpc, request);
 337         }
 338         break;
 339     }
 340     return s->active || s->rpc ? 0 : ETIMEDOUT;
 341 }
 342
 343 static void
 344 ovsdb_jsonrpc_session_run_all(struct ovsdb_jsonrpc_server *svr)
 345 {
 346     struct ovsdb_jsonrpc_session *s, *next;
 347
 348     LIST_FOR_EACH_SAFE (s, next, struct ovsdb_jsonrpc_session, node,
 349                         &svr->sessions) {
 350         int error = ovsdb_jsonrpc_session_run(s);
 351         if (error) {
 352             ovsdb_jsonrpc_session_close(s);
 353         }
 354     }
 355 }
 356
 357 static void
 358 ovsdb_jsonrpc_session_wait(struct ovsdb_jsonrpc_session *s)
 359 {
 360     if (s->rpc) {
 361         jsonrpc_wait(s->rpc);
 362         if (!jsonrpc_get_backlog(s->rpc)) {
 363             jsonrpc_recv_wait(s->rpc);
 364         }
 365     } else if (s->stream) {
 366         stream_connect_wait(s->stream);
 367     }
 368     reconnect_wait(s->reconnect, time_msec());
 369 }
 370
 371 static void
 372 ovsdb_jsonrpc_session_wait_all(struct ovsdb_jsonrpc_server *svr)
 373 {
 374     struct ovsdb_jsonrpc_session *s;
 375
 376     LIST_FOR_EACH (s, struct ovsdb_jsonrpc_session, node, &svr->sessions) {
 377         ovsdb_jsonrpc_session_wait(s);
 378     }
 379 }
 380
 381 static struct jsonrpc_msg *
 382 execute_transaction(struct ovsdb_jsonrpc_session *s,
 383                     struct jsonrpc_msg *request)
 384 {
 385     ovsdb_jsonrpc_trigger_create(s, request->id, request->params);
 386     request->id = NULL;
 387     request->params = NULL;
 388     return NULL;
 389 }
 390
 391 static void
 392 ovsdb_jsonrpc_session_got_request(struct ovsdb_jsonrpc_session *s,
 393                                   struct jsonrpc_msg *request)
 394 {
 395     struct jsonrpc_msg *reply;
 396
 397     if (!strcmp(request->method, "transact")) {
 398         reply = execute_transaction(s, request);
 399     } else if (!strcmp(request->method, "monitor")) {
 400         reply = jsonrpc_create_reply(
 401             ovsdb_jsonrpc_monitor_create(s, request->params), request->id);
 402     } else if (!strcmp(request->method, "monitor_cancel")) {
 403         reply = ovsdb_jsonrpc_monitor_cancel(s, json_array(request->params),
 404                                              request->id);
 405     } else if (!strcmp(request->method, "get_schema")) {
 406         reply = jsonrpc_create_reply(
 407             ovsdb_schema_to_json(s->server->db->schema), request->id);
 408     } else if (!strcmp(request->method, "echo")) {
 409         reply = jsonrpc_create_reply(json_clone(request->params), request->id);
 410     } else {
 411         reply = jsonrpc_create_error(json_string_create("unknown method"),
 412                                      request->id);
 413     }
 414
 415     if (reply) {
 416         jsonrpc_msg_destroy(request);
 417         jsonrpc_send(s->rpc, reply);
 418     }
 419 }
 420
 421 static void
 422 execute_cancel(struct ovsdb_jsonrpc_session *s, struct jsonrpc_msg *request)
 423 {
 424     if (json_array(request->params)->n == 1) {
 425         struct ovsdb_jsonrpc_trigger *t;
 426         struct json *id;
 427
 428         id = request->params->u.array.elems[0];
 429         t = ovsdb_jsonrpc_trigger_find(s, id, json_hash(id, 0));
 430         if (t) {
 431             ovsdb_jsonrpc_trigger_complete(t);
 432         }
 433     }
 434 }
 435
 436 static void
 437 ovsdb_jsonrpc_session_got_notify(struct ovsdb_jsonrpc_session *s,
 438                                  struct jsonrpc_msg *request)
 439 {
 440     if (!strcmp(request->method, "cancel")) {
 441         execute_cancel(s, request);
 442     }
 443     jsonrpc_msg_destroy(request);
 444 }
 445 \f
 446 /* JSON-RPC database server triggers.
 447  *
 448  * (Every transaction is treated as a trigger even if it doesn't actually have
 449  * any "wait" operations.) */
 450
 451 struct ovsdb_jsonrpc_trigger {
 452     struct ovsdb_trigger trigger;
 453     struct ovsdb_jsonrpc_session *session;
 454     struct hmap_node hmap_node; /* In session's "triggers" hmap. */
 455     struct json *id;
 456 };
 457
 458 static void
 459 ovsdb_jsonrpc_trigger_create(struct ovsdb_jsonrpc_session *s,
 460                              struct json *id, struct json *params)
 461 {
 462     struct ovsdb_jsonrpc_trigger *t;
 463     size_t hash;
 464
 465     /* Check for duplicate ID. */
 466     hash = json_hash(id, 0);
 467     t = ovsdb_jsonrpc_trigger_find(s, id, hash);
 468     if (t) {
 469         jsonrpc_send(s->rpc, jsonrpc_create_error(
 470                          json_string_create("duplicate request ID"), id));
 471         json_destroy(id);
 472         json_destroy(params);
 473         return;
 474     }
 475
 476     /* Insert into trigger table. */
 477     t = xmalloc(sizeof *t);
 478     ovsdb_trigger_init(s->server->db,
 479                        &t->trigger, params, &s->completions,
 480                        time_msec());
 481     t->session = s;
 482     t->id = id;
 483     hmap_insert(&s->triggers, &t->hmap_node, hash);
 484
 485     /* Complete early if possible. */
 486     if (ovsdb_trigger_is_complete(&t->trigger)) {
 487         ovsdb_jsonrpc_trigger_complete(t);
 488     }
 489 }
 490
 491 static struct ovsdb_jsonrpc_trigger *
 492 ovsdb_jsonrpc_trigger_find(struct ovsdb_jsonrpc_session *s,
 493                            const struct json *id, size_t hash)
 494 {
 495     struct ovsdb_jsonrpc_trigger *t;
 496
 497     HMAP_FOR_EACH_WITH_HASH (t, struct ovsdb_jsonrpc_trigger, hmap_node, hash,
 498                              &s->triggers) {
 499         if (json_equal(t->id, id)) {
 500             return t;
 501         }
 502     }
 503
 504     return NULL;
 505 }
 506
 507 static void
 508 ovsdb_jsonrpc_trigger_complete(struct ovsdb_jsonrpc_trigger *t)
 509 {
 510     struct ovsdb_jsonrpc_session *s = t->session;
 511
 512     if (s->rpc && !jsonrpc_get_status(s->rpc)) {
 513         struct jsonrpc_msg *reply;
 514         struct json *result;
 515
 516         result = ovsdb_trigger_steal_result(&t->trigger);
 517         if (result) {
 518             reply = jsonrpc_create_reply(result, t->id);
 519         } else {
 520             reply = jsonrpc_create_error(json_string_create("canceled"),
 521                                          t->id);
 522         }
 523         jsonrpc_send(s->rpc, reply);
 524     }
 525
 526     json_destroy(t->id);
 527     ovsdb_trigger_destroy(&t->trigger);
 528     hmap_remove(&s->triggers, &t->hmap_node);
 529     free(t);
 530 }
 531
 532 static void
 533 ovsdb_jsonrpc_trigger_complete_all(struct ovsdb_jsonrpc_session *s)
 534 {
 535     struct ovsdb_jsonrpc_trigger *t, *next;
 536     HMAP_FOR_EACH_SAFE (t, next, struct ovsdb_jsonrpc_trigger, hmap_node,
 537                         &s->triggers) {
 538         ovsdb_jsonrpc_trigger_complete(t);
 539     }
 540 }
 541
 542 static void
 543 ovsdb_jsonrpc_trigger_complete_done(struct ovsdb_jsonrpc_session *s)
 544 {
 545     while (!list_is_empty(&s->completions)) {
 546         struct ovsdb_jsonrpc_trigger *t
 547             = CONTAINER_OF(s->completions.next,
 548                            struct ovsdb_jsonrpc_trigger, trigger.node);
 549         ovsdb_jsonrpc_trigger_complete(t);
 550     }
 551 }
 552 \f
 553 /* JSON-RPC database table monitors. */
 554
 555 enum ovsdb_jsonrpc_monitor_selection {
 556     OJMS_INITIAL = 1 << 0,      /* All rows when monitor is created. */
 557     OJMS_INSERT = 1 << 1,       /* New rows. */
 558     OJMS_DELETE = 1 << 2,       /* Deleted rows. */
 559     OJMS_MODIFY = 1 << 3        /* Modified rows. */
 560 };
 561
 562 struct ovsdb_jsonrpc_monitor_table {
 563     const struct ovsdb_table *table;
 564     enum ovsdb_jsonrpc_monitor_selection select;
 565     struct ovsdb_column_set columns;
 566 };
 567
 568 struct ovsdb_jsonrpc_monitor {
 569     struct ovsdb_replica replica;
 570     struct ovsdb_jsonrpc_session *session;
 571     struct hmap_node node;      /* In ovsdb_jsonrpc_session's "monitors". */
 572
 573     struct json *monitor_id;
 574     struct shash tables;     /* Holds "struct ovsdb_jsonrpc_monitor_table"s. */
 575 };
 576
 577 static const struct ovsdb_replica_class ovsdb_jsonrpc_replica_class;
 578
 579 struct ovsdb_jsonrpc_monitor *ovsdb_jsonrpc_monitor_find(
 580     struct ovsdb_jsonrpc_session *, const struct json *monitor_id);
 581 static void ovsdb_jsonrpc_monitor_destroy(struct ovsdb_replica *);
 582 static struct json *ovsdb_jsonrpc_monitor_get_initial(
 583     const struct ovsdb_jsonrpc_monitor *);
 584
 585 static bool
 586 parse_bool(struct ovsdb_parser *parser, const char *name, bool default_value)
 587 {
 588     const struct json *json;
 589
 590     json = ovsdb_parser_member(parser, name, OP_BOOLEAN | OP_OPTIONAL);
 591     return json ? json_boolean(json) : default_value;
 592 }
 593
 594 struct ovsdb_jsonrpc_monitor *
 595 ovsdb_jsonrpc_monitor_find(struct ovsdb_jsonrpc_session *s,
 596                            const struct json *monitor_id)
 597 {
 598     struct ovsdb_jsonrpc_monitor *m;
 599
 600     HMAP_FOR_EACH_WITH_HASH (m, struct ovsdb_jsonrpc_monitor, node,
 601                              json_hash(monitor_id, 0), &s->monitors) {
 602         if (json_equal(m->monitor_id, monitor_id)) {
 603             return m;
 604         }
 605     }
 606
 607     return NULL;
 608 }
 609
 610 static struct json *
 611 ovsdb_jsonrpc_monitor_create(struct ovsdb_jsonrpc_session *s,
 612                              struct json *params)
 613 {
 614     struct ovsdb_jsonrpc_monitor *m = NULL;
 615     struct json *monitor_id, *monitor_requests;
 616     struct ovsdb_error *error = NULL;
 617     struct shash_node *node;
 618     struct json *json;
 619
 620     if (json_array(params)->n != 2) {
 621         error = ovsdb_syntax_error(params, NULL, "invalid parameters");
 622         goto error;
 623     }
 624     monitor_id = params->u.array.elems[0];
 625     monitor_requests = params->u.array.elems[1];
 626     if (monitor_requests->type != JSON_OBJECT) {
 627         error = ovsdb_syntax_error(monitor_requests, NULL,
 628                                    "monitor-requests must be object");
 629         goto error;
 630     }
 631
 632     if (ovsdb_jsonrpc_monitor_find(s, monitor_id)) {
 633         error = ovsdb_syntax_error(monitor_id, NULL, "duplicate monitor ID");
 634         goto error;
 635     }
 636
 637     m = xzalloc(sizeof *m);
 638     ovsdb_replica_init(&m->replica, &ovsdb_jsonrpc_replica_class);
 639     ovsdb_add_replica(s->server->db, &m->replica);
 640     m->session = s;
 641     hmap_insert(&s->monitors, &m->node, json_hash(monitor_id, 0));
 642     m->monitor_id = json_clone(monitor_id);
 643     shash_init(&m->tables);
 644
 645     SHASH_FOR_EACH (node, json_object(monitor_requests)) {
 646         const struct ovsdb_table *table;
 647         struct ovsdb_jsonrpc_monitor_table *mt;
 648         const struct json *columns_json, *select_json;
 649         struct ovsdb_parser parser;
 650
 651         table = ovsdb_get_table(s->server->db, node->name);
 652         if (!table) {
 653             error = ovsdb_syntax_error(NULL, NULL,
 654                                        "no table named %s", node->name);
 655             goto error;
 656         }
 657
 658         mt = xzalloc(sizeof *mt);
 659         mt->table = table;
 660         mt->select = OJMS_INITIAL | OJMS_INSERT | OJMS_DELETE | OJMS_MODIFY;
 661         ovsdb_column_set_init(&mt->columns);
 662         shash_add(&m->tables, table->schema->name, mt);
 663
 664         ovsdb_parser_init(&parser, node->data, "table %s", node->name);
 665         columns_json = ovsdb_parser_member(&parser, "columns",
 666                                            OP_ARRAY | OP_OPTIONAL);
 667         select_json = ovsdb_parser_member(&parser, "select",
 668                                           OP_OBJECT | OP_OPTIONAL);
 669         error = ovsdb_parser_finish(&parser);
 670         if (error) {
 671             goto error;
 672         }
 673
 674         if (columns_json) {
 675             error = ovsdb_column_set_from_json(columns_json, table,
 676                                                &mt->columns);
 677             if (error) {
 678                 goto error;
 679             }
 680         } else {
 681             struct shash_node *node;
 682
 683             SHASH_FOR_EACH (node, &table->schema->columns) {
 684                 const struct ovsdb_column *column = node->data;
 685                 if (column->index != OVSDB_COL_UUID) {
 686                     ovsdb_column_set_add(&mt->columns, column);
 687                 }
 688             }
 689         }
 690
 691         if (select_json) {
 692             mt->select = 0;
 693             ovsdb_parser_init(&parser, select_json, "table %s select",
 694                               table->schema->name);
 695             if (parse_bool(&parser, "initial", true)) {
 696                 mt->select |= OJMS_INITIAL;
 697             }
 698             if (parse_bool(&parser, "insert", true)) {
 699                 mt->select |= OJMS_INSERT;
 700             }
 701             if (parse_bool(&parser, "delete", true)) {
 702                 mt->select |= OJMS_DELETE;
 703             }
 704             if (parse_bool(&parser, "modify", true)) {
 705                 mt->select |= OJMS_MODIFY;
 706             }
 707             error = ovsdb_parser_finish(&parser);
 708             if (error) {
 709                 goto error;
 710             }
 711         }
 712     }
 713
 714     return ovsdb_jsonrpc_monitor_get_initial(m);
 715
 716 error:
 717     ovsdb_remove_replica(s->server->db, &m->replica);
 718
 719     json = ovsdb_error_to_json(error);
 720     ovsdb_error_destroy(error);
 721     return json;
 722 }
 723
 724 static struct jsonrpc_msg *
 725 ovsdb_jsonrpc_monitor_cancel(struct ovsdb_jsonrpc_session *s,
 726                              struct json_array *params,
 727                              const struct json *request_id)
 728 {
 729     if (params->n != 1) {
 730         return jsonrpc_create_error(json_string_create("invalid parameters"),
 731                                     request_id);
 732     } else {
 733         struct ovsdb_jsonrpc_monitor *m;
 734
 735         m = ovsdb_jsonrpc_monitor_find(s, params->elems[0]);
 736         if (!m) {
 737             return jsonrpc_create_error(json_string_create("unknown monitor"),
 738                                         request_id);
 739         } else {
 740             ovsdb_remove_replica(s->server->db, &m->replica);
 741             return jsonrpc_create_reply(json_object_create(), request_id);
 742         }
 743     }
 744 }
 745
 746 static void
 747 ovsdb_jsonrpc_monitor_remove_all(struct ovsdb_jsonrpc_session *s)
 748 {
 749     struct ovsdb_jsonrpc_monitor *m, *next;
 750
 751     HMAP_FOR_EACH_SAFE (m, next,
 752                         struct ovsdb_jsonrpc_monitor, node, &s->monitors) {
 753         ovsdb_remove_replica(s->server->db, &m->replica);
 754     }
 755 }
 756
 757 static struct ovsdb_jsonrpc_monitor *
 758 ovsdb_jsonrpc_monitor_cast(struct ovsdb_replica *replica)
 759 {
 760     assert(replica->class == &ovsdb_jsonrpc_replica_class);
 761     return CONTAINER_OF(replica, struct ovsdb_jsonrpc_monitor, replica);
 762 }
 763
 764 struct ovsdb_jsonrpc_monitor_aux {
 765     bool initial;               /* Sending initial contents of table? */
 766     const struct ovsdb_jsonrpc_monitor *monitor;
 767     struct json *json;          /* JSON for the whole transaction. */
 768
 769     /* Current table.  */
 770     struct ovsdb_jsonrpc_monitor_table *mt;
 771     struct json *table_json;    /* JSON for table's transaction. */
 772 };
 773
 774 static bool
 775 ovsdb_jsonrpc_monitor_change_cb(const struct ovsdb_row *old,
 776                                 const struct ovsdb_row *new,
 777                                 void *aux_)
 778 {
 779     struct ovsdb_jsonrpc_monitor_aux *aux = aux_;
 780     const struct ovsdb_jsonrpc_monitor *m = aux->monitor;
 781     struct ovsdb_table *table = new ? new->table : old->table;
 782     enum ovsdb_jsonrpc_monitor_selection type;
 783     struct json *old_json, *new_json;
 784     struct json *row_json;
 785     char uuid[UUID_LEN + 1];
 786     int n_changed;
 787     size_t i;
 788
 789     if (!aux->mt || table != aux->mt->table) {
 790         aux->mt = shash_find_data(&m->tables, table->schema->name);
 791         aux->table_json = NULL;
 792         if (!aux->mt) {
 793             /* We don't care about rows in this table at all.  Tell the caller
 794              * to skip it.  */
 795             return false;
 796         }
 797     }
 798
 799     type = (aux->initial ? OJMS_INITIAL
 800             : !old ? OJMS_INSERT
 801             : !new ? OJMS_DELETE
 802             : OJMS_MODIFY);
 803     if (!(aux->mt->select & type)) {
 804         /* We don't care about this type of change (but do want to be called
 805          * back for changes to other rows in the same table). */
 806         return true;
 807     }
 808
 809     old_json = new_json = NULL;
 810     n_changed = 0;
 811     for (i = 0; i < aux->mt->columns.n_columns; i++) {
 812         const struct ovsdb_column *column = aux->mt->columns.columns[i];
 813         unsigned int idx = column->index;
 814         bool changed = false;
 815
 816         if (type == OJMS_MODIFY) {
 817             changed = !ovsdb_datum_equals(&old->fields[idx],
 818                                           &new->fields[idx], &column->type);
 819             n_changed += changed;
 820         }
 821         if (changed || type == OJMS_DELETE) {
 822             if (!old_json) {
 823                 old_json = json_object_create();
 824             }
 825             json_object_put(old_json, column->name,
 826                             ovsdb_datum_to_json(&old->fields[idx],
 827                                                 &column->type));
 828         }
 829         if (type & (OJMS_INITIAL | OJMS_INSERT | OJMS_MODIFY)) {
 830             if (!new_json) {
 831                 new_json = json_object_create();
 832             }
 833             json_object_put(new_json, column->name,
 834                             ovsdb_datum_to_json(&new->fields[idx],
 835                                                 &column->type));
 836         }
 837     }
 838     if ((type == OJMS_MODIFY && !n_changed) || (!old_json && !new_json)) {
 839         /* No reportable changes. */
 840         json_destroy(old_json);
 841         json_destroy(new_json);
 842         return true;
 843     }
 844
 845     /* Create JSON object for transaction overall. */
 846     if (!aux->json) {
 847         aux->json = json_object_create();
 848     }
 849
 850     /* Create JSON object for transaction on this table. */
 851     if (!aux->table_json) {
 852         aux->table_json = json_object_create();
 853         json_object_put(aux->json, aux->mt->table->schema->name,
 854                         aux->table_json);
 855     }
 856
 857     /* Create JSON object for transaction on this row. */
 858     row_json = json_object_create();
 859     if (old_json) {
 860         json_object_put(row_json, "old", old_json);
 861     }
 862     if (new_json) {
 863         json_object_put(row_json, "new", new_json);
 864     }
 865
 866     /* Add JSON row to JSON table. */
 867     snprintf(uuid, sizeof uuid,
 868              UUID_FMT, UUID_ARGS(ovsdb_row_get_uuid(new ? new : old)));
 869     json_object_put(aux->table_json, uuid, row_json);
 870
 871     return true;
 872 }
 873
 874 static void
 875 ovsdb_jsonrpc_monitor_init_aux(struct ovsdb_jsonrpc_monitor_aux *aux,
 876                                const struct ovsdb_jsonrpc_monitor *m,
 877                                bool initial)
 878 {
 879     aux->initial = initial;
 880     aux->monitor = m;
 881     aux->json = NULL;
 882     aux->mt = NULL;
 883     aux->table_json = NULL;
 884 }
 885
 886 static struct ovsdb_error *
 887 ovsdb_jsonrpc_monitor_commit(struct ovsdb_replica *replica,
 888                              const struct ovsdb_txn *txn, bool durable UNUSED)
 889 {
 890     struct ovsdb_jsonrpc_monitor *m = ovsdb_jsonrpc_monitor_cast(replica);
 891     struct ovsdb_jsonrpc_monitor_aux aux;
 892
 893     ovsdb_jsonrpc_monitor_init_aux(&aux, m, false);
 894     ovsdb_txn_for_each_change(txn, ovsdb_jsonrpc_monitor_change_cb, &aux);
 895     if (aux.json) {
 896         struct jsonrpc_msg *msg;
 897         struct json *params;
 898
 899         params = json_array_create_2(json_clone(aux.monitor->monitor_id),
 900                                      aux.json);
 901         msg = jsonrpc_create_notify("update", params);
 902         jsonrpc_send(aux.monitor->session->rpc, msg);
 903     }
 904
 905     return NULL;
 906 }
 907
 908 static struct json *
 909 ovsdb_jsonrpc_monitor_get_initial(const struct ovsdb_jsonrpc_monitor *m)
 910 {
 911     struct ovsdb_jsonrpc_monitor_aux aux;
 912     struct shash_node *node;
 913
 914     ovsdb_jsonrpc_monitor_init_aux(&aux, m, true);
 915     SHASH_FOR_EACH (node, &m->tables) {
 916         struct ovsdb_jsonrpc_monitor_table *mt = node->data;
 917
 918         if (mt->select & OJMS_INITIAL) {
 919             struct ovsdb_row *row;
 920
 921             HMAP_FOR_EACH (row, struct ovsdb_row, hmap_node,
 922                            &mt->table->rows) {
 923                 ovsdb_jsonrpc_monitor_change_cb(NULL, row, &aux);
 924             }
 925         }
 926     }
 927     return aux.json ? aux.json : json_object_create();
 928 }
 929
 930 static void
 931 ovsdb_jsonrpc_monitor_destroy(struct ovsdb_replica *replica)
 932 {
 933     struct ovsdb_jsonrpc_monitor *m = ovsdb_jsonrpc_monitor_cast(replica);
 934     struct shash_node *node;
 935
 936     json_destroy(m->monitor_id);
 937     SHASH_FOR_EACH (node, &m->tables) {
 938         struct ovsdb_jsonrpc_monitor_table *mt = node->data;
 939         ovsdb_column_set_destroy(&mt->columns);
 940         free(mt);
 941     }
 942     shash_destroy(&m->tables);
 943     hmap_remove(&m->session->monitors, &m->node);
 944     free(m);
 945 }
 946
 947 static const struct ovsdb_replica_class ovsdb_jsonrpc_replica_class = {
 948     ovsdb_jsonrpc_monitor_commit,
 949     ovsdb_jsonrpc_monitor_destroy
 950 };