One Level Up Top Level

src/stream/ngx_stream_handler.c - nginx source code

Functions defined

Source code


  2. /*
  3. * Copyright (C) Roman Arutyunyan
  4. * Copyright (C) Nginx, Inc.
  5. */


  8. #include <ngx_config.h>
  9. #include <ngx_core.h>
  10. #include <ngx_event.h>
  11. #include <ngx_stream.h>


  14. static void ngx_stream_log_session(ngx_stream_session_t *s);
  15. static void ngx_stream_close_connection(ngx_connection_t *c);
  16. static u_char *ngx_stream_log_error(ngx_log_t *log, u_char *buf, size_t len);
  17. static void ngx_stream_proxy_protocol_handler(ngx_event_t *rev);


  20. void
  21. ‌ngx_stream_init_connection(ngx_connection_t *c)
  22. {
  23.     u_char                        text[NGX_SOCKADDR_STRLEN];
  24.     size_t                        len;
  25.     ngx_uint_t                    i;
  26.     ngx_time_t                   *tp;
  27.     ngx_event_t                  *rev;
  28.     struct sockaddr              *sa;
  29.     ngx_stream_port_t            *port;
  30.     struct sockaddr_in           *sin;
  31.     ngx_stream_in_addr_t         *addr;
  32.     ngx_stream_session_t         *s;
  33.     ngx_stream_conf_ctx_t        *ctx;
  34.     ngx_stream_addr_conf_t       *addr_conf;
  35. #if (NGX_HAVE_INET6)
  36.     struct sockaddr_in6          *sin6;
  37.     ngx_stream_in6_addr_t        *addr6;
  38. #endif
  39.     ngx_stream_core_srv_conf_t   *cscf;
  40.     ngx_stream_core_main_conf_t  *cmcf;

  42.     /* find the server configuration for the address:port */

  44.     port = c->listening->servers;

  46.     if (port->naddrs > 1) {

  48.         /*
  49.          * There are several addresses on this port and one of them
  50.          * is the "*:port" wildcard so getsockname() is needed to determine
  51.          * the server address.
  52.          *
  53.          * AcceptEx() and recvmsg() already gave this address.
  54.          */

  56.         if (ngx_connection_local_sockaddr(c, NULL, 0) != NGX_OK) {
  57.             ngx_stream_close_connection(c);
  58.             return;
  59.         }

  61.         sa = c->local_sockaddr;

  63.         switch (sa->sa_family) {

  65. #if (NGX_HAVE_INET6)
  66.         case AF_INET6:
  67.             sin6 = (struct sockaddr_in6 *) sa;

  69.             addr6 = port->addrs;

  71.             /* the last address is "*" */

  73.             for (i = 0; i < port->naddrs - 1; i++) {
  74.                 if (ngx_memcmp(&addr6[i].addr6, &sin6->sin6_addr, 16) == 0) {
  75.                     break;
  76.                 }
  77.             }

  79.             addr_conf = &addr6[i].conf;

  81.             break;
  82. #endif

  84.         default: /* AF_INET */
  85.             sin = (struct sockaddr_in *) sa;

  87.             addr = port->addrs;

  89.             /* the last address is "*" */

  91.             for (i = 0; i < port->naddrs - 1; i++) {
  92.                 if (addr[i].addr == sin->sin_addr.s_addr) {
  93.                     break;
  94.                 }
  95.             }

  97.             addr_conf = &addr[i].conf;

  99.             break;
  100.         }

  102.     } else {
  103.         switch (c->local_sockaddr->sa_family) {

  105. #if (NGX_HAVE_INET6)
  106.         case AF_INET6:
  107.             addr6 = port->addrs;
  108.             addr_conf = &addr6[0].conf;
  109.             break;
  110. #endif

  112.         default: /* AF_INET */
  113.             addr = port->addrs;
  114.             addr_conf = &addr[0].conf;
  115.             break;
  116.         }
  117.     }

  119.     s = ngx_pcalloc(c->pool, sizeof(ngx_stream_session_t));
  120.     if (s == NULL) {
  121.         ngx_stream_close_connection(c);
  122.         return;
  123.     }

  125.     ctx = addr_conf->default_server->ctx;

  127.     s->signature = NGX_STREAM_MODULE;
  128.     s->main_conf = ctx->main_conf;
  129.     s->srv_conf = ctx->srv_conf;
  130.     s->virtual_names = addr_conf->virtual_names;

  132. #if (NGX_STREAM_SSL)
  133.     s->ssl = addr_conf->ssl;
  134. #endif

  136.     if (c->buffer) {
  137.         s->received += c->buffer->last - c->buffer->pos;
  138.     }

  140.     s->connection = c;
  141.     c->data = s;

  143.     cscf = ngx_stream_get_module_srv_conf(s, ngx_stream_core_module);

  145.     ngx_set_connection_log(c, cscf->error_log);

  147.     len = ngx_sock_ntop(c->sockaddr, c->socklen, text, NGX_SOCKADDR_STRLEN, 1);

  149.     ngx_log_error(NGX_LOG_INFO, c->log, 0, "*%uA %sclient %*s connected to %V",
  150.                   c->number, c->type == SOCK_DGRAM ? "udp " : "",
  151.                   len, text, &c->listening->addr_text);

  153.     c->log->connection = c->number;
  154.     c->log->handler = ngx_stream_log_error;
  155.     c->log->data = s;
  156.     c->log->action = "initializing session";
  157.     c->log_error = NGX_ERROR_INFO;

  159.     s->ctx = ngx_pcalloc(c->pool, sizeof(void *) * ngx_stream_max_module);
  160.     if (s->ctx == NULL) {
  161.         ngx_stream_close_connection(c);
  162.         return;
  163.     }

  165.     cmcf = ngx_stream_get_module_main_conf(s, ngx_stream_core_module);

  167.     s->variables = ngx_pcalloc(s->connection->pool,
  168.                                cmcf->variables.nelts
  169.                                * sizeof(ngx_stream_variable_value_t));

  171.     if (s->variables == NULL) {
  172.         ngx_stream_close_connection(c);
  173.         return;
  174.     }

  176.     tp = ngx_timeofday();
  177.     s->start_sec = tp->sec;
  178.     s->start_msec = tp->msec;

  180.     rev = c->read;
  181.     rev->handler = ngx_stream_session_handler;

  183.     if (addr_conf->proxy_protocol) {
  184.         c->log->action = "reading PROXY protocol";

  186.         rev->handler = ngx_stream_proxy_protocol_handler;

  188.         if (!rev->ready) {
  189.             ngx_add_timer(rev, cscf->proxy_protocol_timeout);

  191.             if (ngx_handle_read_event(rev, 0) != NGX_OK) {
  192.                 ngx_stream_finalize_session(s,
  193.                                             NGX_STREAM_INTERNAL_SERVER_ERROR);
  194.             }

  196.             return;
  197.         }
  198.     }

  200.     if (ngx_use_accept_mutex) {
  201.         ngx_post_event(rev, &ngx_posted_events);
  202.         return;
  203.     }

  205.     rev->handler(rev);
  206. }


  209. static void
  210. ‌ngx_stream_proxy_protocol_handler(ngx_event_t *rev)
  211. {
  212.     u_char                      *p, buf[NGX_PROXY_PROTOCOL_MAX_HEADER];
  213.     size_t                       size;
  214.     ssize_t                      n;
  215.     ngx_err_t                    err;
  216.     ngx_connection_t            *c;
  217.     ngx_stream_session_t        *s;
  218.     ngx_stream_core_srv_conf_t  *cscf;

  220.     c = rev->data;
  221.     s = c->data;

  223.     ngx_log_debug0(NGX_LOG_DEBUG_STREAM, c->log, 0,
  224.                    "stream PROXY protocol handler");

  226.     if (rev->timedout) {
  227.         ngx_log_error(NGX_LOG_INFO, c->log, NGX_ETIMEDOUT, "client timed out");
  228.         ngx_stream_finalize_session(s, NGX_STREAM_OK);
  229.         return;
  230.     }

  232.     n = recv(c->fd, (char *) buf, sizeof(buf), MSG_PEEK);

  234.     err = ngx_socket_errno;

  236.     ngx_log_debug1(NGX_LOG_DEBUG_STREAM, c->log, 0, "recv(): %z", n);

  238.     if (n == -1) {
  239.         if (err == NGX_EAGAIN) {
  240.             rev->ready = 0;

  242.             if (!rev->timer_set) {
  243.                 cscf = ngx_stream_get_module_srv_conf(s,
  244.                                                       ngx_stream_core_module);

  246.                 ngx_add_timer(rev, cscf->proxy_protocol_timeout);
  247.             }

  249.             if (ngx_handle_read_event(rev, 0) != NGX_OK) {
  250.                 ngx_stream_finalize_session(s,
  251.                                             NGX_STREAM_INTERNAL_SERVER_ERROR);
  252.             }

  254.             return;
  255.         }

  257.         ngx_connection_error(c, err, "recv() failed");

  259.         ngx_stream_finalize_session(s, NGX_STREAM_OK);
  260.         return;
  261.     }

  263.     if (rev->timer_set) {
  264.         ngx_del_timer(rev);
  265.     }

  267.     p = ngx_proxy_protocol_read(c, buf, buf + n);

  269.     if (p == NULL) {
  270.         ngx_stream_finalize_session(s, NGX_STREAM_BAD_REQUEST);
  271.         return;
  272.     }

  274.     size = p - buf;

  276.     if (c->recv(c, buf, size) != (ssize_t) size) {
  277.         ngx_stream_finalize_session(s, NGX_STREAM_INTERNAL_SERVER_ERROR);
  278.         return;
  279.     }

  281.     c->log->action = "initializing session";

  283.     ngx_stream_session_handler(rev);
  284. }


  287. void
  288. ‌ngx_stream_session_handler(ngx_event_t *rev)
  289. {
  290.     ngx_connection_t      *c;
  291.     ngx_stream_session_t  *s;

  293.     c = rev->data;
  294.     s = c->data;

  296.     ngx_stream_core_run_phases(s);
  297. }


  300. void
  301. ‌ngx_stream_finalize_session(ngx_stream_session_t *s, ngx_uint_t rc)
  302. {
  303.     ngx_log_debug1(NGX_LOG_DEBUG_STREAM, s->connection->log, 0,
  304.                    "finalize stream session: %i", rc);

  306.     s->status = rc;

  308.     ngx_stream_log_session(s);

  310.     ngx_stream_close_connection(s->connection);
  311. }


  314. static void
  315. ‌ngx_stream_log_session(ngx_stream_session_t *s)
  316. {
  317.     ngx_uint_t                    i, n;
  318.     ngx_stream_handler_pt        *log_handler;
  319.     ngx_stream_core_main_conf_t  *cmcf;

  321.     cmcf = ngx_stream_get_module_main_conf(s, ngx_stream_core_module);

  323.     log_handler = cmcf->phases[NGX_STREAM_LOG_PHASE].handlers.elts;
  324.     n = cmcf->phases[NGX_STREAM_LOG_PHASE].handlers.nelts;

  326.     for (i = 0; i < n; i++) {
  327.         log_handler[i](s);
  328.     }
  329. }


  332. static void
  333. ‌ngx_stream_close_connection(ngx_connection_t *c)
  334. {
  335.     ngx_pool_t  *pool;

  337.     ngx_log_debug1(NGX_LOG_DEBUG_STREAM, c->log, 0,
  338.                    "close stream connection: %d", c->fd);

  340. #if (NGX_STREAM_SSL)

  342.     if (c->ssl) {
  343.         if (ngx_ssl_shutdown(c) == NGX_AGAIN) {
  344.             c->ssl->handler = ngx_stream_close_connection;
  345.             return;
  346.         }
  347.     }

  349. #endif

  351. #if (NGX_STAT_STUB)
  352.     (void) ngx_atomic_fetch_add(ngx_stat_active, -1);
  353. #endif

  355.     pool = c->pool;

  357.     ngx_close_connection(c);

  359.     ngx_destroy_pool(pool);
  360. }


  363. static u_char *
  364. ‌ngx_stream_log_error(ngx_log_t *log, u_char *buf, size_t len)
  365. {
  366.     u_char                *p;
  367.     ngx_stream_session_t  *s;

  369.     if (log->action) {
  370.         p = ngx_snprintf(buf, len, " while %s", log->action);
  371.         len -= p - buf;
  372.         buf = p;
  373.     }

  375.     s = log->data;

  377.     p = ngx_snprintf(buf, len, ", %sclient: %V, server: %V",
  378.                      s->connection->type == SOCK_DGRAM ? "udp " : "",
  379.                      &s->connection->addr_text,
  380.                      &s->connection->listening->addr_text);
  381.     len -= p - buf;
  382.     buf = p;

  384.     if (s->log_handler) {
  385.         p = s->log_handler(log, buf, len);
  386.     }

  388.     return p;
  389. }

One Level Up Top Level