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, CHXJ_HTTP_USER_AGENT);
 335   if (!handler_ctx.user_agent) {
 336     handler_ctx.user_agent = (char *)apr_table_get(r->headers_in, HTTP_USER_AGENT);
 337   }
 338   handler_ctx.post_data = NULL;
 339   handler_ctx.post_data_len = 0;
 340
 341   handler_ctx.acceptor     = s_accept_response;
 342   handler_ctx.acceptor_ctx = &app_ctx;
 343   handler_ctx.handler      = s_handle_response;
 344   handler_ctx.pool         = pool;
 345   handler_ctx.r            = r;
 346   handler_ctx.response_len = 0;
 347   handler_ctx.response     = NULL;
 348
 349   serf_connection_request_create(connection, s_setup_request, &handler_ctx);
 350
 351   while (1) {
 352     rv = serf_context_run(context, SERF_DURATION_FOREVER, pool);
 353     if (APR_STATUS_IS_TIMEUP(rv))
 354       continue;
 355     if (rv) {
 356       char buf[200];
 357       ERR(r, "Error running context: (%d) %s\n", rv, apr_strerror(rv, buf, sizeof(buf)));
 358       break;
 359     }
 360     if (!apr_atomic_read32(&handler_ctx.requests_outstanding)) {
 361       if (handler_ctx.rv != APR_SUCCESS) {
 362         char buf[200];
 363         ERR(r, "Error running context: (%d) %s\n", handler_ctx.rv, apr_strerror(handler_ctx.rv, buf, sizeof(buf)));
 364       }
 365       break;
 366     }
 367   }
 368
 369   serf_connection_close(connection);
 370   ret = apr_pstrdup(ppool, handler_ctx.response);
 371   if (set_headers_flag) {
 372     r->headers_out = apr_table_copy(pool, handler_ctx.headers_out);
 373     *response_len = handler_ctx.response_len;
 374     char *contentType = (char *)apr_table_get(handler_ctx.headers_out, "Content-Type");
 375     if (contentType) {
 376       chxj_set_content_type(r, contentType);
 377     }
 378   }
 379   return ret;
 380 }
 381
 382
 383 char *
 384 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)
 385 {
 386   apr_pool_t *pool;
 387   apr_uri_t url;
 388   apr_status_t rv;
 389   apr_sockaddr_t *address = NULL;
 390
 391   serf_context_t *context;
 392   serf_connection_t *connection;
 393
 394   app_ctx_t app_ctx;
 395   handler_ctx_t handler_ctx;
 396   char *ret;
 397
 398   DBG(r, "REQ:[%X] start chxj_serf_post()", (unsigned int)(apr_size_t)r);
 399
 400
 401   s_init(ppool, &pool);
 402
 403   apr_uri_parse(pool, url_path, &url);
 404   if (!url.port) {
 405     url.port = apr_uri_port_of_scheme(url.scheme);
 406   }
 407   if (!url.port) {
 408     url.port = 80;
 409   }
 410   if (!url.path) {
 411     url.path = "/";
 412   }
 413   if (!url.hostname) {
 414     url.hostname = "localhost";
 415   }
 416   if (url.query) {
 417     url.path = apr_psprintf(pool, "%s?%s", url.path, url.query);
 418   }
 419
 420   rv = apr_sockaddr_info_get(&address, url.hostname, APR_UNSPEC, url.port, 0, pool);
 421   if (rv != APR_SUCCESS) {
 422     char buf[256];
 423     ERR(r, "apr_sockaddr_info_get() failed: rv:[%d|%s]", rv, apr_strerror(rv, buf, 256));
 424     return NULL;
 425   }
 426   memset(&app_ctx, 0, sizeof(app_ctx_t));
 427
 428   app_ctx.bkt_alloc = serf_bucket_allocator_create(pool, NULL, NULL);
 429   if (strcasecmp(url.scheme, "https") == 0) {
 430     app_ctx.ssl_flag = 1;
 431   }
 432
 433   context = serf_context_create(pool);
 434   connection = serf_connection_create(context, address, s_connection_setup, &app_ctx, s_connection_closed, &app_ctx, pool);
 435
 436   memset(&handler_ctx, 0, sizeof(handler_ctx_t));
 437   handler_ctx.requests_outstanding = 0;
 438   handler_ctx.host = url.hostinfo;
 439   handler_ctx.method = "POST";
 440   handler_ctx.path = url.path;
 441   handler_ctx.user_agent = (char *)apr_table_get(r->headers_in, CHXJ_HTTP_USER_AGENT);
 442   if (! handler_ctx.user_agent) {
 443     handler_ctx.user_agent = (char *)apr_table_get(r->headers_in, HTTP_USER_AGENT);
 444   }
 445   handler_ctx.post_data = post_data;
 446   handler_ctx.post_data_len = post_data_len;
 447
 448   handler_ctx.acceptor     = s_accept_response;
 449   handler_ctx.acceptor_ctx = &app_ctx;
 450   handler_ctx.handler      = s_handle_response;
 451   handler_ctx.pool         = pool;
 452   handler_ctx.r            = r;
 453   handler_ctx.response_len = 0;
 454   handler_ctx.response     = NULL;
 455
 456   serf_connection_request_create(connection, s_setup_request, &handler_ctx);
 457
 458   while (1) {
 459     rv = serf_context_run(context, SERF_DURATION_FOREVER, pool);
 460     if (APR_STATUS_IS_TIMEUP(rv))
 461       continue;
 462     if (rv) {
 463       char buf[200];
 464       ERR(r, "Error running context: (%d) %s\n", rv, apr_strerror(rv, buf, sizeof(buf)));
 465       break;
 466     }
 467     if (!apr_atomic_read32(&handler_ctx.requests_outstanding)) {
 468       if (handler_ctx.rv != APR_SUCCESS) {
 469         char buf[200];
 470         ERR(r, "Error running context: (%d) %s\n", handler_ctx.rv, apr_strerror(handler_ctx.rv, buf, sizeof(buf)));
 471       }
 472       break;
 473     }
 474   }
 475
 476   DBG(r, "end of serf request");
 477   DBG(r, "response_code:[%d]", handler_ctx.response_code);
 478   DBG(r, "response:[%s][%" APR_SIZE_T_FMT "]", handler_ctx.response, handler_ctx.response_len);
 479   serf_connection_close(connection);
 480   if (handler_ctx.response) {
 481     ret = apr_pstrdup(ppool, handler_ctx.response);
 482   }
 483   else {
 484     ret = apr_pstrdup(ppool, "");
 485   }
 486   if (set_headers_flag) {
 487     r->headers_out = apr_table_copy(pool, handler_ctx.headers_out);
 488     *response_len = handler_ctx.response_len;
 489     char *contentType = (char *)apr_table_get(handler_ctx.headers_out, "Content-Type");
 490     if (contentType) {
 491       DBG(r, "response content type[%s]", contentType);
 492       chxj_set_content_type(r, apr_pstrdup(r->pool, contentType));
 493     }
 494   }
 495   *response_code = handler_ctx.response_code;
 496   DBG(r, "REQ:[%X] end chxj_serf_post()", (unsigned int)(apr_size_t)r);
 497   return ret;
 498 }
 499 /*
 500  * vim:ts=2 et
 501  */