@@ -36,6 +36,20 @@ typedef struct {
36
36
#define NGX_RTMP_CONTROL_DROP 0x02
37
37
38
38
39
+ enum {
40
+ NGX_RTMP_CONTROL_DROP_PUBLISHER ,
41
+ NGX_RTMP_CONTROL_DROP_SUBSCRIBER ,
42
+ NGX_RTMP_CONTROL_DROP_CLIENT ,
43
+ };
44
+
45
+
46
+ typedef struct {
47
+ ngx_uint_t method ;
48
+ ngx_str_t addr ;
49
+ ngx_uint_t ndropped ;
50
+ } ngx_rtmp_control_drop_t ;
51
+
52
+
39
53
typedef struct {
40
54
ngx_uint_t control ;
41
55
} ngx_rtmp_control_loc_conf_t ;
@@ -194,6 +208,10 @@ ngx_rtmp_control_parse_live(ngx_http_request_t *r,
194
208
ngx_memzero (& name , sizeof (name ));
195
209
ngx_http_arg (r , (u_char * ) "name" , sizeof ("name" ) - 1 , & name );
196
210
211
+ if (name .len == 0 ) {
212
+ return NGX_CONF_OK ;
213
+ }
214
+
197
215
live -> lacf = core -> cacf -> app_conf [ngx_rtmp_live_module .ctx_index ];
198
216
199
217
/* find live stream by name */
@@ -246,6 +264,7 @@ ngx_rtmp_control_record(ngx_http_request_t *r, ngx_str_t *method)
246
264
goto error ;
247
265
}
248
266
267
+ ngx_memzero (& live , sizeof (live ));
249
268
msg = ngx_rtmp_control_parse_live (r , & core , & live );
250
269
if (msg != NGX_CONF_OK ) {
251
270
goto error ;
@@ -329,68 +348,221 @@ ngx_rtmp_control_record(ngx_http_request_t *r, ngx_str_t *method)
329
348
}
330
349
331
350
351
+ static const char *
352
+ ngx_rtmp_control_drop_session (ngx_http_request_t * r ,
353
+ ngx_rtmp_control_drop_t * drop ,
354
+ ngx_rtmp_live_ctx_t * lctx )
355
+ {
356
+ ngx_rtmp_session_t * s ;
357
+ ngx_str_t * paddr ;
358
+
359
+ s = lctx -> session ;
360
+
361
+ if (s == NULL || s -> connection == NULL )
362
+ {
363
+ return NGX_CONF_OK ;
364
+ }
365
+
366
+ if (drop -> addr .len ) {
367
+ paddr = & s -> connection -> addr_text ;
368
+ if (paddr -> len != drop -> addr .len ||
369
+ ngx_strncmp (paddr -> data , drop -> addr .data , drop -> addr .len ))
370
+ {
371
+ return NGX_CONF_OK ;
372
+ }
373
+ }
374
+
375
+ switch (drop -> method ) {
376
+ case NGX_RTMP_CONTROL_DROP_PUBLISHER :
377
+ if (!lctx -> publishing ) {
378
+ return NGX_CONF_OK ;
379
+ }
380
+
381
+ case NGX_RTMP_CONTROL_DROP_SUBSCRIBER :
382
+ if (lctx -> publishing ) {
383
+ return NGX_CONF_OK ;
384
+ }
385
+
386
+ case NGX_RTMP_CONTROL_DROP_CLIENT :
387
+ break ;
388
+ }
389
+
390
+ ngx_rtmp_finalize_session (s );
391
+ ++ drop -> ndropped ;
392
+
393
+ return NGX_CONF_OK ;
394
+ }
395
+
396
+
397
+ static const char *
398
+ ngx_rtmp_control_drop_stream (ngx_http_request_t * r ,
399
+ ngx_rtmp_control_drop_t * drop ,
400
+ ngx_rtmp_live_stream_t * ls )
401
+ {
402
+ ngx_rtmp_live_ctx_t * lctx ;
403
+ const char * s ;
404
+
405
+ for (lctx = ls -> ctx ; lctx ; lctx = lctx -> next ) {
406
+ s = ngx_rtmp_control_drop_session (r , drop , lctx );
407
+ if (s != NGX_CONF_OK ) {
408
+ return s ;
409
+ }
410
+ }
411
+
412
+ return NGX_CONF_OK ;
413
+ }
414
+
415
+
416
+ static const char *
417
+ ngx_rtmp_control_drop_app (ngx_http_request_t * r ,
418
+ ngx_rtmp_control_drop_t * drop ,
419
+ ngx_rtmp_core_app_conf_t * cacf )
420
+ {
421
+ ngx_rtmp_live_app_conf_t * lacf ;
422
+ ngx_rtmp_live_stream_t * ls ;
423
+ ngx_str_t name ;
424
+ const char * s ;
425
+ size_t len ;
426
+ ngx_uint_t n ;
427
+
428
+ ngx_memzero (& name , sizeof (name ));
429
+ ngx_http_arg (r , (u_char * ) "name" , sizeof ("name" ) - 1 , & name );
430
+
431
+ lacf = cacf -> app_conf [ngx_rtmp_live_module .ctx_index ];
432
+
433
+ if (name .len == 0 ) {
434
+ for (n = 0 ; n < (ngx_uint_t ) lacf -> nbuckets ; ++ n ) {
435
+ for (ls = lacf -> streams [n ]; ls ; ls = ls -> next )
436
+ {
437
+ s = ngx_rtmp_control_drop_stream (r , drop , ls );
438
+ if (s != NGX_CONF_OK ) {
439
+ return s ;
440
+ }
441
+ }
442
+ }
443
+
444
+ return NGX_CONF_OK ;
445
+ }
446
+
447
+ for (ls = lacf -> streams [ngx_hash_key (name .data , name .len ) % lacf -> nbuckets ];
448
+ ls ; ls = ls -> next )
449
+ {
450
+ len = ngx_strlen (ls -> name );
451
+ if (name .len != len || ngx_strncmp (name .data , ls -> name , name .len )) {
452
+ continue ;
453
+ }
454
+
455
+ s = ngx_rtmp_control_drop_stream (r , drop , ls );
456
+ if (s != NGX_CONF_OK ) {
457
+ return s ;
458
+ }
459
+ }
460
+
461
+ return NGX_CONF_OK ;
462
+ }
463
+
464
+
465
+ static const char *
466
+ ngx_rtmp_control_drop_srv (ngx_http_request_t * r ,
467
+ ngx_rtmp_control_drop_t * drop ,
468
+ ngx_rtmp_core_srv_conf_t * cscf )
469
+ {
470
+ ngx_rtmp_core_app_conf_t * * pcacf ;
471
+ ngx_str_t app ;
472
+ ngx_uint_t n ;
473
+ const char * s ;
474
+
475
+ ngx_memzero (& app , sizeof (app ));
476
+ ngx_http_arg (r , (u_char * ) "app" , sizeof ("app" ) - 1 , & app );
477
+
478
+ pcacf = cscf -> applications .elts ;
479
+
480
+ for (n = 0 ; n < cscf -> applications .nelts ; ++ n , ++ pcacf ) {
481
+ if (app .len && ((* pcacf )-> name .len != app .len ||
482
+ ngx_strncmp ((* pcacf )-> name .data , app .data , app .len )))
483
+ {
484
+ continue ;
485
+ }
486
+
487
+ s = ngx_rtmp_control_drop_app (r , drop , * pcacf );
488
+ if (s != NGX_CONF_OK ) {
489
+ return s ;
490
+ }
491
+ }
492
+
493
+ return NGX_CONF_OK ;
494
+ }
495
+
496
+
497
+ static const char *
498
+ ngx_rtmp_control_drop_main (ngx_http_request_t * r ,
499
+ ngx_rtmp_control_drop_t * drop ,
500
+ ngx_rtmp_core_main_conf_t * cmcf )
501
+ {
502
+ ngx_rtmp_core_srv_conf_t * * pcscf ;
503
+ ngx_str_t srv ;
504
+ ngx_uint_t sn ;
505
+
506
+ sn = 0 ;
507
+ if (ngx_http_arg (r , (u_char * ) "srv" , sizeof ("srv" ) - 1 , & srv ) == NGX_OK ) {
508
+ sn = ngx_atoi (srv .data , srv .len );
509
+ }
510
+
511
+ if (sn >= cmcf -> servers .nelts ) {
512
+ return "Server index out of range" ;
513
+ }
514
+
515
+ pcscf = cmcf -> servers .elts ;
516
+ pcscf += sn ;
517
+
518
+ return ngx_rtmp_control_drop_srv (r , drop , * pcscf );
519
+ }
520
+
521
+
332
522
static ngx_int_t
333
523
ngx_rtmp_control_drop (ngx_http_request_t * r , ngx_str_t * method )
334
524
{
335
- ngx_rtmp_control_core_t core ;
336
- ngx_rtmp_control_live_t live ;
337
- ngx_rtmp_live_ctx_t * lctx ;
338
- ngx_str_t addr , * paddr ;
339
- const char * msg ;
340
- ngx_uint_t ndropped ;
525
+ ngx_rtmp_control_drop_t drop ;
341
526
size_t len ;
342
527
u_char * p ;
343
528
ngx_buf_t * b ;
344
529
ngx_chain_t cl ;
530
+ const char * msg ;
345
531
346
- msg = ngx_rtmp_control_parse_core (r , & core );
347
- if (msg != NGX_CONF_OK ) {
348
- goto error ;
349
- }
350
-
351
- msg = ngx_rtmp_control_parse_live (r , & core , & live );
352
- if (msg != NGX_CONF_OK ) {
532
+ if (ngx_rtmp_core_main_conf == NULL ) {
533
+ msg = "Empty main conf" ;
353
534
goto error ;
354
535
}
355
536
356
- ndropped = 0 ;
537
+ ngx_memzero ( & drop , sizeof ( drop )) ;
357
538
358
539
if (method -> len == sizeof ("publisher" ) - 1 &&
359
- ngx_strncmp (method -> data , "publisher" , method -> len ) == 0 )
540
+ ngx_memcmp (method -> data , "publisher" , method -> len ) == 0 )
360
541
{
361
- for (lctx = live .ls -> ctx ; lctx ; lctx = lctx -> next ) {
362
- if (lctx -> publishing ) {
363
- ngx_rtmp_finalize_session (lctx -> session );
364
- ++ ndropped ;
365
- break ;
366
- }
367
- }
542
+ drop .method = NGX_RTMP_CONTROL_DROP_PUBLISHER ;
368
543
369
- } else if (method -> len == sizeof ("client " ) - 1 &&
370
- ngx_strncmp (method -> data , "client " , method -> len ) == 0 )
544
+ } else if (method -> len == sizeof ("subscriber " ) - 1 &&
545
+ ngx_memcmp (method -> data , "subscriber " , method -> len ) == 0 )
371
546
{
372
- ngx_memzero (& addr , sizeof (addr ));
373
- ngx_http_arg (r , (u_char * ) "addr" , sizeof ("addr" ) - 1 , & addr );
374
-
375
- for (lctx = live .ls -> ctx ; lctx ; lctx = lctx -> next ) {
376
- if (addr .len && lctx -> session && lctx -> session -> connection ) {
377
- paddr = & lctx -> session -> connection -> addr_text ;
378
- if (paddr -> len != addr .len ||
379
- ngx_strncmp (paddr -> data , addr .data , addr .len ))
380
- {
381
- continue ;
382
- }
383
- }
547
+ drop .method = NGX_RTMP_CONTROL_DROP_SUBSCRIBER ;
384
548
385
- ngx_rtmp_finalize_session (lctx -> session );
386
- ++ ndropped ;
387
- }
549
+ } else if (method -> len == sizeof ("client" ) - 1 &&
550
+ ngx_memcmp (method -> data , "client" , method -> len ) == 0 )
551
+ {
552
+ drop .method = NGX_RTMP_CONTROL_DROP_CLIENT ;
388
553
389
554
} else {
390
555
msg = "Undefined method" ;
391
556
goto error ;
392
557
}
393
558
559
+ ngx_http_arg (r , (u_char * ) "addr" , sizeof ("addr" ) - 1 , & drop .addr );
560
+
561
+ msg = ngx_rtmp_control_drop_main (r , & drop , ngx_rtmp_core_main_conf );
562
+ if (msg != NGX_CONF_OK ) {
563
+ goto error ;
564
+ }
565
+
394
566
/* output ndropped */
395
567
396
568
len = NGX_INT_T_LEN ;
@@ -400,7 +572,7 @@ ngx_rtmp_control_drop(ngx_http_request_t *r, ngx_str_t *method)
400
572
return NGX_ERROR ;
401
573
}
402
574
403
- len = (size_t ) (ngx_snprintf (p , len , "%ui" , ndropped ) - p );
575
+ len = (size_t ) (ngx_snprintf (p , len , "%ui" , drop . ndropped ) - p );
404
576
405
577
r -> headers_out .status = NGX_HTTP_OK ;
406
578
r -> headers_out .content_length_n = len ;
0 commit comments