push is working

ngx_rtmp_access_module.c

@@ -230,6 +230,11 @@ ngx_rtmp_access(ngx_rtmp_session_t *s, ngx_uint_t flag)
         return NGX_ERROR;
     }
 
+    /* relay etc */
+    if (s->connection->sockaddr == NULL) {
+        return NGX_OK;
+    }
+
     switch (s->connection->sockaddr->sa_family) {
 
     case AF_INET:

ngx_rtmp_cmd_module.c

@@ -205,7 +205,7 @@ ngx_rtmp_cmd_connect(ngx_rtmp_session_t *s, ngx_rtmp_connect_t *v)
 
 
 #define NGX_RTMP_SET_STRPAR(name)                                             \
-    s->name.len = ngx_strlen(v->name);                                         \
+    s->name.len = ngx_strlen(v->name);                                        \
     s->name.data = ngx_palloc(s->connection->pool, s->name.len);              \
     ngx_memcpy(s->name.data, v->name, s->name.len)
 

ngx_rtmp_handshake.c

@@ -172,6 +172,10 @@ ngx_rtmp_alloc_handshake_buffer(ngx_rtmp_session_t *s, int short_buf)
     ngx_chain_t                *cl;
     ngx_buf_t                  *b;
 
+    ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
+            "handshake: allocating %sbuffer",
+            short_buf ? "short " : "");
+
     cscf = ngx_rtmp_get_module_srv_conf(s, ngx_rtmp_core_module);
 
     if (cscf->free_hs) {
@@ -243,7 +247,7 @@ ngx_rtmp_old_handshake_response(ngx_rtmp_session_t *s)
     size_t                  len;
 
     ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0, 
-            "RTMP old-style handshake");
+            "handshake: old-style handshake");
 
     src = s->hs_bufs[0]->pos + 8;
     len = s->hs_bufs[0]->last - src;
@@ -276,7 +280,7 @@ ngx_rtmp_handshake_response(ngx_rtmp_session_t *s)
     b = s->hs_bufs[0];
     if (*b->pos != '\x03') {
         ngx_log_error(NGX_LOG_INFO, s->connection->log, 0, 
-                "Unexpected RTMP version: %i", (ngx_int_t)*b->pos);
+                "handshake: unexpected RTMP version: %i", (ngx_int_t)*b->pos);
         return NGX_ERROR;
     }
     ++b->pos;
@@ -284,7 +288,7 @@ ngx_rtmp_handshake_response(ngx_rtmp_session_t *s)
 
     p = b->pos + 4;
     ngx_log_debug5(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
-            "RTMP client version=%i.%i.%i.%i epoch=%uD",
+            "handshake: client version=%i.%i.%i.%i epoch=%uD",
             (ngx_int_t)p[3], (ngx_int_t)p[2],
             (ngx_int_t)p[1], (ngx_int_t)p[0],
             s->peer_epoch);
@@ -300,7 +304,7 @@ ngx_rtmp_handshake_response(ngx_rtmp_session_t *s)
     }
     if (offs == NGX_ERROR) {
         ngx_log_error(NGX_LOG_INFO, s->connection->log, 0, 
-                "RTMP digest not found");
+                "handshake: digest not found");
         return ngx_rtmp_old_handshake_response(s);
     }
     b->pos += offs;
@@ -311,7 +315,7 @@ ngx_rtmp_handshake_response(ngx_rtmp_session_t *s)
         return NGX_ERROR;
     }
     ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
-            "RTMP digest found at pos=%i", offs);
+            "handshake: digest found at pos=%i", offs);
 
     /* create first output buffer */
     b = s->hs_bufs[1];
@@ -364,10 +368,12 @@ ngx_rtmp_handshake_make_client_request(ngx_rtmp_session_t *s)
     b->last = ngx_rtmp_rcpymem(b->last, &s->epoch, 4);
     b->last = ngx_rtmp_rcpymem(b->last, ngx_rtmp_client_version, 4);
     ngx_rtmp_fill_random_buffer(b);
+    ++b->pos;
     if (ngx_rtmp_write_digest(b, &ngx_rtmp_client_partial_key,
                 0, s->connection->log) != NGX_OK) {
         return NGX_ERROR;
     }
+    --b->pos;
     
     return NGX_OK;
 }
