TAILQ_INIT(&base->registered_rpcs);
TAILQ_INIT(&base->input_hooks);
TAILQ_INIT(&base->output_hooks);
+
+ TAILQ_INIT(&base->paused_requests);
+
base->http_server = http_server;
return (base);
void *
evrpc_add_hook(void *vbase,
enum EVRPC_HOOK_TYPE hook_type,
- int (*cb)(struct evhttp_request *, struct evbuffer *, void *),
+ int (*cb)(void *, struct evhttp_request *, struct evbuffer *, void *),
void *cb_arg)
{
struct _evrpc_hooks *base = vbase;
}
static int
-evrpc_process_hooks(struct evrpc_hook_list *head,
+evrpc_process_hooks(struct evrpc_hook_list *head, void *ctx,
struct evhttp_request *req, struct evbuffer *evbuf)
{
struct evrpc_hook *hook;
TAILQ_FOREACH(hook, head, next) {
- if (hook->process(req, evbuf, hook->process_arg) == -1)
+ if (hook->process(ctx, req, evbuf, hook->process_arg) == -1)
return (-1);
}
return (0);
}
+static int evrpc_pause_request(void *vbase, void *ctx,
+ void (*cb)(void *, enum EVRPC_HOOK_RESULT));
+static void evrpc_request_cb_closure(void *, enum EVRPC_HOOK_RESULT);
+
static void
evrpc_request_cb(struct evhttp_request *req, void *arg)
{
struct evrpc *rpc = arg;
struct evrpc_req_generic *rpc_state = NULL;
+ int hook_res;
/* let's verify the outside parameters */
if (req->type != EVHTTP_REQ_POST ||
EVBUFFER_LENGTH(req->input_buffer) <= 0)
goto error;
+ rpc_state = event_calloc(1, sizeof(struct evrpc_req_generic));
+ if (rpc_state == NULL)
+ goto error;
+ rpc_state->rpc = rpc;
+ rpc_state->http_req = req;
+ rpc_state->rpc_data = NULL;
+ rpc_state->done = evrpc_request_done;
+
+
/*
* we might want to allow hooks to suspend the processing,
* but at the moment, we assume that they just act as simple
* filters.
*/
- if (evrpc_process_hooks(&rpc->base->input_hooks,
- req, req->input_buffer) == -1)
+ hook_res = evrpc_process_hooks(&rpc->base->input_hooks,
+ rpc_state, req, req->input_buffer);
+ switch (hook_res) {
+ case EVRPC_TERMINATE:
goto error;
+ case EVRPC_PAUSE:
+ evrpc_pause_request(rpc->base, rpc_state,
+ evrpc_request_cb_closure);
+ return;
+ case EVRPC_CONTINUE:
+ break;
+ default:
+ assert(hook_res == EVRPC_TERMINATE ||
+ hook_res == EVRPC_CONTINUE || hook_res == EVRPC_PAUSE);
+ }
- rpc_state = event_calloc(1, sizeof(struct evrpc_req_generic));
- if (rpc_state == NULL)
+ evrpc_request_cb_closure(rpc_state, EVRPC_CONTINUE);
+ return;
+
+error:
+ if (rpc_state != NULL)
+ evrpc_reqstate_free(rpc_state);
+ evhttp_send_error(req, HTTP_SERVUNAVAIL, "Service Error");
+ return;
+}
+
+static void
+evrpc_request_cb_closure(void *arg, enum EVRPC_HOOK_RESULT hook_res)
+{
+ struct evrpc_req_generic *rpc_state = arg;
+ struct evrpc *rpc = rpc_state->rpc;
+ struct evhttp_request *req = rpc_state->http_req;
+
+ if (hook_res == EVRPC_TERMINATE)
goto error;
/* let's check that we can parse the request */
if (rpc_state->request == NULL)
goto error;
- rpc_state->rpc = rpc;
-
if (rpc->request_unmarshal(
rpc_state->request, req->input_buffer) == -1) {
/* we failed to parse the request; that's a bummer */
if (rpc_state->reply == NULL)
goto error;
- rpc_state->http_req = req;
- rpc_state->done = evrpc_request_done;
-
/* give the rpc to the user; they can deal with it */
rpc->cb(rpc_state, rpc->cb_arg);
return;
error:
- evrpc_reqstate_free(rpc_state);
+ if (rpc_state != NULL)
+ evrpc_reqstate_free(rpc_state);
evhttp_send_error(req, HTTP_SERVUNAVAIL, "Service Error");
return;
}
+
void
evrpc_reqstate_free(struct evrpc_req_generic* rpc_state)
{
+ struct evrpc *rpc;
+ assert(rpc_state != NULL);
+ rpc = rpc_state->rpc;
+
/* clean up all memory */
- if (rpc_state != NULL) {
- struct evrpc *rpc = rpc_state->rpc;
-
- if (rpc_state->request != NULL)
- rpc->request_free(rpc_state->request);
- if (rpc_state->reply != NULL)
- rpc->reply_free(rpc_state->reply);
- event_free(rpc_state);
- }
+ if (rpc_state->request != NULL)
+ rpc->request_free(rpc_state->request);
+ if (rpc_state->reply != NULL)
+ rpc->reply_free(rpc_state->reply);
+ if (rpc_state->rpc_data != NULL)
+ evbuffer_free(rpc_state->rpc_data);
+ event_free(rpc_state);
}
+static void
+evrpc_request_done_closure(void *, enum EVRPC_HOOK_RESULT);
+
void
-evrpc_request_done(struct evrpc_req_generic* rpc_state)
+evrpc_request_done(struct evrpc_req_generic *rpc_state)
{
struct evhttp_request *req = rpc_state->http_req;
struct evrpc *rpc = rpc_state->rpc;
- struct evbuffer* data = NULL;
+ int hook_res;
if (rpc->reply_complete(rpc_state->reply) == -1) {
/* the reply was not completely filled in. error out */
goto error;
}
- if ((data = evbuffer_new()) == NULL) {
+ if ((rpc_state->rpc_data = evbuffer_new()) == NULL) {
/* out of memory */
goto error;
}
/* serialize the reply */
- rpc->reply_marshal(data, rpc_state->reply);
+ rpc->reply_marshal(rpc_state->rpc_data, rpc_state->reply);
/* do hook based tweaks to the request */
- if (evrpc_process_hooks(&rpc->base->output_hooks,
- req, data) == -1)
+ hook_res = evrpc_process_hooks(&rpc->base->output_hooks,
+ rpc_state, req, rpc_state->rpc_data);
+ switch (hook_res) {
+ case EVRPC_TERMINATE:
goto error;
+ case EVRPC_PAUSE:
+ if (evrpc_pause_request(rpc->base, rpc_state,
+ evrpc_request_done_closure) == -1)
+ goto error;
+ return;
+ case EVRPC_CONTINUE:
+ break;
+ default:
+ assert(hook_res == EVRPC_TERMINATE ||
+ hook_res == EVRPC_CONTINUE || hook_res == EVRPC_PAUSE);
+ }
- evhttp_send_reply(req, HTTP_OK, "OK", data);
+ evrpc_request_done_closure(rpc_state, EVRPC_CONTINUE);
+ return;
+
+error:
+ if (rpc_state != NULL)
+ evrpc_reqstate_free(rpc_state);
+ evhttp_send_error(req, HTTP_SERVUNAVAIL, "Service Error");
+ return;
+}
+
+static void
+evrpc_request_done_closure(void *arg, enum EVRPC_HOOK_RESULT hook_res)
+{
+ struct evrpc_req_generic *rpc_state = arg;
+ struct evhttp_request *req = rpc_state->http_req;
+
+ if (hook_res == EVRPC_TERMINATE)
+ goto error;
- evbuffer_free(data);
+ evhttp_send_reply(req, HTTP_OK, "OK", rpc_state->rpc_data);
evrpc_reqstate_free(rpc_state);
return;
error:
- if (data != NULL)
- evbuffer_free(data);
- evrpc_reqstate_free(rpc_state);
+ if (rpc_state != NULL)
+ evrpc_reqstate_free(rpc_state);
evhttp_send_error(req, HTTP_SERVUNAVAIL, "Service Error");
return;
}
+
/* Client implementation of RPC site */
static int evrpc_schedule_request(struct evhttp_connection *connection,
TAILQ_INIT(&pool->connections);
TAILQ_INIT(&pool->requests);
+ TAILQ_INIT(&pool->paused_requests);
+
TAILQ_INIT(&pool->input_hooks);
TAILQ_INIT(&pool->output_hooks);
{
struct evhttp_connection *connection;
struct evrpc_request_wrapper *request;
+ struct evrpc_hook_ctx *pause;
struct evrpc_hook *hook;
while ((request = TAILQ_FIRST(&pool->requests)) != NULL) {
TAILQ_REMOVE(&pool->requests, request, next);
- /* if this gets more complicated we need our own function */
evrpc_request_wrapper_free(request);
}
+ while ((pause = TAILQ_FIRST(&pool->paused_requests)) != NULL) {
+ TAILQ_REMOVE(&pool->paused_requests, pause, next);
+ event_free(pause);
+ }
+
while ((connection = TAILQ_FIRST(&pool->connections)) != NULL) {
TAILQ_REMOVE(&pool->connections, connection, next);
evhttp_connection_free(connection);
return (NULL);
}
+/*
+ * Prototypes responsible for evrpc scheduling and hooking
+ */
+
+static void evrpc_schedule_request_closure(void *ctx, enum EVRPC_HOOK_RESULT);
+
/*
* We assume that the ctx is no longer queued on the pool.
*/
struct evhttp_request *req = NULL;
struct evrpc_pool *pool = ctx->pool;
struct evrpc_status status;
- char *uri = NULL;
- int res = 0;
+ int hook_res = 0;
if ((req = evhttp_request_new(evrpc_reply_done, ctx)) == NULL)
goto error;
/* serialize the request data into the output buffer */
ctx->request_marshal(req->output_buffer, ctx->request);
- uri = evrpc_construct_uri(ctx->name);
- if (uri == NULL)
- goto error;
-
/* we need to know the connection that we might have to abort */
ctx->evcon = connection;
+ /* if we get paused we also need to know the request */
+ ctx->req = req;
+
/* apply hooks to the outgoing request */
- if (evrpc_process_hooks(&pool->output_hooks,
- req, req->output_buffer) == -1)
+ hook_res = evrpc_process_hooks(&pool->output_hooks,
+ ctx, req, req->output_buffer);
+
+ switch (hook_res) {
+ case EVRPC_TERMINATE:
+ goto error;
+ case EVRPC_PAUSE:
+ /* we need to be explicitly resumed */
+ if (evrpc_pause_request(pool, ctx,
+ evrpc_schedule_request_closure) == -1)
+ goto error;
+ return (0);
+ case EVRPC_CONTINUE:
+ /* we can just continue */
+ break;
+ default:
+ assert(hook_res == EVRPC_TERMINATE ||
+ hook_res == EVRPC_CONTINUE || hook_res == EVRPC_PAUSE);
+ }
+
+ evrpc_schedule_request_closure(ctx, EVRPC_CONTINUE);
+ return (0);
+
+error:
+ memset(&status, 0, sizeof(status));
+ status.error = EVRPC_STATUS_ERR_UNSTARTED;
+ (*ctx->cb)(&status, ctx->request, ctx->reply, ctx->cb_arg);
+ evrpc_request_wrapper_free(ctx);
+ return (-1);
+}
+
+static void
+evrpc_schedule_request_closure(void *arg, enum EVRPC_HOOK_RESULT hook_res)
+{
+ struct evrpc_request_wrapper *ctx = arg;
+ struct evhttp_connection *connection = ctx->evcon;
+ struct evhttp_request *req = ctx->req;
+ struct evrpc_pool *pool = ctx->pool;
+ struct evrpc_status status;
+ char *uri = NULL;
+ int res = 0;
+
+ if (hook_res == EVRPC_TERMINATE)
+ goto error;
+
+ uri = evrpc_construct_uri(ctx->name);
+ if (uri == NULL)
goto error;
if (pool->timeout > 0) {
if (res == -1)
goto error;
- return (0);
+ return;
error:
memset(&status, 0, sizeof(status));
status.error = EVRPC_STATUS_ERR_UNSTARTED;
(*ctx->cb)(&status, ctx->request, ctx->reply, ctx->cb_arg);
evrpc_request_wrapper_free(ctx);
- return (-1);
+}
+
+/* we just queue the paused request on the pool under the req object */
+static int
+evrpc_pause_request(void *vbase, void *ctx,
+ void (*cb)(void *, enum EVRPC_HOOK_RESULT))
+{
+ struct _evrpc_hooks *base = vbase;
+ struct evrpc_hook_ctx *pause = event_malloc(sizeof(*pause));
+ if (pause == NULL)
+ return (-1);
+
+ pause->ctx = ctx;
+ pause->cb = cb;
+
+ TAILQ_INSERT_TAIL(&base->pause_requests, pause, next);
+ return (0);
+}
+
+int
+evrpc_resume_request(void *vbase, void *ctx, enum EVRPC_HOOK_RESULT res)
+{
+ struct _evrpc_hooks *base = vbase;
+ struct evrpc_pause_list *head = &base->pause_requests;
+ struct evrpc_hook_ctx *pause;
+
+ TAILQ_FOREACH(pause, head, next) {
+ if (pause->ctx == ctx)
+ break;
+ }
+
+ if (pause == NULL)
+ return (-1);
+
+ (*pause->cb)(pause->ctx, res);
+ TAILQ_REMOVE(head, pause, next);
+ return (0);
}
int
return (0);
}
+static void
+evrpc_reply_done_closure(void *, enum EVRPC_HOOK_RESULT);
+
static void
evrpc_reply_done(struct evhttp_request *req, void *arg)
{
struct evrpc_request_wrapper *ctx = arg;
struct evrpc_pool *pool = ctx->pool;
- struct evrpc_status status;
- int res = -1;
+ int hook_res;
/* cancel any timeout we might have scheduled */
event_del(&ctx->ev_timeout);
+ /* if we get paused we also need to know the request */
+ ctx->req = req;
+
+ /* we need to get the reply now */
+ if (req == NULL) {
+ evrpc_reply_done_closure(ctx, EVRPC_CONTINUE);
+ return;
+ }
+
+ /* apply hooks to the incoming request */
+ hook_res = evrpc_process_hooks(&pool->input_hooks,
+ ctx, req, req->input_buffer);
+
+ switch (hook_res) {
+ case EVRPC_TERMINATE:
+ case EVRPC_CONTINUE:
+ evrpc_reply_done_closure(ctx, hook_res);
+ return;
+ case EVRPC_PAUSE:
+ evrpc_pause_request(pool, ctx, evrpc_reply_done_closure);
+ return;
+ default:
+ assert(hook_res == EVRPC_TERMINATE ||
+ hook_res == EVRPC_CONTINUE || hook_res == EVRPC_PAUSE);
+ }
+}
+
+static void
+evrpc_reply_done_closure(void *arg, enum EVRPC_HOOK_RESULT hook_res)
+{
+ struct evrpc_request_wrapper *ctx = arg;
+ struct evhttp_request *req = ctx->req;
+ struct evrpc_pool *pool = ctx->pool;
+ struct evrpc_status status;
+ int res = -1;
+
memset(&status, 0, sizeof(status));
status.http_req = req;
/* we need to get the reply now */
- if (req != NULL) {
- /* apply hooks to the incoming request */
- if (evrpc_process_hooks(&pool->input_hooks,
- req, req->input_buffer) == -1) {
- status.error = EVRPC_STATUS_ERR_HOOKABORTED;
- res = -1;
- } else {
- res = ctx->reply_unmarshal(ctx->reply,
- req->input_buffer);
- if (res == -1) {
- status.error = EVRPC_STATUS_ERR_BADPAYLOAD;
- }
- }
- } else {
+ if (req == NULL) {
status.error = EVRPC_STATUS_ERR_TIMEOUT;
+ } else if (hook_res == EVRPC_TERMINATE) {
+ status.error = EVRPC_STATUS_ERR_HOOKABORTED;
+ } else {
+ res = ctx->reply_unmarshal(ctx->reply, req->input_buffer);
+ if (res == -1)
+ status.error = EVRPC_STATUS_ERR_BADPAYLOAD;
}
if (res == -1) {
}
static int
-rpc_hook_add_header(struct evhttp_request *req,
+rpc_hook_add_header(void *ctx, struct evhttp_request *req,
struct evbuffer *evbuf, void *arg)
{
const char *hook_type = arg;
}
static int
-rpc_hook_remove_header(struct evhttp_request *req,
+rpc_hook_remove_header(void *ctx, struct evhttp_request *req,
struct evbuffer *evbuf, void *arg)
{
const char *header = evhttp_find_header(req->input_headers, "X-Hook");
evrpc_pool_free(pool);
evhttp_free(http);
+
+ need_input_hook = 0;
+ need_output_hook = 0;
}
/*
event_loopexit(NULL);
}
+/* we just pause the rpc and continue it in the next callback */
+
+struct _rpc_hook_ctx {
+ void *vbase;
+ void *ctx;
+};
+
+static void
+rpc_hook_pause_cb(int fd, short what, void *arg)
+{
+ struct _rpc_hook_ctx *ctx = arg;
+ evrpc_resume_request(ctx->vbase, ctx->ctx, EVRPC_CONTINUE);
+}
+
+static int
+rpc_hook_pause(void *ctx, struct evhttp_request *req, struct evbuffer *evbuf,
+ void *arg)
+{
+ struct _rpc_hook_ctx *tmp = malloc(sizeof(*tmp));
+ struct timeval tv;
+
+ assert(tmp != NULL);
+ tmp->vbase = arg;
+ tmp->ctx = ctx;
+
+ memset(&tv, 0, sizeof(tv));
+ event_once(-1, EV_TIMEOUT, rpc_hook_pause_cb, tmp, &tv);
+ return EVRPC_PAUSE;
+}
+
+static void
+rpc_basic_client_with_pause(void)
+{
+ short port;
+ struct evhttp *http = NULL;
+ struct evrpc_base *base = NULL;
+ struct evrpc_pool *pool = NULL;
+ struct msg *msg;
+ struct kill *kill;
+
+ fprintf(stdout, "Testing RPC Client with pause hooks: ");
+
+ rpc_setup(&http, &port, &base);
+
+ assert(evrpc_add_hook(base, INPUT, rpc_hook_pause, base));
+ assert(evrpc_add_hook(base, OUTPUT, rpc_hook_pause, base));
+
+ pool = rpc_pool_with_connection(port);
+
+ assert(evrpc_add_hook(pool, INPUT, rpc_hook_pause, pool));
+ assert(evrpc_add_hook(pool, OUTPUT, rpc_hook_pause, pool));
+
+ /* set up the basic message */
+ msg = msg_new();
+ EVTAG_ASSIGN(msg, from_name, "niels");
+ EVTAG_ASSIGN(msg, to_name, "tester");
+
+ kill = kill_new();
+
+ EVRPC_MAKE_REQUEST(Message, pool, msg, kill, GotKillCb, NULL);
+
+ test_ok = 0;
+
+ event_dispatch();
+
+ if (test_ok != 1) {
+ fprintf(stdout, "FAILED (1)\n");
+ exit(1);
+ }
+
+ /* we do it twice to make sure that reuse works correctly */
+ kill_clear(kill);
+
+ EVRPC_MAKE_REQUEST(Message, pool, msg, kill, GotKillCb, NULL);
+
+ event_dispatch();
+
+ rpc_teardown(base);
+
+ if (test_ok != 2) {
+ fprintf(stdout, "FAILED (2)\n");
+ exit(1);
+ }
+
+ fprintf(stdout, "OK\n");
+
+ msg_free(msg);
+ kill_free(kill);
+
+ evrpc_pool_free(pool);
+ evhttp_free(http);
+}
+
static void
rpc_client_timeout(void)
{
rpc_basic_message();
rpc_basic_client();
rpc_basic_queued_client();
+ rpc_basic_client_with_pause();
rpc_client_timeout();
}
projects / libevent / commitdiff