text: 
  /*
   * Copyright (C) Roman Arutyunyan
   */
  
  
  #include <ngx_config.h>
  #include <ngx_core.h>
  #include "ngx_rtmp.h"
  #include "ngx_rtmp_cmd_module.h"
  #include "ngx_live_relay.h"
  #include "ngx_rtmp_proxy_protocol.h"
  #include "ngx_http_client.h"
  #include "ngx_rbuf.h"
  #include "ngx_poold.h"
  
  
  static void ngx_rtmp_close_connection(ngx_connection_t *c);
  static u_char * ngx_rtmp_log_error(ngx_log_t *log, u_char *buf, size_t len);
  
  
  typedef struct {
      ngx_str_t              *client;
      ngx_rtmp_session_t     *session;
      void                   *data;       // combined log
      ngx_log_handler_pt      handler;    // combined log handler
  } ngx_rtmp_error_log_ctx_t;
  
  
  char *ngx_live_stage[] = {
      "init",
      "handshake_done",
      "connect",
      "create_stream",
      "publish",
      "play",
      "audio_video",
      "close_stream",
  };
  
  
  char *ngx_live_err[] = {
      "internal_err",
      "normal_close",
      "rtmp_send_err",
      "rtmp_send_timeout",
      "flv_send_err",
      "flv_send_timeout",
      "rtmp_recv_err",
      "flv_recv_err",
      "relay_transit",
      "relay_timeout",
      "control_drop",
      "drop_idle",
      "oclp_notify_err",
      "oclp_relay_err",
      "oclp_para_err",
      "relay_close",
  };
  
  
  void
  ngx_rtmp_init_connection(ngx_connection_t *c)
  {
      ngx_uint_t             i;
      ngx_rtmp_port_t       *port;
      struct sockaddr       *sa;
      struct sockaddr_in    *sin;
      ngx_rtmp_in_addr_t    *addr;
      ngx_rtmp_session_t    *s;
      ngx_rtmp_addr_conf_t  *addr_conf;
      ngx_int_t              unix_socket;
  #if (NGX_HAVE_INET6)
      struct sockaddr_in6   *sin6;
      ngx_rtmp_in6_addr_t   *addr6;
  #endif
  
      ++ngx_rtmp_naccepted;
  
      /* find the server configuration for the address:port */
  
      /* AF_INET only */
  
      port = c->listening->servers;
      unix_socket = 0;
  
      if (port->naddrs > 1) {
  
          /*
           * There are several addresses on this port and one of them
           * is the "*:port" wildcard so getsockname() is needed to determine
           * the server address.
           *
           * AcceptEx() already gave this address.
           */
  
          if (ngx_connection_local_sockaddr(c, NULL, 0) != NGX_OK) {
              ngx_rtmp_close_connection(c);
              return;
          }
  
          sa = c->local_sockaddr;
  
          switch (sa->sa_family) {
  
  #if (NGX_HAVE_INET6)
          case AF_INET6:
              sin6 = (struct sockaddr_in6 *) sa;
  
              addr6 = port->addrs;
  
              /* the last address is "*" */
  
              for (i = 0; i < port->naddrs - 1; i++) {
                  if (ngx_memcmp(&addr6[i].addr6, &sin6->sin6_addr, 16) == 0) {
                      break;
                  }
              }
  
              addr_conf = &addr6[i].conf;
  
              break;
  #endif
  
          case AF_UNIX:
              unix_socket = 1;
  
              sin = (struct sockaddr_in *) sa;
  
              addr = port->addrs;
  
              /* the last address is "*" */
  
              for (i = 0; i < port->naddrs - 1; i++) {
                  if (addr[i].addr == sin->sin_addr.s_addr) {
                      break;
                  }
              }
  
              addr_conf = &addr[i].conf;
  
              break;
  
          default: /* AF_INET */
              sin = (struct sockaddr_in *) sa;
  
              addr = port->addrs;
  
              /* the last address is "*" */
  
              for (i = 0; i < port->naddrs - 1; i++) {
                  if (addr[i].addr == sin->sin_addr.s_addr) {
                      break;
                  }
              }
  
              addr_conf = &addr[i].conf;
  
              break;
          }
  
      } else {
          switch (c->local_sockaddr->sa_family) {
  
  #if (NGX_HAVE_INET6)
          case AF_INET6:
              addr6 = port->addrs;
              addr_conf = &addr6[0].conf;
              break;
  #endif
  
          case AF_UNIX:
              unix_socket = 1;
              addr = port->addrs;
              addr_conf = &addr[0].conf;
              break;
  
          default: /* AF_INET */
              addr = port->addrs;
              addr_conf = &addr[0].conf;
              break;
          }
      }
  
      ngx_log_error(NGX_LOG_INFO, c->log, 0, "*%ui client connected '%V'",
                    c->number, &c->addr_text);
  
      s = ngx_rtmp_create_session(addr_conf);
      if (s == NULL) {
          ngx_log_error(NGX_LOG_INFO, c->log, 0, "create rtmp session failed");
          return;
      }
      s->log->connection = c->number;
      s->number = c->number;
      s->remote_addr_text.data = ngx_pcalloc(s->pool, c->addr_text.len);
      s->remote_addr_text.len = c->addr_text.len;
      ngx_memcpy(s->remote_addr_text.data, c->addr_text.data, c->addr_text.len);
      s->sockaddr = ngx_pcalloc(s->pool, sizeof(struct sockaddr));
      ngx_memcpy(s->sockaddr, c->sockaddr, sizeof(struct sockaddr));
  
      ngx_rtmp_init_session(s, c);
  
      /* only auto-pushed connections are
       * done through unix socket */
  
      s->interprocess = unix_socket;
  
      if (addr_conf->proxy_protocol) {
          ngx_rtmp_proxy_protocol(s);
  
      } else {
          ngx_rtmp_handshake(s);
      }
  }
  
  
  static u_char *
  ngx_rtmp_log_error(ngx_log_t *log, u_char *buf, size_t len)
  {
      u_char                     *p;
      ngx_rtmp_session_t         *s;
      ngx_rtmp_error_log_ctx_t   *ctx;
  
      p = buf;
  
      ctx = log->data;
  
      if (ctx->client) {
          p = ngx_snprintf(buf, len, ", client: %V", ctx->client);
          len -= p - buf;
          buf = p;
      }
  
      s = ctx->session;
  
      if (s == NULL) {
          return p;
      }
  
      p = ngx_snprintf(buf, len, ", server: %V, session: %p", s->addr_text, s);
      len -= p - buf;
      buf = p;
  
      if (s->stream.len) {
          p = ngx_snprintf(buf, len, ", stream: %V", &s->stream);
          len -= p - buf;
          buf = p;
      }
  
      if (ctx->data) { // combined log
          log->data = ctx->data;
          p = ctx->handler(log, p, len);
          log->data = ctx;
      }
  
      return p;
  }
  
  
  static void
  ngx_rtmp_close_connection(ngx_connection_t *c)
  {
      ngx_pool_t                         *pool;
  
      if (c == NULL) {
          return;
      }
  
      ngx_log_debug0(NGX_LOG_DEBUG_RTMP, c->log, 0, "close connection");
  
  #if (NGX_STAT_STUB)
      (void) ngx_atomic_fetch_add(ngx_stat_active, -1);
  #endif
  
      pool = c->pool;
      ngx_close_connection(c);
      ngx_destroy_pool(pool);
  }
  
  
  static void
  ngx_rtmp_close_session(ngx_rtmp_session_t *s)
  {
      if (s->ping_evt.timer_set) {
          ngx_del_timer(&s->ping_evt);
      }
  
      if (s->in_old_pool) {
          NGX_DESTROY_POOL(s->in_old_pool);
      }
  
      if (s->in_pool) {
          NGX_DESTROY_POOL(s->in_pool);
      }
  
      if (s->live_type == NGX_HTTP_FLV_LIVE) {
          ngx_put_chainbufs(s->in_streams[0].in);
      }
  
      ngx_rtmp_free_handshake_buffers(s);
  
      ngx_rtmp_free_merge_frame(s);
  
      if (s->live_type == NGX_MPEGTS_LIVE) {
          while (s->out_pos != s->out_last) {
              ngx_rtmp_shared_free_mpegts_frame(s->mpegts_out[s->out_pos++]);
              s->out_pos %= s->out_queue;
          }
      } else if (s->live_type == NGX_RTMP_LIVE ||
              s->live_type == NGX_HTTP_FLV_LIVE)
      {
          while (s->out_pos != s->out_last) {
              ngx_rtmp_shared_free_frame(s->out[s->out_pos++]);
              s->out_pos %= s->out_queue;
          }
      }
  
      NGX_DESTROY_POOL(s->pool);
  }
  
  
  static void
  ngx_rtmp_close_session_handler(ngx_event_t *e)
  {
      ngx_rtmp_session_t                 *s;
      ngx_connection_t                   *c;
  
      s = e->data;
      c = s->connection;
      if (c) {
          c->destroyed = 1;
      }
  
      ngx_log_error(NGX_LOG_INFO, s->log, 0, "async close session");
  
      ngx_rtmp_fire_event(s, NGX_RTMP_DISCONNECT, NULL, NULL);
  
      // close connection before session, connection log use session log
      // may cause memory error
      ngx_rtmp_close_connection(c);
  
      ngx_rtmp_close_session(s);
  }
  
  
  static void
  ngx_rtmp_async_finalize_http_client(ngx_event_t *ev)
  {
      ngx_rtmp_session_t         *s;
      ngx_http_request_t         *hcr;
  
      s = ev->data;
      hcr = s->request;
  
      ngx_log_error(NGX_LOG_INFO, s->log, 0, "asyn finalize http client");
  
      if (hcr) {
          ngx_http_client_finalize_request(hcr, 1);
      }
  }
  
  
  static void
  ngx_rtmp_async_finalize_http_request(ngx_event_t *ev)
  {
      ngx_rtmp_session_t         *s;
      ngx_http_request_t         *r;
  
      s = ev->data;
      r = s->request;
  
      ngx_log_error(NGX_LOG_INFO, s->log, 0, "asyn finalize http request");
  
      if (r->header_sent) {
          ngx_http_finalize_request(r, NGX_HTTP_CLIENT_CLOSED_REQUEST);
          ngx_http_run_posted_requests(r->connection);
      } else {
          r->error_page = 1;
  
          if (s->status) {
              ngx_http_finalize_request(r, s->status);
          } else {
              ngx_http_finalize_request(r, NGX_HTTP_SERVICE_UNAVAILABLE);
          }
      }
  }
  
  
  void
  ngx_rtmp_finalize_session(ngx_rtmp_session_t *s)
  {
      ngx_event_t        *e;
      ngx_connection_t   *c;
  
      if (s->live_type == NGX_HLS_LIVE) {
          ngx_rtmp_finalize_fake_session(s);
          return;
      }
  
      if (s->destroyed) {
          return;
      }
  
      s->destroyed = 1;
  
      c = s->connection;
      if (c && c->destroyed) {
          return;
      }
  
      ngx_log_error(NGX_LOG_INFO, s->log, 0, "finalize session");
  
      if (s->live_type != NGX_RTMP_LIVE) {
          e = &s->close;
          e->data = s;
          if (s->relay) {
              e->handler = ngx_rtmp_async_finalize_http_client;
          } else {
              e->handler = ngx_rtmp_async_finalize_http_request;
          }
          e->log = s->log;
  
          ngx_post_event(e, &ngx_posted_events);
  
          return;
      }
  
      e = &s->close;
      e->data = s;
      e->handler = ngx_rtmp_close_session_handler;
      e->log = s->log;
  
      ngx_post_event(e, &ngx_posted_events);
  }
  
  
  // session combined with http request or http client request
  void
  ngx_rtmp_finalize_fake_session(ngx_rtmp_session_t *s)
  {
      ngx_log_error(NGX_LOG_INFO, s->log, 0, "finalize fake session");
  
      if (s->destroyed) {
          return;
      }
  
      s->destroyed = 1;
  
      ngx_rtmp_fire_event(s, NGX_RTMP_DISCONNECT, NULL, NULL);
  
      ngx_rtmp_close_session(s);
  }
  
  
  void
  ngx_rtmp_set_combined_log(ngx_rtmp_session_t *s, void *d, ngx_log_handler_pt h)
  {
      ngx_rtmp_error_log_ctx_t       *ctx;
  
      ctx = s->log->data;
      ctx->data = d;
      ctx->handler = h;
  }
  
  
  // Only rtmp need to init session
  void
  ngx_rtmp_init_session(ngx_rtmp_session_t *s, ngx_connection_t *c)
  {
      ngx_rtmp_error_log_ctx_t       *ctx;
  
      c->data = s;
      s->connection = c;
  
      c->log = s->log;
      c->write->log = s->log;
      c->read->log = s->log;
  
      ctx = s->log->data;
      ctx->client = &c->addr_text;
  
      s->log->connection = c->number;
  
      c->log_error = NGX_ERROR_INFO;
  
      ngx_rtmp_set_chunk_size(s, NGX_RTMP_DEFAULT_CHUNK_SIZE);
  
      return;
  }
  
  
  ngx_rtmp_session_t *
  ngx_rtmp_create_relay_session(ngx_rtmp_session_t *s, void *tag)
  {
      ngx_rtmp_session_t         *rs;
      ngx_live_relay_ctx_t       *rctx;
  
      rs = ngx_rtmp_create_session(s->addr_conf);
      if (rs == NULL) {
          return NULL;
      }
  
      rs->relay = 1;
      rs->main_conf = s->main_conf;
      rs->srv_conf = s->srv_conf;
      rs->app_conf = s->app_conf;
  
      // clone para from origin session
  #define NGX_RTMP_RELAY_SESSION_COPY_PARA(para)                      \
      if (ngx_copy_str(rs->pool, &rs->para, &s->para) != NGX_OK) {    \
          goto destroy;                                               \
      }
  
      NGX_RTMP_RELAY_SESSION_COPY_PARA(stream);
  
      NGX_RTMP_RELAY_SESSION_COPY_PARA(name);
      NGX_RTMP_RELAY_SESSION_COPY_PARA(pargs);
  
      NGX_RTMP_RELAY_SESSION_COPY_PARA(app);
      NGX_RTMP_RELAY_SESSION_COPY_PARA(args);
      NGX_RTMP_RELAY_SESSION_COPY_PARA(flashver);
      NGX_RTMP_RELAY_SESSION_COPY_PARA(swf_url);
      NGX_RTMP_RELAY_SESSION_COPY_PARA(tc_url);
      NGX_RTMP_RELAY_SESSION_COPY_PARA(page_url);
  
      NGX_RTMP_RELAY_SESSION_COPY_PARA(serverid);
  #undef NGX_RTMP_RELAY_SESSION_COPY_PARA
  
      rs->acodecs = s->acodecs;
      rs->vcodecs = s->vcodecs;
  
      ngx_rtmp_cmd_middleware_init(rs);
  
      // create relay ctx
      rctx = ngx_pcalloc(rs->pool, sizeof(ngx_live_relay_ctx_t));
      if (rctx == NULL) {
          goto destroy;
      }
      rctx->domain = rs->domain;
      rctx->app = rs->app;
      rctx->args = rs->args;
      rctx->name = rs->name;
      rctx->pargs = rs->pargs;
  
      rctx->referer = rs->page_url;
      rctx->user_agent = rs->flashver;
      rctx->swf_url = rs->swf_url;
      rctx->acodecs = rs->acodecs;
      rctx->vcodecs = rs->vcodecs;
  
      rctx->tag = tag;
  
      ngx_rtmp_set_ctx(rs, rctx, ngx_live_relay_module);
  
      return rs;
  
  destroy:
      ngx_rtmp_close_session(rs);
  
      return NULL;
  }
  
  
  ngx_rtmp_session_t *
  ngx_rtmp_create_static_session(ngx_live_relay_t *relay,
          ngx_rtmp_addr_conf_t *addr_conf, void *tag)
  {
      ngx_rtmp_session_t         *rs;
      ngx_live_relay_ctx_t       *rctx;
      ngx_uint_t                  n;
      ngx_rtmp_core_srv_conf_t   *cscf;
      ngx_rtmp_core_app_conf_t  **cacfp;
  
      rs = ngx_rtmp_create_session(addr_conf);
      if (rs == NULL) {
          return NULL;
      }
  
      rs->static_pull = 1;
      rs->relay = 1;
  
      // clone para from origin session
  #define NGX_RTMP_RELAY_SESSION_COPY_PARA(to, from)                  \
      if (ngx_copy_str(rs->pool, &rs->to, &relay->from) != NGX_OK) {  \
          goto destroy;                                               \
      }
  
      NGX_RTMP_RELAY_SESSION_COPY_PARA(app, app);
      NGX_RTMP_RELAY_SESSION_COPY_PARA(name, name);
      NGX_RTMP_RELAY_SESSION_COPY_PARA(pargs, pargs);
      NGX_RTMP_RELAY_SESSION_COPY_PARA(page_url, referer);
      NGX_RTMP_RELAY_SESSION_COPY_PARA(flashver, user_agent);
  
      NGX_RTMP_RELAY_SESSION_COPY_PARA(stream, stream);
  #undef NGX_RTMP_RELAY_SESSION_COPY_PARA
  
      rs->tc_url.len = sizeof("rtmp://") - 1 + relay->domain.len
                     + sizeof("/") - 1 + relay->app.len;
      rs->tc_url.data = ngx_pcalloc(rs->pool, rs->tc_url.len);
      if (rs->tc_url.data == NULL) {
          goto destroy;
      }
      ngx_snprintf(rs->tc_url.data, rs->tc_url.len, "rtmp://%V/%V",
              &relay->domain, &relay->app);
  
      ngx_rtmp_cmd_middleware_init(rs);
  
      if (ngx_rtmp_set_virtual_server(rs, &rs->domain)) {
          goto destroy;
      }
      cscf = ngx_rtmp_get_module_srv_conf(rs, ngx_rtmp_core_module);
  
      rs->live_server = ngx_live_create_server(&rs->serverid);
  
      cacfp = cscf->applications.elts;
      for (n = 0; n < cscf->applications.nelts; ++n, ++cacfp) {
          if ((*cacfp)->name.len == rs->app.len &&
              ngx_strncmp((*cacfp)->name.data, rs->app.data, rs->app.len) == 0)
          {
              /* found app! */
              rs->app_conf = (*cacfp)->app_conf;
              break;
          }
      }
  
      if (rs->app_conf == NULL) {
  
          if (cscf->default_app == NULL || cscf->default_app->app_conf == NULL) {
              ngx_log_error(NGX_LOG_ERR, rs->log, 0,
                      "static session: application not found: '%V'", &rs->app);
              goto destroy;
          }
  
          rs->app_conf = cscf->default_app->app_conf;
      }
  
      // create relay ctx
      rctx = ngx_pcalloc(rs->pool, sizeof(ngx_live_relay_ctx_t));
      if (rctx == NULL) {
          goto destroy;
      }
      rctx->domain = rs->domain;
      rctx->app = rs->app;
      rctx->args = rs->args;
      rctx->name = rs->name;
      rctx->pargs = rs->pargs;
  
      rctx->referer = rs->page_url;
      rctx->user_agent = rs->flashver;
      rctx->swf_url = rs->swf_url;
      rctx->acodecs = rs->acodecs;
      rctx->vcodecs = rs->vcodecs;
  
      rctx->tag = tag;
  
      ngx_rtmp_set_ctx(rs, rctx, ngx_live_relay_module);
  
      return rs;
  
  destroy:
      ngx_rtmp_close_session(rs);
  
      return NULL;
  }
  
  
  // if return NULL, memory must be error
  ngx_rtmp_session_t *
  ngx_rtmp_create_session(ngx_rtmp_addr_conf_t *addr_conf)
  {
      ngx_rtmp_session_t             *s;
      ngx_rtmp_core_srv_conf_t       *cscf;
      ngx_pool_t                     *pool;
      ngx_rtmp_error_log_ctx_t       *ctx;
  
      pool = NGX_CREATE_POOL(4096, ngx_cycle->log);
      if (pool == NULL) {
          return NULL;
      }
  
      s = ngx_pcalloc(pool, sizeof(ngx_rtmp_session_t) +
              sizeof(ngx_rtmp_frame_t *) * ((ngx_rtmp_core_srv_conf_t *)
                  addr_conf->default_server->ctx-> srv_conf[ngx_rtmp_core_module
                      .ctx_index])->out_queue);
      if (s == NULL) {
          goto destroy;
      }
      s->pool = pool;
  
      s->addr_conf = addr_conf;
  
      s->main_conf = addr_conf->default_server->ctx->main_conf;
      s->srv_conf = addr_conf->default_server->ctx->srv_conf;
  
      s->addr_text = ngx_pcalloc(s->pool, sizeof(ngx_str_t));
      s->addr_text->data = ngx_pcalloc(s->pool, addr_conf->addr_text.len);
      s->addr_text->len = addr_conf->addr_text.len;
      ngx_memcpy(s->addr_text->data,
          addr_conf->addr_text.data, addr_conf->addr_text.len);
  
      s->ctx = ngx_pcalloc(pool, sizeof(void *) * ngx_rtmp_max_module);
      if (s->ctx == NULL) {
          goto destroy;
      }
  
      cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module);
  
      s->out_queue = cscf->out_queue;
      s->out_cork = cscf->out_cork;
      s->in_streams = ngx_pcalloc(pool, sizeof(ngx_rtmp_stream_t)
              * cscf->max_streams);
      if (s->in_streams == NULL) {
          goto destroy;
      }
  
  #if (nginx_version >= 1007005)
      ngx_queue_init(&s->posted_dry_events);
  #endif
  
      s->epoch = ngx_current_msec;
      s->timeout = cscf->timeout;
      s->buflen = cscf->buflen;
  
      // init log
      ctx = ngx_pcalloc(pool, sizeof(ngx_rtmp_error_log_ctx_t));
      if (ctx == NULL) {
          goto destroy;
      }
  
      ctx->session = s;
  
      s->log = ngx_pcalloc(pool, sizeof(ngx_log_t));
      if (s->pool == NULL) {
          goto destroy;
      }
      *s->log = ngx_cycle->new_log;
  
      s->log->handler = ngx_rtmp_log_error;
      s->log->data = ctx;
  
      s->stage = NGX_LIVE_INIT;
      s->init_time = ngx_current_msec;
  
      s->mpegts_out = ngx_pcalloc(s->pool, sizeof(ngx_mpegts_frame_t *) *
                  ((ngx_rtmp_core_srv_conf_t *) addr_conf->default_server->ctx->
                  srv_conf[ngx_rtmp_core_module.ctx_index])->out_queue);
  
      return s;
  
  destroy:
      if (pool) {
          NGX_DESTROY_POOL(pool);
      }
  
      return NULL;
  }
  