@@ -379,7 +385,7 @@ ngx_rtmp_handshake_done(ngx_rtmp_session_t *s)
     ngx_rtmp_free_handshake_buffers(s);
 
     ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
-            "RTMP handshake done");
+            "handshake: done");
 
     if (ngx_rtmp_fire_event(s, NGX_RTMP_HANDSHAKE_DONE, 
                 NULL, NULL) != NGX_OK) 
@@ -408,7 +414,8 @@ ngx_rtmp_handshake_recv(ngx_event_t *rev)
     }
 
     if (rev->timedout) {
-        ngx_log_error(NGX_LOG_INFO, c->log, NGX_ETIMEDOUT, "client timed out");
+        ngx_log_error(NGX_LOG_INFO, c->log, NGX_ETIMEDOUT, 
+                "handshake: client timed out");
         c->timedout = 1;
         ngx_rtmp_finalize_session(s);
         return;
@@ -440,20 +447,24 @@ ngx_rtmp_handshake_recv(ngx_event_t *rev)
     }
 
     if (rev->active) {
-        ngx_del_event(c->read, NGX_READ_EVENT, 0);
+        ngx_del_event(rev, NGX_READ_EVENT, 0);
     }
 
-    switch (++s->hs_stage) {
+    ++s->hs_stage;
+    ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
+            "handshake: stage %ui", s->hs_stage);
+
+    switch (s->hs_stage) {
         case NGX_RTMP_HANDSHAKE_SERVER_SEND_CHALLENGE:
             s->hs_bufs[1] = ngx_rtmp_alloc_handshake_buffer(s, 0);
             s->hs_bufs[2] = ngx_rtmp_alloc_handshake_buffer(s, 1);
             if (ngx_rtmp_handshake_response(s) != NGX_OK) {
                 ngx_log_error(NGX_LOG_INFO, c->log, 0, 
-                        "RTMP handshake error");
+                        "handshake: response error");
                 ngx_rtmp_finalize_session(s);
                 return;
             }
-            s->hs_buf = s->hs_bufs[0];
+            s->hs_buf = s->hs_bufs[1];
             ngx_rtmp_handshake_send(c->write);
             break;
 
@@ -470,7 +481,7 @@ ngx_rtmp_handshake_recv(ngx_event_t *rev)
         case NGX_RTMP_HANDSHAKE_CLIENT_SEND_RESPONSE:
             if (ngx_rtmp_handshake_client_response(s) != NGX_OK) {
                 ngx_log_error(NGX_LOG_INFO, c->log, 0, 
-                        "RTMP client handshake error");
+                        "handshake: client response error");
                 ngx_rtmp_finalize_session(s);
                 return;
             }
@@ -498,7 +509,7 @@ ngx_rtmp_handshake_send(ngx_event_t *wev)
 
     if (wev->timedout) {
         ngx_log_error(NGX_LOG_INFO, c->log, NGX_ETIMEDOUT, 
-                "client timed out");
+                "handshake: client timed out");
         c->timedout = 1;
         ngx_rtmp_finalize_session(s);
         return;
@@ -529,7 +540,15 @@ ngx_rtmp_handshake_send(ngx_event_t *wev)
         b->pos += n;
     }
 
-    switch (++s->hs_stage) {
+    if (wev->active) {
+        ngx_del_event(wev, NGX_WRITE_EVENT, 0);
+    }
+
+    ++s->hs_stage;
+    ngx_log_debug1(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
+            "handshake: stage %ui", s->hs_stage);
+
+    switch (s->hs_stage) {
         case NGX_RTMP_HANDSHAKE_SERVER_SEND_RESPONSE:
             s->hs_buf = s->hs_bufs[2];
             ngx_rtmp_handshake_send(wev);
@@ -563,6 +582,9 @@ ngx_rtmp_handshake(ngx_rtmp_session_t *s)
     c->read->handler =  ngx_rtmp_handshake_recv;
     c->write->handler = ngx_rtmp_handshake_send;
 
+    ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
+            "handshake: start server handshake");
+
     s->hs_bufs[0] = ngx_rtmp_alloc_handshake_buffer(s, 0);
     s->hs_buf = s->hs_bufs[0];
     s->hs_stage = NGX_RTMP_HANDSHAKE_SERVER_RECV_CHALLENGE;
@@ -580,6 +602,9 @@ ngx_rtmp_client_handshake(ngx_rtmp_session_t *s)
     c->read->handler =  ngx_rtmp_handshake_recv;
     c->write->handler = ngx_rtmp_handshake_send;
 
+    ngx_log_debug0(NGX_LOG_DEBUG_RTMP, s->connection->log, 0,
+            "handshake: start client handshake");
+
     s->hs_bufs[0] = ngx_rtmp_alloc_handshake_buffer(s, 0);
     s->hs_buf = s->hs_bufs[0];
     s->hs_stage = NGX_RTMP_HANDSHAKE_CLIENT_SEND_CHALLENGE;
@@ -589,6 +614,6 @@ ngx_rtmp_client_handshake(ngx_rtmp_session_t *s)
         return;
     }
 
