Mercurial > hg > nginx

comparison src/stream/ngx_stream_proxy_module.c @ 7392:04ff25798002

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Stream: session completion check code moved to a separate function. The code refactored to simplify the ngx_stream_proxy_process() function and facilitate adding new session termination conditions.
author Vladimir Homutov <vl@nginx.com>
date Mon, 12 Nov 2018 12:05:03 +0300
parents 8b68d50090e4
children 4698cede59ff
comparison
equal deleted inserted replaced
7391:27559d4a5151 7392:04ff25798002
71 ngx_uint_t from_upstream); 71 ngx_uint_t from_upstream);
72 static void ngx_stream_proxy_connect_handler(ngx_event_t *ev); 72 static void ngx_stream_proxy_connect_handler(ngx_event_t *ev);
73 static ngx_int_t ngx_stream_proxy_test_connect(ngx_connection_t *c); 73 static ngx_int_t ngx_stream_proxy_test_connect(ngx_connection_t *c);
74 static void ngx_stream_proxy_process(ngx_stream_session_t *s, 74 static void ngx_stream_proxy_process(ngx_stream_session_t *s,
75 ngx_uint_t from_upstream, ngx_uint_t do_write); 75 ngx_uint_t from_upstream, ngx_uint_t do_write);
76 static ngx_int_t ngx_stream_proxy_test_finalize(ngx_stream_session_t *s,
77 ngx_uint_t from_upstream);
76 static void ngx_stream_proxy_next_upstream(ngx_stream_session_t *s); 78 static void ngx_stream_proxy_next_upstream(ngx_stream_session_t *s);
77 static void ngx_stream_proxy_finalize(ngx_stream_session_t *s, ngx_uint_t rc); 79 static void ngx_stream_proxy_finalize(ngx_stream_session_t *s, ngx_uint_t rc);
78 static u_char *ngx_stream_proxy_log_error(ngx_log_t *log, u_char *buf, 80 static u_char *ngx_stream_proxy_log_error(ngx_log_t *log, u_char *buf,
79 size_t len); 81 size_t len);
80 82
1644 break; 1646 break;
1645 } 1647 }
1646 1648
1647 c->log->action = "proxying connection"; 1649 c->log->action = "proxying connection";
1648 1650
1649 if (c->type == SOCK_DGRAM 1651 if (ngx_stream_proxy_test_finalize(s, from_upstream) == NGX_OK) {
1650 && pscf->responses != NGX_MAX_INT32_VALUE 1652 return;
1651 && u->responses >= pscf->responses * u->requests 1653 }
1652 && !src->buffered && dst && !dst->buffered) 1654
1653 { 1655 flags = src->read->eof ? NGX_CLOSE_EVENT : 0;
1656
1657 if (!src->shared && ngx_handle_read_event(src->read, flags) != NGX_OK) {
1658 ngx_stream_proxy_finalize(s, NGX_STREAM_INTERNAL_SERVER_ERROR);
1659 return;
1660 }
1661
1662 if (dst) {
1663 if (!dst->shared && ngx_handle_write_event(dst->write, 0) != NGX_OK) {
1664 ngx_stream_proxy_finalize(s, NGX_STREAM_INTERNAL_SERVER_ERROR);
1665 return;
1666 }
1667
1668 if (!c->read->delayed && !pc->read->delayed) {
1669 ngx_add_timer(c->write, pscf->timeout);
1670
1671 } else if (c->write->timer_set) {
1672 ngx_del_timer(c->write);
1673 }
1674 }
1675 }
1676
1677
1678 static ngx_int_t
1679 ngx_stream_proxy_test_finalize(ngx_stream_session_t *s,
1680 ngx_uint_t from_upstream)
1681 {
1682 ngx_connection_t *c, *pc;
1683 ngx_log_handler_pt handler;
1684 ngx_stream_upstream_t *u;
1685 ngx_stream_proxy_srv_conf_t *pscf;
1686
1687 pscf = ngx_stream_get_module_srv_conf(s, ngx_stream_proxy_module);
1688
1689 c = s->connection;
1690 u = s->upstream;
1691 pc = u->connected ? u->peer.connection : NULL;
1692
1693 if (c->type == SOCK_DGRAM) {
1694
1695 if (pscf->responses == NGX_MAX_INT32_VALUE
1696 || u->responses < pscf->responses * u->requests)
1697 {
1698 return NGX_DECLINED;
1699 }
1700
1701 if (pc == NULL || c->buffered || pc->buffered) {
1702 return NGX_DECLINED;
1703 }
1704
1654 handler = c->log->handler; 1705 handler = c->log->handler;
1655 c->log->handler = NULL; 1706 c->log->handler = NULL;
1656 1707
1657 ngx_log_error(NGX_LOG_INFO, c->log, 0, 1708 ngx_log_error(NGX_LOG_INFO, c->log, 0,
1658 "udp done" 1709 "udp done"
1663 s->received, c->sent, u->received, pc ? pc->sent : 0); 1714 s->received, c->sent, u->received, pc ? pc->sent : 0);
1664 1715
1665 c->log->handler = handler; 1716 c->log->handler = handler;
1666 1717
1667 ngx_stream_proxy_finalize(s, NGX_STREAM_OK); 1718 ngx_stream_proxy_finalize(s, NGX_STREAM_OK);
1668 return; 1719
1669 } 1720 return NGX_OK;
1670 1721 }
1671 if (c->type == SOCK_STREAM 1722
1672 && src->read->eof && dst && (dst->read->eof || !dst->buffered)) 1723 /* c->type == SOCK_STREAM */
1724
1725 if (pc == NULL
1726 || (!c->read->eof && !pc->read->eof)
1727 || (!c->read->eof && c->buffered)
1728 || (!pc->read->eof && pc->buffered))
1673 { 1729 {
1674 handler = c->log->handler; 1730 return NGX_DECLINED;
1675 c->log->handler = NULL; 1731 }
1676 1732
1677 ngx_log_error(NGX_LOG_INFO, c->log, 0, 1733 handler = c->log->handler;
1678 "%s disconnected" 1734 c->log->handler = NULL;
1679 ", bytes from/to client:%O/%O" 1735
1680 ", bytes from/to upstream:%O/%O", 1736 ngx_log_error(NGX_LOG_INFO, c->log, 0,
1681 from_upstream ? "upstream" : "client", 1737 "%s disconnected"
1682 s->received, c->sent, u->received, pc ? pc->sent : 0); 1738 ", bytes from/to client:%O/%O"
1683 1739 ", bytes from/to upstream:%O/%O",
1684 c->log->handler = handler; 1740 from_upstream ? "upstream" : "client",
1685 1741 s->received, c->sent, u->received, pc ? pc->sent : 0);
1686 ngx_stream_proxy_finalize(s, NGX_STREAM_OK); 1742
1687 return; 1743 c->log->handler = handler;
1688 } 1744
1689 1745 ngx_stream_proxy_finalize(s, NGX_STREAM_OK);
1690 flags = src->read->eof ? NGX_CLOSE_EVENT : 0; 1746
1691 1747 return NGX_OK;
1692 if (!src->shared && ngx_handle_read_event(src->read, flags) != NGX_OK) {
1693 ngx_stream_proxy_finalize(s, NGX_STREAM_INTERNAL_SERVER_ERROR);
1694 return;
1695 }
1696
1697 if (dst) {
1698 if (!dst->shared && ngx_handle_write_event(dst->write, 0) != NGX_OK) {
1699 ngx_stream_proxy_finalize(s, NGX_STREAM_INTERNAL_SERVER_ERROR);
1700 return;
1701 }
1702
1703 if (!c->read->delayed && !pc->read->delayed) {
1704 ngx_add_timer(c->write, pscf->timeout);
1705
1706 } else if (c->write->timer_set) {
1707 ngx_del_timer(c->write);
1708 }
1709 }
1710 } 1748 }
1711 1749
1712 1750
1713 static void 1751 static void
1714 ngx_stream_proxy_next_upstream(ngx_stream_session_t *s) 1752 ngx_stream_proxy_next_upstream(ngx_stream_session_t *s)