X-Git-Url: https://pintos-os.org/cgi-bin/gitweb.cgi?a=blobdiff_plain;f=lib%2Fjsonrpc.c;h=f42035e80f5884e7cf25aeca0a730dea391f4c27;hb=fe55ad159d8fd396a9e4914a03eea93d096d03b1;hp=bd019f7381694d3b0806315375ee7a2717ae9105;hpb=58fda1dab104041fc693032475ec4662c1a52849;p=openvswitch diff --git a/lib/jsonrpc.c b/lib/jsonrpc.c index bd019f73..f42035e8 100644 --- a/lib/jsonrpc.c +++ b/lib/jsonrpc.c @@ -1,5 +1,5 @@ /* - * Copyright (c) 2009 Nicira Networks. + * Copyright (c) 2009, 2010 Nicira Networks. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,6 +18,7 @@ #include "jsonrpc.h" +#include #include #include "byteq.h" @@ -88,6 +89,7 @@ jsonrpc_run(struct jsonrpc *rpc) return; } + stream_run(rpc->stream); while (!queue_is_empty(&rpc->output)) { struct ofpbuf *buf = rpc->output.head; int retval; @@ -113,8 +115,11 @@ jsonrpc_run(struct jsonrpc *rpc) void jsonrpc_wait(struct jsonrpc *rpc) { - if (!rpc->status && !queue_is_empty(&rpc->output)) { - stream_send_wait(rpc->stream); + if (!rpc->status) { + stream_run_wait(rpc->stream); + if (!queue_is_empty(&rpc->output)) { + stream_send_wait(rpc->stream); + } } } @@ -630,6 +635,9 @@ struct jsonrpc_session { unsigned int seqno; }; +/* Creates and returns a jsonrpc_session that connects and reconnects, with + * back-off, to 'name', which should be a string acceptable to + * stream_open(). */ struct jsonrpc_session * jsonrpc_session_open(const char *name) { @@ -646,6 +654,25 @@ jsonrpc_session_open(const char *name) return s; } +/* Creates and returns a jsonrpc_session that is initially connected to + * 'jsonrpc'. If the connection is dropped, it will not be reconnected. */ +struct jsonrpc_session * +jsonrpc_session_open_unreliably(struct jsonrpc *jsonrpc) +{ + struct jsonrpc_session *s; + + s = xmalloc(sizeof *s); + s->reconnect = reconnect_create(time_msec()); + reconnect_set_name(s->reconnect, jsonrpc_get_name(jsonrpc)); + reconnect_set_max_tries(s->reconnect, 0); + reconnect_connected(s->reconnect, time_msec()); + s->rpc = jsonrpc; + s->stream = NULL; + s->seqno = 0; + + return s; +} + void jsonrpc_session_close(struct jsonrpc_session *s) { @@ -699,7 +726,10 @@ jsonrpc_session_run(struct jsonrpc_session *s) jsonrpc_session_disconnect(s); } } else if (s->stream) { - int error = stream_connect(s->stream); + int error; + + stream_run(s->stream); + error = stream_connect(s->stream); if (!error) { reconnect_connected(s->reconnect, time_msec()); s->rpc = jsonrpc_open(s->stream); @@ -741,6 +771,7 @@ jsonrpc_session_wait(struct jsonrpc_session *s) if (s->rpc) { jsonrpc_wait(s->rpc); } else if (s->stream) { + stream_run_wait(s->stream); stream_connect_wait(s->stream); } reconnect_wait(s->reconnect, time_msec()); @@ -767,14 +798,28 @@ jsonrpc_session_send(struct jsonrpc_session *s, struct jsonrpc_msg *msg) struct jsonrpc_msg * jsonrpc_session_recv(struct jsonrpc_session *s) { - struct jsonrpc_msg *msg = NULL; if (s->rpc) { + struct jsonrpc_msg *msg; jsonrpc_recv(s->rpc, &msg); if (msg) { reconnect_received(s->reconnect, time_msec()); + if (msg->type == JSONRPC_REQUEST && !strcmp(msg->method, "echo")) { + /* Echo request. Send reply. */ + struct jsonrpc_msg *reply; + + reply = jsonrpc_create_reply(json_clone(msg->params), msg->id); + jsonrpc_session_send(s, reply); + } else if (msg->type == JSONRPC_REPLY + && msg->id && msg->id->type == JSON_STRING + && !strcmp(msg->id->u.string, "echo")) { + /* It's a reply to our echo request. Suppress it. */ + } else { + return msg; + } + jsonrpc_msg_destroy(msg); } } - return msg; + return NULL; } void @@ -785,6 +830,12 @@ jsonrpc_session_recv_wait(struct jsonrpc_session *s) } } +bool +jsonrpc_session_is_alive(const struct jsonrpc_session *s) +{ + return s->rpc || s->stream || reconnect_get_max_tries(s->reconnect); +} + bool jsonrpc_session_is_connected(const struct jsonrpc_session *s) {