-    ngx_rtmp_handshake_send(c->read);
+    ngx_rtmp_handshake_send(c->write);
 }
 

ngx_rtmp_relay_module.c

@@ -138,7 +138,6 @@ ngx_rtmp_relay_create_app_conf(ngx_conf_t *cf)
     }
 
     ngx_array_init(&racf->targets, cf->pool, 1, sizeof(ngx_rtmp_relay_target_t));
-
     racf->nbuckets = 1024;
     racf->log = &cf->cycle->new_log;
 
@@ -149,6 +148,11 @@ ngx_rtmp_relay_create_app_conf(ngx_conf_t *cf)
 static char *
 ngx_rtmp_relay_merge_app_conf(ngx_conf_t *cf, void *parent, void *child)
 {
+    ngx_rtmp_relay_app_conf_t *conf = child;
+
+    conf->ctx = ngx_pcalloc(cf->pool, sizeof(ngx_rtmp_relay_ctx_t *) 
+            * conf->nbuckets);
+
     return NGX_CONF_OK;
 }
 
@@ -168,7 +172,7 @@ ngx_rtmp_relay_free_peer(ngx_peer_connection_t *pc, void *data,
 
 
 static void
-ngx_rtmp_relay_create(ngx_rtmp_relay_ctx_t *ctx,
+ngx_rtmp_relay_create(ngx_rtmp_session_t *s,
         ngx_rtmp_relay_target_t *target)
 {
     ngx_rtmp_relay_app_conf_t      *racf;
@@ -177,12 +181,17 @@ ngx_rtmp_relay_create(ngx_rtmp_relay_ctx_t *ctx,
     ngx_connection_t               *c;
     ngx_pool_t                     *pool;
     ngx_int_t                       rc;
-    ngx_rtmp_session_t             *s;
     ngx_rtmp_relay_ctx_t           *rctx, **cctx;
     ngx_uint_t                      hash;
     ngx_rtmp_addr_conf_t            addr_conf;
+    ngx_rtmp_conf_ctx_t             addr_ctx;
+    ngx_rtmp_relay_ctx_t           *ctx;
 
-    racf = ngx_rtmp_get_module_app_conf(ctx->session, ngx_rtmp_relay_module);
+    racf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_relay_module);
+    ctx = ngx_rtmp_get_module_ctx(s, ngx_rtmp_relay_module);
+    if (ctx == NULL) {
+        return;
+    }
 
     pool = NULL;
     pool = ngx_create_pool(4096, racf->log);
@@ -241,8 +250,9 @@ ngx_rtmp_relay_create(ngx_rtmp_relay_ctx_t *ctx,
     c->pool = pool;
 
     ngx_memzero(&addr_conf, sizeof(addr_conf));
-    addr_conf.ctx->main_conf = s->main_conf;
+    addr_conf.ctx = &addr_ctx;
-    addr_conf.ctx->srv_conf  = s->srv_conf;
+    addr_ctx.main_conf = s->main_conf;
+    addr_ctx.srv_conf  = s->srv_conf;
     ngx_str_set(&addr_conf.addr_text, "ngx-relay");
 
     rs = ngx_rtmp_init_session(c, &addr_conf);
@@ -251,11 +261,13 @@ ngx_rtmp_relay_create(ngx_rtmp_relay_ctx_t *ctx,
         return;
     }
 
-    rs->app_conf = ctx->session->app_conf;
+    rs->app_conf = s->app_conf;
     rctx->session = rs;
     ngx_rtmp_set_ctx(rs, rctx, ngx_rtmp_relay_module);
 
-    ngx_rtmp_client_handshake(s);
+    ngx_str_set(&rs->flashver, "ngx-local-relay");
+
+    ngx_rtmp_client_handshake(rs);
 
     return;
 
@@ -296,8 +308,8 @@ ngx_rtmp_relay_init(ngx_rtmp_session_t *s, u_char *name)
     ngx_memcpy(ctx->name.data, name, len);
 
     /* find relay stream */
-    hash = ngx_hash_key(name, len) % racf->nbuckets;
+    hash = ngx_hash_key(name, len);
-    cctx = &racf->ctx[hash];
+    cctx = &racf->ctx[hash % racf->nbuckets];
     for (sctx = *cctx; sctx; sctx = sctx->next) {
         if (sctx->name.len == len
             && !ngx_memcmp(name, sctx->name.data, len))
@@ -327,7 +339,7 @@ ngx_rtmp_relay_init(ngx_rtmp_session_t *s, u_char *name)
                 "relay: create: name='%s' url='%V'",
                 name, &target->url);
 
-            ngx_rtmp_relay_create(ctx, target);
+            ngx_rtmp_relay_create(s, target);
         }
     }
 
@@ -420,7 +432,7 @@ ngx_rtmp_relay_send_connect(ngx_rtmp_session_t *s)
 
         { NGX_RTMP_AMF_STRING, 
           ngx_string("flashVer"),
-          "ngx-rtmp-relay", 0 }
+          "ngx-remote-relay", 0 }
     };
 
     static ngx_rtmp_amf_elt_t   out_elts[] = {
@@ -502,6 +514,10 @@ ngx_rtmp_relay_send_publish(ngx_rtmp_session_t *s)
           ngx_null_string,
           &trans, 0 },
 
