src/chxj_serf.c

   1 /*
   2  * Copyright (C) 2005-2008 Atsushi Konno All rights reserved.
   3  * Copyright (C) 2005 QSDN,Inc. All rights reserved.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  */
  17 #include "chxj_serf.h"
  18 #include "mod_chxj.h"
  19 #include "apr_pools.h"
  20
  21 typedef struct __app_ctx_t     app_ctx_t;
  22 typedef struct __handler_ctx_t handler_ctx_t;
  23
  24 struct __app_ctx_t {
  25   int                 ssl_flag;
  26   serf_ssl_context_t  *ssl_ctx;
  27   serf_bucket_alloc_t *bkt_alloc;
  28 };
  29
  30 struct __handler_ctx_t {
  31 #if APR_MAJOR_VERSION > 0
  32   apr_uint32_t requests_outstanding;
  33 #else
  34   apr_atomic_t requests_outstanding;
  35 #endif
  36
  37   serf_response_acceptor_t acceptor;
  38   app_ctx_t                *acceptor_ctx;
  39
  40   serf_response_handler_t  handler;
  41
  42   const char *host;
  43   const char *method;
  44   const char *path;
  45   const char *user_agent;
  46
  47   apr_status_t rv;
  48   const char *reason;
  49
  50   char *response;
  51   apr_size_t response_len;
  52   char *post_data;
  53   apr_size_t post_data_len;
  54   apr_table_t *headers_out;
  55   apr_pool_t *pool;
  56   request_rec *r;
  57 };
  58
  59 char *default_chxj_serf_get(request_rec *r, apr_pool_t *ppool, const char *url_path, int set_headers_flag, apr_size_t *response_len);
  60 char *(*chxj_serf_get)(request_rec *r, apr_pool_t *ppool, const char *url_path, int set_headers_flag, apr_size_t *response_len) = default_chxj_serf_get;
  61 char *default_chxj_serf_post(request_rec *r, apr_pool_t *ppool, const char *url_path, char *post_data, apr_size_t post_data_len, int set_headers_flag, apr_size_t *response_len);
  62 char *(*chxj_serf_post)(request_rec *r, apr_pool_t *ppool, const char *url_path, char *post_data, apr_size_t post_data_len, int set_headers_flag, apr_size_t *response_len) = default_chxj_serf_post;
  63
  64
  65 void
  66 s_init(apr_pool_t *ppool, apr_pool_t **pool)
  67 {
  68   apr_pool_create(pool, ppool);
  69   apr_atomic_init(*pool);
  70 }
  71
  72
  73
  74 static serf_bucket_t *
  75 s_connection_setup(apr_socket_t *skt, void *setup_ctx, apr_pool_t *UNUSED(pool))
  76 {
  77   serf_bucket_t  *c;
  78   app_ctx_t      *ctx = (app_ctx_t *)setup_ctx;
  79
  80   c = serf_bucket_socket_create(skt, ctx->bkt_alloc);
  81   if (ctx->ssl_flag) {
  82     c = serf_bucket_ssl_decrypt_create(c, ctx->ssl_ctx, ctx->bkt_alloc);
  83     if (!ctx->ssl_ctx) {
  84       ctx->ssl_ctx = serf_bucket_ssl_decrypt_context_get(c);
  85       serf_ssl_use_default_certificates(ctx->ssl_ctx);
  86       serf_ssl_server_cert_callback_set(ctx->ssl_ctx, NULL, NULL);
  87     }
  88     return c;
  89   }
  90   return c;
  91 }
  92
  93
  94 static void
  95 s_connection_closed(serf_connection_t *UNUSED(conn), void *UNUSED(closed_baton), apr_status_t UNUSED(why), apr_pool_t *UNUSED(pool))
  96 {
  97   /* nothing */
  98 }
  99
 100
 101 static serf_bucket_t *
 102 s_accept_response(serf_request_t *request, serf_bucket_t *stream, void *UNUSED(acceptor_baton), apr_pool_t *UNUSED(pool))
 103 {
 104     serf_bucket_alloc_t *bkt_alloc;
 105     serf_bucket_t       *c;
 106
 107     bkt_alloc = serf_request_get_alloc(request);
 108     c = serf_bucket_barrier_create(stream, bkt_alloc);
 109     return serf_bucket_response_create(c, bkt_alloc);
 110 }
 111
 112
 113 static apr_status_t
 114 s_handle_response(serf_request_t *UNUSED(request), serf_bucket_t *response, void *handler_ctx, apr_pool_t *UNUSED(pool))
 115 {
 116   const char      *data;
 117   apr_size_t      len;
 118   serf_status_line sl;
 119   apr_status_t     rv;
 120   handler_ctx_t  *ctx = handler_ctx;
 121
 122   rv = serf_bucket_response_status(response, &sl);
 123   if (rv != APR_SUCCESS) {
 124     if (APR_STATUS_IS_EAGAIN(rv)) {
 125       return rv;
 126     }
 127     ctx->rv = rv;
 128     apr_atomic_dec32(&ctx->requests_outstanding);
 129     return rv;
 130   }
 131   ctx->reason = sl.reason;
 132
 133   while (1) {
 134     rv = serf_bucket_read(response, 2048, &data, &len);
 135     if (SERF_BUCKET_READ_ERROR(rv)) {
 136       ctx->rv = rv;
 137       apr_atomic_dec32(&ctx->requests_outstanding);
 138       return rv;
 139     }
 140     if (APR_STATUS_IS_EAGAIN(rv)) {
 141       /* 0 byte return if EAGAIN returned. */
 142       DBG(ctx->r, "REQ[%X] end of s_handle_response() (EAGAIN) len:[%d]", (unsigned int)(apr_size_t)ctx->r, (int)len);
 143       return rv;
 144     }
 145
 146     if (len > 0) {
 147       if (! ctx->response) {
 148         ctx->response = apr_palloc(ctx->pool, len);
 149         ctx->response[0] = 0;
 150         ctx->response_len = 0;
 151       }
 152       else {
 153         char *tmp = apr_palloc(ctx->pool, ctx->response_len);
 154         memcpy(tmp, ctx->response, ctx->response_len);
 155         ctx->response = apr_palloc(ctx->pool, ctx->response_len + len);
 156         memcpy(ctx->response, tmp, ctx->response_len);
 157       }
 158       memcpy(&ctx->response[ctx->response_len], data, len);
 159       ctx->response_len += len;
 160       ctx->response[ctx->response_len] = 0;
 161     }
 162
 163     if (APR_STATUS_IS_EOF(rv)) {
 164       serf_bucket_t *hdrs;
 165       char *tmp_headers = "";
 166       hdrs = serf_bucket_response_get_headers(response);
 167       while (1) {
 168         rv = serf_bucket_read(hdrs, 2048, &data, &len);
 169         if (SERF_BUCKET_READ_ERROR(rv))
 170           return rv;
 171         tmp_headers = apr_pstrcat(ctx->pool, tmp_headers, apr_psprintf(ctx->pool , "%.*s", (unsigned int)len, data), NULL);
 172         if (APR_STATUS_IS_EOF(rv)) {
 173           break;
 174         }
 175       }
 176       ctx->headers_out = apr_table_make(ctx->pool, 0);
 177
 178       char *pstat;
 179       char *pair = NULL;
 180       for (;;) {
 181         pair = apr_strtok(tmp_headers, "\n", &pstat);
 182         if (!pair) break;
 183         tmp_headers = NULL;
 184         char *key;
 185         char *val;
 186
 187         char *tpair = apr_pstrdup(ctx->pool, pair);
 188         key = tpair;
 189         val = strchr(tpair, ':');
 190         if (val) {
 191           *val = 0;
 192           val++;
 193           key = qs_trim_string(ctx->pool, key);
 194           val = qs_trim_string(ctx->pool, val);
 195           DBG(ctx->r, "key:[%s], val:[%s]", key, val);
 196           apr_table_add(ctx->headers_out, key, val);
 197         }
 198       }
 199       ctx->rv = APR_SUCCESS;
 200       apr_atomic_dec32(&ctx->requests_outstanding);
 201       DBG(ctx->r, "end of s_handle_response()(NORMAL)");
 202       return APR_EOF;
 203     }
 204
 205     if (APR_STATUS_IS_EAGAIN(rv)) {
 206       DBG(ctx->r, "end of s_handle_response() (EAGAIN)");
 207       return rv;
 208     }
 209   }
 210 }
 211
 212 static apr_status_t
 213 s_setup_request(serf_request_t           *request,
 214                 void                     *setup_ctx,
 215                 serf_bucket_t            **req_bkt,
 216                 serf_response_acceptor_t *acceptor,
 217                 void                     **acceptor_ctx,
 218                 serf_response_handler_t  *handler,
 219                 void                     **handler_ctx,
 220                 apr_pool_t               *UNUSED(pool))
 221 {
 222   handler_ctx_t *ctx = setup_ctx;
 223   serf_bucket_t *hdrs_bkt;
 224   serf_bucket_t *body_bkt = NULL;
 225   request_rec *r = ctx->r;
 226   int ii;
 227
 228   if (ctx->post_data) {
 229     body_bkt = serf_bucket_simple_create(ctx->post_data, ctx->post_data_len, NULL, NULL, serf_request_get_alloc(request));
 230   }
 231
 232   *req_bkt = serf_bucket_request_create(ctx->method, ctx->path, body_bkt, serf_request_get_alloc(request));
 233   hdrs_bkt = serf_bucket_request_get_headers(*req_bkt);
 234
 235
 236   apr_array_header_t *headers = (apr_array_header_t*)apr_table_elts(r->headers_in);
 237   apr_table_entry_t  *hentryp = (apr_table_entry_t*)headers->elts;
 238   for (ii=headers->nelts-1; ii>=0; ii--) {
 239     serf_bucket_headers_setc(hdrs_bkt, hentryp[ii].key, hentryp[ii].val);
 240     DBG(ctx->r, "REQ[%X] REQUEST key:[%s], val:[%s]", (unsigned int)(apr_size_t)ctx->r, hentryp[ii].key, hentryp[ii].val);
 241   }
 242   if (ctx->post_data) {
 243     serf_bucket_headers_setc(hdrs_bkt, "X-Chxj-Forward", "Done");
 244     serf_bucket_headers_setc(hdrs_bkt, "X-Chxj-Content-Length", apr_psprintf(r->pool, "%" APR_SIZE_T_FMT , ctx->post_data_len));
 245     DBG(ctx->r, "REQ[%X] REQUEST key:[%s], val:[%s]", (unsigned int)(apr_size_t)ctx->r, "X-Chxj-Forward", "Done");
 246     DBG(ctx->r, "REQ[%X] REQUEST key:[%s], val:[%s]", (unsigned int)(apr_size_t)ctx->r, "X-Chxj-Content-Length", apr_psprintf(r->pool, "%" APR_SIZE_T_FMT, ctx->post_data_len));
 247
 248   }
 249   DBG(ctx->r, "REQ[%X] REQUEST Content-Length:[%s]", (unsigned int)(apr_size_t)r, serf_bucket_headers_get(hdrs_bkt, "Content-Length"));
 250
 251   apr_atomic_inc32(&(ctx->requests_outstanding));
 252   if (ctx->acceptor_ctx->ssl_flag) {
 253     serf_bucket_alloc_t *req_alloc;
 254     app_ctx_t *app_ctx = ctx->acceptor_ctx;
 255
 256     req_alloc = serf_request_get_alloc(request);
 257
 258     if (app_ctx->ssl_ctx == NULL) {
 259       *req_bkt = serf_bucket_ssl_encrypt_create(*req_bkt, NULL, app_ctx->bkt_alloc);
 260       app_ctx->ssl_ctx = serf_bucket_ssl_encrypt_context_get(*req_bkt);
 261     }
 262     else {
 263       *req_bkt = serf_bucket_ssl_encrypt_create(*req_bkt, app_ctx->ssl_ctx, app_ctx->bkt_alloc);
 264     }
 265   }
 266   *acceptor       = ctx->acceptor;
 267   *acceptor_ctx   = ctx->acceptor_ctx;
 268   *handler        = ctx->handler;
 269   *handler_ctx    = ctx;
 270
 271   return APR_SUCCESS;
 272 }
 273
 274 char *
 275 default_chxj_serf_get(request_rec *r, apr_pool_t *ppool, const char *url_path, int set_headers_flag, apr_size_t *response_len)
 276 {
 277   apr_pool_t *pool;
 278   apr_uri_t url;
 279   apr_status_t rv;
 280   apr_sockaddr_t *address = NULL;
 281
 282   serf_context_t *context;
 283   serf_connection_t *connection;
 284
 285   app_ctx_t app_ctx;
 286   handler_ctx_t handler_ctx;
 287   char *ret;
 288
 289
 290   s_init(ppool, &pool);
 291
 292   apr_uri_parse(pool, url_path, &url);
 293   if (!url.port) {
 294     url.port = apr_uri_port_of_scheme(url.scheme);
 295   }
 296   if (!url.port) {
 297     url.port = 80;
 298   }
 299   if (!url.path) {
 300     url.path = "/";
 301   }
 302   if (!url.hostname) {
 303     url.hostname = "localhost";
 304   }
 305
 306   rv = apr_sockaddr_info_get(&address, url.hostname, APR_UNSPEC, url.port, 0, pool);
 307   if (rv != APR_SUCCESS) {
 308     char buf[256];
 309     ERR(r, "apr_sockaddr_info_get() failed: rv:[%d|%s]", rv, apr_strerror(rv, buf, 256));
 310     return NULL;
 311   }
 312   memset(&app_ctx, 0, sizeof(app_ctx_t));
 313
 314   app_ctx.bkt_alloc = serf_bucket_allocator_create(pool, NULL, NULL);
 315   if (strcasecmp(url.scheme, "https") == 0) {
 316     app_ctx.ssl_flag = 1;
 317   }
 318
 319   context = serf_context_create(pool);
 320   connection = serf_connection_create(context, address, s_connection_setup, &app_ctx, s_connection_closed, &app_ctx, pool);
 321
 322   memset(&handler_ctx, 0, sizeof(handler_ctx_t));
 323   handler_ctx.requests_outstanding = 0;
 324   handler_ctx.host = url.hostinfo;
 325   handler_ctx.method = "GET";
 326   handler_ctx.path = url.path;
 327   handler_ctx.user_agent = (char *)apr_table_get(r->headers_in, "User-Agent");
 328   handler_ctx.post_data = NULL;
 329   handler_ctx.post_data_len = 0;
 330
 331   handler_ctx.acceptor     = s_accept_response;
 332   handler_ctx.acceptor_ctx = &app_ctx;
 333   handler_ctx.handler      = s_handle_response;
 334   handler_ctx.pool         = pool;
 335   handler_ctx.r            = r;
 336   handler_ctx.response_len = 0;
 337   handler_ctx.response     = NULL;
 338
 339   serf_connection_request_create(connection, s_setup_request, &handler_ctx);
 340
 341   while (1) {
 342     rv = serf_context_run(context, SERF_DURATION_FOREVER, pool);
 343     if (APR_STATUS_IS_TIMEUP(rv))
 344       continue;
 345     if (rv) {
 346       char buf[200];
 347       ERR(r, "Error running context: (%d) %s\n", rv, apr_strerror(rv, buf, sizeof(buf)));
 348       break;
 349     }
 350     if (!apr_atomic_read32(&handler_ctx.requests_outstanding)) {
 351       if (handler_ctx.rv != APR_SUCCESS) {
 352         char buf[200];
 353         ERR(r, "Error running context: (%d) %s\n", handler_ctx.rv, apr_strerror(handler_ctx.rv, buf, sizeof(buf)));
 354       }
 355       break;
 356     }
 357   }
 358
 359   serf_connection_close(connection);
 360   ret = apr_pstrdup(ppool, handler_ctx.response);
 361   if (set_headers_flag) {
 362     r->headers_out = apr_table_copy(pool, handler_ctx.headers_out);
 363     *response_len = handler_ctx.response_len;
 364     char *contentType = (char *)apr_table_get(handler_ctx.headers_out, "Content-Type");
 365     if (contentType) {
 366       chxj_set_content_type(r, contentType);
 367     }
 368   }
 369   return ret;
 370 }
 371
 372
 373 char *
 374 default_chxj_serf_post(request_rec *r, apr_pool_t *ppool, const char *url_path, char *post_data, apr_size_t post_data_len, int set_headers_flag, apr_size_t *response_len)
 375 {
 376   apr_pool_t *pool;
 377   apr_uri_t url;
 378   apr_status_t rv;
 379   apr_sockaddr_t *address = NULL;
 380
 381   serf_context_t *context;
 382   serf_connection_t *connection;
 383
 384   app_ctx_t app_ctx;
 385   handler_ctx_t handler_ctx;
 386   char *ret;
 387
 388   DBG(r, "start chxj_serf_post()");
 389
 390
 391   s_init(ppool, &pool);
 392
 393   apr_uri_parse(pool, url_path, &url);
 394   if (!url.port) {
 395     url.port = apr_uri_port_of_scheme(url.scheme);
 396   }
 397   if (!url.port) {
 398     url.port = 80;
 399   }
 400   if (!url.path) {
 401     url.path = "/";
 402   }
 403   if (!url.hostname) {
 404     url.hostname = "localhost";
 405   }
 406
 407   rv = apr_sockaddr_info_get(&address, url.hostname, APR_UNSPEC, url.port, 0, pool);
 408   if (rv != APR_SUCCESS) {
 409     char buf[256];
 410     ERR(r, "apr_sockaddr_info_get() failed: rv:[%d|%s]", rv, apr_strerror(rv, buf, 256));
 411     return NULL;
 412   }
 413   memset(&app_ctx, 0, sizeof(app_ctx_t));
 414
 415   app_ctx.bkt_alloc = serf_bucket_allocator_create(pool, NULL, NULL);
 416   if (strcasecmp(url.scheme, "https") == 0) {
 417     app_ctx.ssl_flag = 1;
 418   }
 419
 420   context = serf_context_create(pool);
 421   connection = serf_connection_create(context, address, s_connection_setup, &app_ctx, s_connection_closed, &app_ctx, pool);
 422
 423   memset(&handler_ctx, 0, sizeof(handler_ctx_t));
 424   handler_ctx.requests_outstanding = 0;
 425   handler_ctx.host = url.hostinfo;
 426   handler_ctx.method = "POST";
 427   handler_ctx.path = url.path;
 428   handler_ctx.user_agent = (char *)apr_table_get(r->headers_in, "User-Agent");
 429   handler_ctx.post_data = post_data;
 430   handler_ctx.post_data_len = post_data_len;
 431
 432   handler_ctx.acceptor     = s_accept_response;
 433   handler_ctx.acceptor_ctx = &app_ctx;
 434   handler_ctx.handler      = s_handle_response;
 435   handler_ctx.pool         = pool;
 436   handler_ctx.r            = r;
 437   handler_ctx.response_len = 0;
 438   handler_ctx.response     = NULL;
 439
 440   serf_connection_request_create(connection, s_setup_request, &handler_ctx);
 441
 442   while (1) {
 443     rv = serf_context_run(context, SERF_DURATION_FOREVER, pool);
 444     if (APR_STATUS_IS_TIMEUP(rv))
 445       continue;
 446     if (rv) {
 447       char buf[200];
 448       ERR(r, "Error running context: (%d) %s\n", rv, apr_strerror(rv, buf, sizeof(buf)));
 449       break;
 450     }
 451     if (!apr_atomic_read32(&handler_ctx.requests_outstanding)) {
 452       if (handler_ctx.rv != APR_SUCCESS) {
 453         char buf[200];
 454         ERR(r, "Error running context: (%d) %s\n", handler_ctx.rv, apr_strerror(handler_ctx.rv, buf, sizeof(buf)));
 455       }
 456       break;
 457     }
 458   }
 459
 460   DBG(r, "end of serf request");
 461   DBG(r, "response:[%s][%" APR_SIZE_T_FMT "]", handler_ctx.response, handler_ctx.response_len);
 462   serf_connection_close(connection);
 463   ret = apr_pstrdup(ppool, handler_ctx.response);
 464   if (set_headers_flag) {
 465     r->headers_out = apr_table_copy(pool, handler_ctx.headers_out);
 466     *response_len = handler_ctx.response_len;
 467     char *contentType = (char *)apr_table_get(handler_ctx.headers_out, "Content-Type");
 468     if (contentType) {
 469       DBG(r, "response content type[%s]", contentType);
 470       chxj_set_content_type(r, apr_pstrdup(r->pool, contentType));
 471     }
 472   }
 473   DBG(r, "end chxj_serf_post()");
 474   return ret;
 475 }
 476 /*
 477  * vim:ts=2 et
 478  */