src/chxj_serf.c

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