+        { NGX_RTMP_AMF_NULL,
+          ngx_null_string,
+          NULL, 0 },
+
         { NGX_RTMP_AMF_STRING,
           ngx_null_string,
           NULL, 0 }, /* <- to fill */
@@ -520,8 +536,8 @@ ngx_rtmp_relay_send_publish(ngx_rtmp_session_t *s)
         return NGX_ERROR;
     }
 
-    out_elts[2].data = ctx->name.data;
+    out_elts[3].data = ctx->name.data;
-    out_elts[2].len  = ctx->name.len;
+    out_elts[3].len  = ctx->name.len;
 
     ngx_memzero(&h, sizeof(h));
     h.csid = NGX_RTMP_RELAY_CSID_AMF;
@@ -547,6 +563,10 @@ ngx_rtmp_relay_send_play(ngx_rtmp_session_t *s)
           ngx_null_string,
           &trans, 0 },
 
+        { NGX_RTMP_AMF_NULL,
+          ngx_null_string,
+          NULL, 0 },
+
         { NGX_RTMP_AMF_STRING,
           ngx_null_string,
           NULL, 0 }, /* <- to fill */
@@ -561,8 +581,8 @@ ngx_rtmp_relay_send_play(ngx_rtmp_session_t *s)
         return NGX_ERROR;
     }
 
-    out_elts[2].data = ctx->name.data;
+    out_elts[3].data = ctx->name.data;
-    out_elts[2].len  = ctx->name.len;
+    out_elts[3].len  = ctx->name.len;
 
     ngx_memzero(&h, sizeof(h));
     h.csid = NGX_RTMP_RELAY_CSID_AMF;

stat.xsl

@@ -71,7 +71,7 @@
         <td>
             <a href="">
                 <xsl:attribute name="onclick">
-                    var d=document.getElementById('<xsl:value-of select="name"/>');
+                    var d=document.getElementById('<xsl:value-of select="../../name"/>-<xsl:value-of select="name"/>');
                     d.style.display=d.style.display=='none'?'':'none';
                     return false
                 </xsl:attribute>
@@ -91,7 +91,7 @@
     </tr>
     <tr style="display:none">
         <xsl:attribute name="id">
-            <xsl:value-of select="name"/>
+            <xsl:value-of select="../../name"/>-<xsl:value-of select="name"/>
         </xsl:attribute>
         <td colspan="7" ngcolor="#eeeeee">
             <table cellspacing="1" cellpadding="5">

test/nginx.conf

@@ -44,6 +44,11 @@ rtmp {
             on_record_done http://localhost:8080/record_done;
         }
 
+        application mypull {
+            live on;
+            pull myapp mystream localhost;
+        }
+
     }
 }