diff --git a/internal/version/version.go b/internal/version/version.go
index 0fe33ae5c7..bf69fc8bdc 100644
--- a/internal/version/version.go
+++ b/internal/version/version.go
@@ -15,7 +15,7 @@ func VersionMinor() int {
}
func VersionRevision() int {
- return 41
+ return 44
}
func Version() string {
diff --git a/skills/srs-develop/references/integration-tests.md b/skills/srs-develop/references/integration-tests.md
index b81615bf80..0e6ce5c09c 100644
--- a/skills/srs-develop/references/integration-tests.md
+++ b/skills/srs-develop/references/integration-tests.md
@@ -24,15 +24,15 @@ Run focused and component-native tests first, then run every command below seque
```bash
bash skills/srs-develop/scripts/proxy-e2e-redis-test.sh
```
-6. RTMP publish with RTMP, HTTP-FLV, and HLS playback verification; WHEP remains a placeholder:
+6. RTMP publish with RTMP, HTTP-FLV, HLS, and WHEP playback verification:
```bash
bash skills/srs-develop/scripts/proxy-e2e-transmux-test.sh
```
-7. SRT publish with SRT, RTMP, HTTP-FLV, and HLS playback verification; WHEP remains a placeholder:
+7. SRT publish with SRT, RTMP, HTTP-FLV, HLS, and WHEP playback verification:
```bash
bash skills/srs-develop/scripts/proxy-e2e-srt-test.sh
```
-8. WHIP publish with RTMP, HTTP-FLV, and HLS playback verification; WHEP remains a placeholder:
+8. WHIP publish with RTMP, HTTP-FLV, HLS, and WHEP playback verification:
```bash
bash skills/srs-develop/scripts/proxy-e2e-whip-test.sh
```
@@ -41,7 +41,7 @@ Run focused and component-native tests first, then run every command below seque
bash skills/srs-develop/scripts/proxy-e2e-bearer-auth-test.sh
```
-The SRT test requires an FFmpeg build with libsrt. The WHIP test requires the `whip` muxer and OpenSSL. Both scripts automatically run `skills/srs-develop/scripts/setup-ffmpeg-with-whip.sh` on macOS when no suitable FFmpeg is available. If an environmental dependency is unavailable, run the script, preserve its exact result, and report the blocked coverage instead of claiming full verification.
+The SRT test requires an FFmpeg build with libsrt. The WHIP test requires the `whip` muxer and OpenSSL. Both scripts automatically run `skills/srs-develop/scripts/setup-ffmpeg-with-whip.sh` on macOS when no suitable FFmpeg is available. If an environmental dependency is unavailable, run the script, preserve its exact result, and report the blocked coverage instead of claiming full verification. The transmux, SRT, and WHIP tests play WHEP with `tools/pion-whep`, because FFmpeg has no WHEP demuxer; they build it with `go` when the binary is missing or stale.
Run feature-specific bundled tests in addition to this matrix when the routed workflow or the touched area requires them:
diff --git a/skills/srs-develop/scripts/proxy-e2e-srt-test.sh b/skills/srs-develop/scripts/proxy-e2e-srt-test.sh
index 0e78743cc7..f0be1baffe 100755
--- a/skills/srs-develop/scripts/proxy-e2e-srt-test.sh
+++ b/skills/srs-develop/scripts/proxy-e2e-srt-test.sh
@@ -6,7 +6,7 @@
# - RTMP play (via srt_to_rtmp on origin)
# - HTTP-FLV (HTTP remux of the bridged RTMP)
# - HLS (m3u8 + TS segments)
-# - WebRTC WHEP (placeholder only, not actually verified here)
+# - WebRTC WHEP (via rtmp_to_rtc on origin, played with tools/pion-whep)
set -e
SCRIPT_DIR="$(cd -P "$(dirname "$0")" && pwd)"
@@ -100,6 +100,50 @@ probe_has_audio_video() {
fi
}
+# Play over WHEP with tools/pion-whep, because FFmpeg has no WHEP demuxer, and
+# require both video and audio RTP packets. The tool is rebuilt when its binary
+# is missing or older than its sources.
+verify_whep_playback() {
+ local url="$1"
+ local log="$2"
+ local tool_dir="$WORKSPACE/tools/pion-whep"
+ local tool_bin="$tool_dir/objs/pion-whep"
+ local summary video audio
+
+ if [[ ! -x "$tool_bin" ]] || [[ -n "$(find "$tool_dir" -maxdepth 1 \( -name '*.go' -o -name 'go.mod' -o -name 'go.sum' \) -newer "$tool_bin")" ]]; then
+ echo "Building pion-whep: $tool_bin"
+ (cd "$tool_dir" && mkdir -p objs && go build -o objs/pion-whep .)
+ fi
+
+ echo "Verifying WHEP playback: $url"
+ if ! "$tool_bin" -hide_banner -f whep -i "$url" -t 5 -f null - >"$log" 2>&1; then
+ echo "FAIL: WHEP playback failed. pion-whep log:" >&2
+ cat "$log" >&2
+ exit 1
+ fi
+
+ summary="$(grep -a "^Received video=" "$log" | tail -1)"
+ video="$(echo "$summary" | sed -n 's/^Received video=\([0-9]*\),.*/\1/p')"
+ audio="$(echo "$summary" | sed -n 's/^Received video=[0-9]*, audio=\([0-9]*\) .*/\1/p')"
+ echo "pion-whep: $summary"
+
+ if [[ -n "$video" && "$video" -gt 0 ]]; then
+ echo "PASS: WHEP video packets received."
+ else
+ echo "FAIL: WHEP no video packets received." >&2
+ cat "$log" >&2
+ exit 1
+ fi
+
+ if [[ -n "$audio" && "$audio" -gt 0 ]]; then
+ echo "PASS: WHEP audio packets received."
+ else
+ echo "FAIL: WHEP no audio packets received." >&2
+ cat "$log" >&2
+ exit 1
+ fi
+}
+
wait_for_hls_playlist() {
local url="$1"
local deadline=45
@@ -133,6 +177,10 @@ if ! command -v curl &>/dev/null; then
echo "Error: curl not found in PATH" >&2
exit 1
fi
+if ! command -v go &>/dev/null; then
+ echo "Error: go not found in PATH" >&2
+ exit 1
+fi
# SRT URLs need libsrt compiled into ffmpeg/ffprobe. The default Homebrew
# ffmpeg formula does NOT include libsrt. Resolution order:
@@ -233,7 +281,10 @@ echo "Proxy started."
echo "=== Step 4: Starting SRS origin ==="
ulimit -n 10000 2>/dev/null || true
cd "$WORKSPACE/trunk"
-./objs/srs -c conf/origin1-for-proxy.conf >/tmp/srs-origin-srt-e2e.log 2>&1 &
+# The proxy rewrites only the port of the WebRTC candidate, so the origin must
+# advertise an IP the WHEP player can reach.
+env CANDIDATE="127.0.0.1" \
+ ./objs/srs -c conf/origin1-for-proxy.conf >/tmp/srs-origin-srt-e2e.log 2>&1 &
ORIGIN_PID=$!
echo "SRS origin PID: $ORIGIN_PID"
@@ -285,11 +336,10 @@ HLS_URL="http://localhost:$PROXY_HTTP_SERVER_PORT/$STREAM_URL.m3u8"
wait_for_hls_playlist "$HLS_URL"
probe_has_audio_video "HLS" "$HLS_URL"
-# --- Step 10: WebRTC WHEP playback (placeholder) ---
-echo "=== Step 10: WebRTC WHEP playback (placeholder) ==="
-echo "SKIP: WebRTC WHEP playback is not verified by this script."
-echo " The origin has rtmp_to_rtc enabled, so SRT->RTMP->RTC should work end-to-end,"
-echo " but actual playback verification is intentionally left as a TODO here."
+# --- Step 10: Verify WebRTC WHEP playback (srt_to_rtmp + rtmp_to_rtc) ---
+echo "=== Step 10: Verifying WebRTC WHEP playback via proxy ==="
+verify_whep_playback "http://localhost:$PROXY_HTTP_API_PORT/rtc/v1/whep/?app=live&stream=${STREAM_URL#live/}" \
+ /tmp/srs-pion-whep-srt-e2e.log
echo ""
echo "=== E2E SRT Proxy Test PASSED ==="
diff --git a/skills/srs-develop/scripts/proxy-e2e-transmux-test.sh b/skills/srs-develop/scripts/proxy-e2e-transmux-test.sh
index 61ba84a7cc..fa01b8d678 100755
--- a/skills/srs-develop/scripts/proxy-e2e-transmux-test.sh
+++ b/skills/srs-develop/scripts/proxy-e2e-transmux-test.sh
@@ -1,8 +1,8 @@
#!/bin/bash
# E2E test for RTMP-to-multiple-protocol transmuxing through the proxy:
# starts one proxy with memory load balancer + one SRS origin, publishes one
-# RTMP stream, then verifies RTMP, HTTP-FLV, and HLS playback through the
-# proxy. WebRTC WHEP verification is intentionally a placeholder (not run).
+# RTMP stream, then verifies RTMP, HTTP-FLV, HLS, and WebRTC WHEP playback
+# through the proxy. WHEP is played with tools/pion-whep.
set -e
SCRIPT_DIR="$(cd -P "$(dirname "$0")" && pwd)"
@@ -92,6 +92,50 @@ probe_has_audio_video() {
fi
}
+# Play over WHEP with tools/pion-whep, because FFmpeg has no WHEP demuxer, and
+# require both video and audio RTP packets. The tool is rebuilt when its binary
+# is missing or older than its sources.
+verify_whep_playback() {
+ local url="$1"
+ local log="$2"
+ local tool_dir="$WORKSPACE/tools/pion-whep"
+ local tool_bin="$tool_dir/objs/pion-whep"
+ local summary video audio
+
+ if [[ ! -x "$tool_bin" ]] || [[ -n "$(find "$tool_dir" -maxdepth 1 \( -name '*.go' -o -name 'go.mod' -o -name 'go.sum' \) -newer "$tool_bin")" ]]; then
+ echo "Building pion-whep: $tool_bin"
+ (cd "$tool_dir" && mkdir -p objs && go build -o objs/pion-whep .)
+ fi
+
+ echo "Verifying WHEP playback: $url"
+ if ! "$tool_bin" -hide_banner -f whep -i "$url" -t 5 -f null - >"$log" 2>&1; then
+ echo "FAIL: WHEP playback failed. pion-whep log:" >&2
+ cat "$log" >&2
+ exit 1
+ fi
+
+ summary="$(grep -a "^Received video=" "$log" | tail -1)"
+ video="$(echo "$summary" | sed -n 's/^Received video=\([0-9]*\),.*/\1/p')"
+ audio="$(echo "$summary" | sed -n 's/^Received video=[0-9]*, audio=\([0-9]*\) .*/\1/p')"
+ echo "pion-whep: $summary"
+
+ if [[ -n "$video" && "$video" -gt 0 ]]; then
+ echo "PASS: WHEP video packets received."
+ else
+ echo "FAIL: WHEP no video packets received." >&2
+ cat "$log" >&2
+ exit 1
+ fi
+
+ if [[ -n "$audio" && "$audio" -gt 0 ]]; then
+ echo "PASS: WHEP audio packets received."
+ else
+ echo "FAIL: WHEP no audio packets received." >&2
+ cat "$log" >&2
+ exit 1
+ fi
+}
+
wait_for_hls_playlist() {
local url="$1"
local deadline=45
@@ -133,6 +177,10 @@ if ! command -v curl &>/dev/null; then
echo "Error: curl not found in PATH" >&2
exit 1
fi
+if ! command -v go &>/dev/null; then
+ echo "Error: go not found in PATH" >&2
+ exit 1
+fi
# --- Step 0: Clean up stale state ---
rm -f "$WORKSPACE/trunk/objs/origin1.pid"
@@ -184,7 +232,10 @@ echo "Proxy started."
echo "=== Step 4: Starting SRS origin ==="
ulimit -n 10000 2>/dev/null || true
cd "$WORKSPACE/trunk"
-./objs/srs -c conf/origin1-for-proxy.conf >/tmp/srs-origin-transmux-e2e.log 2>&1 &
+# The proxy rewrites only the port of the WebRTC candidate, so the origin must
+# advertise an IP the WHEP player can reach.
+env CANDIDATE="127.0.0.1" \
+ ./objs/srs -c conf/origin1-for-proxy.conf >/tmp/srs-origin-transmux-e2e.log 2>&1 &
ORIGIN_PID=$!
echo "SRS origin PID: $ORIGIN_PID"
@@ -230,11 +281,10 @@ HLS_URL="http://localhost:$PROXY_HTTP_SERVER_PORT/$STREAM_URL.m3u8"
wait_for_hls_playlist "$HLS_URL"
probe_has_audio_video "HLS" "$HLS_URL"
-# --- Step 9: WebRTC WHEP playback (placeholder) ---
-echo "=== Step 9: WebRTC WHEP playback (placeholder) ==="
-echo "SKIP: WebRTC WHEP playback is not verified by this script."
-echo " The origin has rtmp_to_rtc enabled, so RTMP->RTC should work end-to-end,"
-echo " but actual playback verification is intentionally left as a TODO here."
+# --- Step 9: Verify WebRTC WHEP playback (rtmp_to_rtc) ---
+echo "=== Step 9: Verifying WebRTC WHEP playback via proxy ==="
+verify_whep_playback "http://localhost:$PROXY_HTTP_API_PORT/rtc/v1/whep/?app=live&stream=${STREAM_URL#live/}" \
+ /tmp/srs-pion-whep-transmux-e2e.log
echo ""
echo "NOTE: RTSP is not tested here because the Go proxy currently has no RTSP listener."
diff --git a/skills/srs-develop/scripts/proxy-e2e-whip-test.sh b/skills/srs-develop/scripts/proxy-e2e-whip-test.sh
index 05a50ec855..307481e6b5 100755
--- a/skills/srs-develop/scripts/proxy-e2e-whip-test.sh
+++ b/skills/srs-develop/scripts/proxy-e2e-whip-test.sh
@@ -6,7 +6,7 @@
# - RTMP play (via rtc_to_rtmp on origin)
# - HTTP-FLV (HTTP remux of the bridged RTMP)
# - HLS (m3u8 + TS segments)
-# - WebRTC WHEP (placeholder only, not actually verified here)
+# - WebRTC WHEP (RTC passthrough, played with tools/pion-whep)
set -e
SCRIPT_DIR="$(cd -P "$(dirname "$0")" && pwd)"
@@ -107,6 +107,50 @@ probe_has_audio_video() {
fi
}
+# Play over WHEP with tools/pion-whep, because FFmpeg has no WHEP demuxer, and
+# require both video and audio RTP packets. The tool is rebuilt when its binary
+# is missing or older than its sources.
+verify_whep_playback() {
+ local url="$1"
+ local log="$2"
+ local tool_dir="$WORKSPACE/tools/pion-whep"
+ local tool_bin="$tool_dir/objs/pion-whep"
+ local summary video audio
+
+ if [[ ! -x "$tool_bin" ]] || [[ -n "$(find "$tool_dir" -maxdepth 1 \( -name '*.go' -o -name 'go.mod' -o -name 'go.sum' \) -newer "$tool_bin")" ]]; then
+ echo "Building pion-whep: $tool_bin"
+ (cd "$tool_dir" && mkdir -p objs && go build -o objs/pion-whep .)
+ fi
+
+ echo "Verifying WHEP playback: $url"
+ if ! "$tool_bin" -hide_banner -f whep -i "$url" -t 5 -f null - >"$log" 2>&1; then
+ echo "FAIL: WHEP playback failed. pion-whep log:" >&2
+ cat "$log" >&2
+ exit 1
+ fi
+
+ summary="$(grep -a "^Received video=" "$log" | tail -1)"
+ video="$(echo "$summary" | sed -n 's/^Received video=\([0-9]*\),.*/\1/p')"
+ audio="$(echo "$summary" | sed -n 's/^Received video=[0-9]*, audio=\([0-9]*\) .*/\1/p')"
+ echo "pion-whep: $summary"
+
+ if [[ -n "$video" && "$video" -gt 0 ]]; then
+ echo "PASS: WHEP video packets received."
+ else
+ echo "FAIL: WHEP no video packets received." >&2
+ cat "$log" >&2
+ exit 1
+ fi
+
+ if [[ -n "$audio" && "$audio" -gt 0 ]]; then
+ echo "PASS: WHEP audio packets received."
+ else
+ echo "FAIL: WHEP no audio packets received." >&2
+ cat "$log" >&2
+ exit 1
+ fi
+}
+
wait_for_hls_playlist() {
local url="$1"
local deadline=60
@@ -183,6 +227,10 @@ if ! command -v curl &>/dev/null; then
echo "Error: curl not found in PATH" >&2
exit 1
fi
+if ! command -v go &>/dev/null; then
+ echo "Error: go not found in PATH" >&2
+ exit 1
+fi
# WHIP needs an ffmpeg with the `whip` muxer (added in ffmpeg 7.1, requires
# --enable-openssl at build time for DTLS-SRTP). Neither vanilla brew nor the
@@ -340,11 +388,10 @@ wait_for_hls_playlist "$HLS_URL"
wait_for_hls_to_skip_first_segment "$HLS_URL"
probe_has_audio_video "HLS" "$HLS_URL"
-# --- Step 9: WebRTC WHEP playback (placeholder) ---
-echo "=== Step 9: WebRTC WHEP playback (placeholder) ==="
-echo "SKIP: WebRTC WHEP playback is not verified by this script."
-echo " The origin has rtmp_to_rtc enabled, so WHIP->RTMP->RTC should work end-to-end,"
-echo " but actual playback verification is intentionally left as a TODO here."
+# --- Step 9: Verify WebRTC WHEP playback (RTC passthrough) ---
+echo "=== Step 9: Verifying WebRTC WHEP playback via proxy ==="
+verify_whep_playback "http://localhost:$PROXY_HTTP_API_PORT/rtc/v1/whep/?app=live&stream=$STREAM_NAME" \
+ /tmp/srs-pion-whep-whip-e2e.log
echo ""
echo "=== E2E WHIP Proxy Test PASSED ==="
diff --git a/trunk/configure b/trunk/configure
index 087d0ef3ec..ce2b44739e 100755
--- a/trunk/configure
+++ b/trunk/configure
@@ -388,7 +388,8 @@ if [[ $SRS_UTEST == YES ]]; then
MODULE_FILES+=("srs_utest_ai01" "srs_utest_ai02" "srs_utest_ai03" "srs_utest_ai04" "srs_utest_ai05"
"srs_utest_ai06" "srs_utest_ai07" "srs_utest_ai08" "srs_utest_ai09" "srs_utest_ai10" "srs_utest_ai11"
"srs_utest_ai12" "srs_utest_ai13" "srs_utest_ai14" "srs_utest_ai15" "srs_utest_ai16" "srs_utest_ai17"
- "srs_utest_ai18" "srs_utest_ai19" "srs_utest_ai20" "srs_utest_ai24" "srs_utest_ai25" "srs_utest_ai26" "srs_utest_ai27")
+ "srs_utest_ai18" "srs_utest_ai19" "srs_utest_ai20" "srs_utest_ai24" "srs_utest_ai25" "srs_utest_ai26" "srs_utest_ai27"
+ "srs_utest_ai28" "srs_utest_ai29" "srs_utest_ai30" "srs_utest_ai31" "srs_utest_ai32")
if [[ $SRS_GB28181 == YES ]]; then
MODULE_FILES+=("srs_utest_manual_gb28181" "srs_utest_ai23")
fi
diff --git a/trunk/doc/CHANGELOG.md b/trunk/doc/CHANGELOG.md
index 4363a9d1e3..d4ec39ef52 100644
--- a/trunk/doc/CHANGELOG.md
+++ b/trunk/doc/CHANGELOG.md
@@ -7,6 +7,9 @@ The changelog for SRS.
## SRS 8.0 Changelog
+* v8.0, 2026-09-22, Timer: Notify every subscriber when one unsubscribes during the round. v8.0.44
+* v8.0, 2026-09-22, Log: Clear the log file descriptor when the logger reopens it. v8.0.43
+* v8.0, 2026-09-22, HTTP: Clamp the VOD MP4 range end to the last byte of the file. v8.0.42
* v8.0, 2026-09-21, Merge [#4746](https://github.com/ossrs/srs/pull/4746): WebRTC: Support RFC 4588 RTX retransmission with SSRC multiplexing, preferred by nack_prefer_rtx. v8.0.41 (#4746)
* v8.0, 2026-09-21, RTC: Reset the play track cache when the publisher republishes with new SSRCs. v8.0.40
* v8.0, 2026-09-21, RTC: Resolve the publish and play tracks once per SSRC through the fast cache. v8.0.39
diff --git a/trunk/src/app/srs_app_config.hpp b/trunk/src/app/srs_app_config.hpp
index 572ab443fc..732c97ddde 100644
--- a/trunk/src/app/srs_app_config.hpp
+++ b/trunk/src/app/srs_app_config.hpp
@@ -352,6 +352,9 @@ class ISrsAppConfig : public ISrsConfig
virtual std::string get_https_stream_ssl_cert() = 0;
virtual std::string get_http_stream_dir() = 0;
virtual bool get_http_stream_crossdomain() = 0;
+ virtual bool get_vhost_http_enabled(std::string vhost) = 0;
+ virtual std::string get_vhost_http_mount(std::string vhost) = 0;
+ virtual std::string get_vhost_http_dir(std::string vhost) = 0;
public:
// WebRTC config
@@ -377,8 +380,21 @@ class ISrsAppConfig : public ISrsConfig
public:
// SRT config
virtual std::vector get_srt_listens() = 0;
+ virtual int64_t get_srto_maxbw() = 0;
+ virtual int get_srto_mss() = 0;
+ virtual bool get_srto_tsbpdmode() = 0;
+ virtual int get_srto_latency() = 0;
+ virtual int get_srto_recv_latency() = 0;
+ virtual int get_srto_peer_latency() = 0;
+ virtual bool get_srto_tlpktdrop() = 0;
+ virtual srs_utime_t get_srto_conntimeout() = 0;
// Get the srt SRTO_PEERIDLETIMEO, peer idle timeout, default is 10000ms.
virtual srs_utime_t get_srto_peeridletimeout() = 0;
+ virtual int get_srto_sendbuf() = 0;
+ virtual int get_srto_recvbuf() = 0;
+ virtual int get_srto_payloadsize() = 0;
+ virtual std::string get_srto_passphrase() = 0;
+ virtual int get_srto_pbkeylen() = 0;
public:
// Stream caster config
@@ -640,6 +656,13 @@ class ISrsAppConfig : public ISrsConfig
virtual std::string get_engine_output(SrsConfDirective *conf) = 0;
virtual bool get_security_enabled(std::string vhost) = 0;
virtual SrsConfDirective *get_security_rules(std::string vhost) = 0;
+ // Whether write log to file, otherwise to console.
+ virtual bool get_log_tank_file() = 0;
+ // The file to write log to, empty if not configured.
+ virtual std::string get_log_file() = 0;
+ virtual std::string get_log_level() = 0;
+ virtual std::string get_log_level_v2() = 0;
+ virtual bool get_utc_time() = 0;
};
// The config service provider.
diff --git a/trunk/src/app/srs_app_factory.cpp b/trunk/src/app/srs_app_factory.cpp
index da41a504ed..252b9493e2 100644
--- a/trunk/src/app/srs_app_factory.cpp
+++ b/trunk/src/app/srs_app_factory.cpp
@@ -27,6 +27,7 @@
#include
#include
#include
+#include
#include
#include
#include
@@ -92,7 +93,9 @@ ISrsOriginHub *SrsAppFactory::create_origin_hub()
ISrsHourGlass *SrsAppFactory::create_hourglass(const std::string &name, ISrsHourGlassHandler *handler, srs_utime_t interval)
{
- return new SrsHourGlass(name, handler, interval);
+ SrsHourGlass *timer = new SrsHourGlass(name, handler, interval);
+ timer->assemble();
+ return timer;
}
ISrsBasicRtmpClient *SrsAppFactory::create_rtmp_client(std::string url, srs_utime_t cto, srs_utime_t sto)
@@ -184,6 +187,11 @@ ISrsIpListener *SrsAppFactory::create_tcp_listener(ISrsTcpHandler *handler)
return new SrsTcpListener(handler);
}
+ISrsSrtListener *SrsAppFactory::create_srt_listener(ISrsSrtHandler *handler, std::string ip, int port)
+{
+ return new SrsSrtListener(handler, ip, port);
+}
+
ISrsRtcConnection *SrsAppFactory::create_rtc_connection(ISrsExecRtcAsyncTask *exec, const SrsContextId &cid)
{
SrsRtcConnection *session = new SrsRtcConnection(exec, cid);
diff --git a/trunk/src/app/srs_app_factory.hpp b/trunk/src/app/srs_app_factory.hpp
index 8fd41ed7f3..947b939b12 100644
--- a/trunk/src/app/srs_app_factory.hpp
+++ b/trunk/src/app/srs_app_factory.hpp
@@ -55,6 +55,8 @@ class SrsRtcFrameBuilder;
class ISrsFrameTarget;
class ISrsRtcFrameBuilderAudioPacketCache;
class ISrsAudioTranscoder;
+class ISrsSrtListener;
+class ISrsSrtHandler;
// The factory to create app objects.
class ISrsAppFactory : public ISrsKernelFactory
@@ -92,6 +94,7 @@ class ISrsAppFactory : public ISrsKernelFactory
virtual ISrsFragmentedMp4 *create_fragmented_mp4() = 0;
virtual SrsHlsM4sSegment *create_hls_m4s_segment(ISrsFileWriter *fw) = 0;
virtual ISrsIpListener *create_tcp_listener(ISrsTcpHandler *handler) = 0;
+ virtual ISrsSrtListener *create_srt_listener(ISrsSrtHandler *handler, std::string ip, int port) = 0;
virtual ISrsRtcConnection *create_rtc_connection(ISrsExecRtcAsyncTask *exec, const SrsContextId &cid) = 0;
virtual ISrsFFMPEG *create_ffmpeg(std::string ffmpeg_bin) = 0;
virtual ISrsIngesterFFMPEG *create_ingester_ffmpeg() = 0;
@@ -146,6 +149,7 @@ class SrsAppFactory : public ISrsAppFactory
virtual ISrsFragmentedMp4 *create_fragmented_mp4();
virtual SrsHlsM4sSegment *create_hls_m4s_segment(ISrsFileWriter *fw);
virtual ISrsIpListener *create_tcp_listener(ISrsTcpHandler *handler);
+ virtual ISrsSrtListener *create_srt_listener(ISrsSrtHandler *handler, std::string ip, int port);
virtual ISrsRtcConnection *create_rtc_connection(ISrsExecRtcAsyncTask *exec, const SrsContextId &cid);
virtual ISrsFFMPEG *create_ffmpeg(std::string ffmpeg_bin);
virtual ISrsIngesterFFMPEG *create_ingester_ffmpeg();
diff --git a/trunk/src/app/srs_app_gb28181.cpp b/trunk/src/app/srs_app_gb28181.cpp
index 51be1af693..b92dca6440 100644
--- a/trunk/src/app/srs_app_gb28181.cpp
+++ b/trunk/src/app/srs_app_gb28181.cpp
@@ -76,6 +76,7 @@ SrsGbSession::SrsGbSession() : media_(new SrsGbMediaTcpConn())
reinviting_starttime_ = 0;
ppp_ = new SrsAlonePithyPrint();
+ ppp_->assemble();
startime_ = srs_time_now_realtime();
connecting_starttime_ = startime_;
media_connect_timeout_ = 0;
diff --git a/trunk/src/app/srs_app_http_conn.cpp b/trunk/src/app/srs_app_http_conn.cpp
index eb6b5f273c..f500e15099 100644
--- a/trunk/src/app/srs_app_http_conn.cpp
+++ b/trunk/src/app/srs_app_http_conn.cpp
@@ -566,11 +566,14 @@ ISrsHttpServer::~ISrsHttpServer()
SrsHttpServer::SrsHttpServer()
{
http_stream_ = new SrsHttpStreamServer();
- http_stream_->assemble();
-
http_static_ = new SrsHttpStaticServer();
}
+void SrsHttpServer::assemble()
+{
+ http_stream_->assemble();
+}
+
SrsHttpServer::~SrsHttpServer()
{
srs_freep(http_stream_);
@@ -597,7 +600,6 @@ srs_error_t SrsHttpServer::initialize()
return err;
}
-// LCOV_EXCL_START
srs_error_t SrsHttpServer::handle(std::string pattern, ISrsHttpHandler *handler)
{
return http_static_->mux()->handle(pattern, handler);
@@ -631,7 +633,6 @@ srs_error_t SrsHttpServer::serve_http(ISrsHttpResponseWriter *w, ISrsHttpMessage
// Use http static as default server.
return http_static_->mux()->serve_http(w, r);
}
-// LCOV_EXCL_STOP
srs_error_t SrsHttpServer::http_mount(ISrsRequest *r)
{
diff --git a/trunk/src/app/srs_app_http_conn.hpp b/trunk/src/app/srs_app_http_conn.hpp
index d696b16145..8420d91c72 100644
--- a/trunk/src/app/srs_app_http_conn.hpp
+++ b/trunk/src/app/srs_app_http_conn.hpp
@@ -276,6 +276,7 @@ class SrsHttpServer : public ISrsHttpServer
virtual ~SrsHttpServer();
public:
+ void assemble(); // Construct object, to avoid call function in constructor.
virtual srs_error_t initialize();
// Interface ISrsCommonHttpHandler
public:
diff --git a/trunk/src/app/srs_app_http_static.cpp b/trunk/src/app/srs_app_http_static.cpp
index c3ae3b175d..f88264b82a 100644
--- a/trunk/src/app/srs_app_http_static.cpp
+++ b/trunk/src/app/srs_app_http_static.cpp
@@ -134,6 +134,16 @@ srs_error_t SrsHlsStream::serve_m3u8_ctx(ISrsHttpResponseWriter *w, ISrsHttpMess
} else {
// Create a m3u8 in memory, contains the session id(ctx).
err = serve_new_session(w, r, req, ctx);
+
+ // Never keep the ctx of a refused viewer alive, because the client may choose the ctx, and a
+ // retry with it would be served as an existing session, which skips the security check and
+ // the on_play hook. The viewer was added to statistic before the checks and has no session
+ // to expire, so remove it now.
+ if (err != srs_success) {
+ stat_->on_disconnect(ctx, err);
+ srs_http_stream_serve_error(w, err);
+ return err;
+ }
}
// Always make the ctx alive now.
@@ -411,6 +421,7 @@ void SrsHlsStream::http_hooks_on_stop(ISrsRequest *req)
return;
}
+// LCOV_EXCL_STOP
srs_error_t SrsHlsStream::on_timer(srs_utime_t interval)
{
@@ -421,16 +432,15 @@ srs_error_t SrsHlsStream::on_timer(srs_utime_t interval)
string ctx = it->first;
SrsHlsVirtualConn *info = it->second;
- srs_utime_t hls_window = _srs_config->get_hls_window(info->req_->vhost_);
+ srs_utime_t hls_window = config_->get_hls_window(info->req_->vhost_);
if (info->request_time_ + (2 * hls_window) < srs_time_now_cached()) {
SrsContextRestore(_srs_context->get_id());
_srs_context->set_id(SrsContextId().set_value(ctx));
http_hooks_on_stop(info->req_);
- SrsStatistic *stat = _srs_stat;
// TODO: FIXME: Should finger out the err.
- stat->on_disconnect(ctx, srs_success);
+ stat_->on_disconnect(ctx, srs_success);
srs_freep(info);
map_ctx_info_.erase(it++);
@@ -441,7 +451,6 @@ srs_error_t SrsHlsStream::on_timer(srs_utime_t interval)
return err;
}
-// LCOV_EXCL_STOP
bool SrsHlsStream::is_interrupt(std::string id)
{
@@ -454,6 +463,7 @@ bool SrsHlsStream::is_interrupt(std::string id)
SrsVodStream::SrsVodStream(string root_dir) : SrsHttpFileServer(root_dir)
{
+ config_ = _srs_config;
}
void SrsVodStream::assemble()
@@ -463,6 +473,7 @@ void SrsVodStream::assemble()
SrsVodStream::~SrsVodStream()
{
+ config_ = NULL;
}
srs_error_t SrsVodStream::serve_flv_stream(ISrsHttpResponseWriter *w, ISrsHttpMessage *r, string fullpath, int64_t offset)
@@ -562,7 +573,13 @@ srs_error_t SrsVodStream::serve_mp4_stream(ISrsHttpResponseWriter *w, ISrsHttpMe
end = fs->filesize() - 1;
}
- if (end > fs->filesize() || start > end || end < 0) {
+ // The end is the last byte position and it is inclusive, so the last byte a client may ask for is filesize-1.
+ // Clamp an end that reaches or passes the end of the file, as the clients expect the bytes that do exist.
+ if (end >= fs->filesize()) {
+ end = fs->filesize() - 1;
+ }
+
+ if (start > end || end < 0) {
return srs_error_new(ERROR_HTTP_REMUX_OFFSET_OVERFLOW, "http mp4 streaming %s overflow. size=%" PRId64 ", offset=%d",
fullpath.c_str(), fs->filesize(), start);
}
@@ -602,7 +619,7 @@ srs_error_t SrsVodStream::serve_m3u8_ctx(ISrsHttpResponseWriter *w, ISrsHttpMess
SrsUniquePtr req(hr->to_request(hr->host())->as_http());
// discovery vhost, resolve the vhost from config
- SrsConfDirective *parsed_vhost = _srs_config->get_vhost(req->vhost_);
+ SrsConfDirective *parsed_vhost = config_->get_vhost(req->vhost_);
if (parsed_vhost) {
req->vhost_ = parsed_vhost->arg0();
}
@@ -657,11 +674,15 @@ ISrsHttpStaticServer::~ISrsHttpStaticServer()
SrsHttpStaticServer::SrsHttpStaticServer()
{
mux_ = new SrsHttpServeMux();
+
+ config_ = _srs_config;
}
SrsHttpStaticServer::~SrsHttpStaticServer()
{
srs_freep(mux_);
+
+ config_ = NULL;
}
// LCOV_EXCL_START
@@ -669,6 +690,7 @@ srs_error_t SrsHttpStaticServer::serve_http(ISrsHttpResponseWriter *w, ISrsHttpM
{
return mux_->serve_http(w, r);
}
+// LCOV_EXCL_STOP
srs_error_t SrsHttpStaticServer::initialize()
{
@@ -677,7 +699,7 @@ srs_error_t SrsHttpStaticServer::initialize()
bool default_root_exists = false;
// http static file and flv vod stream mount for each vhost.
- SrsConfDirective *root = _srs_config->get_root();
+ SrsConfDirective *root = config_->get_root();
for (int i = 0; i < (int)root->directives_.size(); i++) {
SrsConfDirective *conf = root->at(i);
@@ -693,14 +715,14 @@ srs_error_t SrsHttpStaticServer::initialize()
if (pmount == "/") {
default_root_exists = true;
- std::string dir = _srs_config->get_vhost_http_dir(vhost);
+ std::string dir = config_->get_vhost_http_dir(vhost);
srs_warn("http: root mount to %s", dir.c_str());
}
}
if (!default_root_exists) {
// add root
- std::string dir = _srs_config->get_http_stream_dir();
+ std::string dir = config_->get_http_stream_dir();
SrsVodStream *stream = new SrsVodStream(dir);
stream->assemble();
if ((err = mux_->handle("/", stream)) != srs_success) {
@@ -711,30 +733,28 @@ srs_error_t SrsHttpStaticServer::initialize()
return err;
}
-// LCOV_EXCL_STOP
ISrsHttpServeMux *SrsHttpStaticServer::mux()
{
return mux_;
}
-// LCOV_EXCL_START
srs_error_t SrsHttpStaticServer::mount_vhost(string vhost, string &pmount)
{
srs_error_t err = srs_success;
// when vhost disabled, ignore.
- if (!_srs_config->get_vhost_enabled(vhost)) {
+ if (!config_->get_vhost_enabled(vhost)) {
return err;
}
// when vhost http_static disabled, ignore.
- if (!_srs_config->get_vhost_http_enabled(vhost)) {
+ if (!config_->get_vhost_http_enabled(vhost)) {
return err;
}
- std::string mount = _srs_config->get_vhost_http_mount(vhost);
- std::string dir = _srs_config->get_vhost_http_dir(vhost);
+ std::string mount = config_->get_vhost_http_mount(vhost);
+ std::string dir = config_->get_vhost_http_dir(vhost);
// replace the vhost variable
mount = srs_strings_replace(mount, "[vhost]", vhost);
@@ -760,4 +780,3 @@ srs_error_t SrsHttpStaticServer::mount_vhost(string vhost, string &pmount)
return err;
}
-// LCOV_EXCL_STOP
diff --git a/trunk/src/app/srs_app_http_static.hpp b/trunk/src/app/srs_app_http_static.hpp
index 30123104aa..b69b4f742e 100644
--- a/trunk/src/app/srs_app_http_static.hpp
+++ b/trunk/src/app/srs_app_http_static.hpp
@@ -90,6 +90,10 @@ class SrsVodStream : public SrsHttpFileServer
SRS_DECLARE_PRIVATE: // clang-format on
SrsHlsStream hls_;
+// clang-format off
+SRS_DECLARE_PRIVATE: // clang-format on
+ ISrsAppConfig *config_;
+
public:
SrsVodStream(std::string root_dir);
virtual ~SrsVodStream();
@@ -132,6 +136,10 @@ class SrsHttpStaticServer : public ISrsHttpStaticServer
SRS_DECLARE_PRIVATE: // clang-format on
ISrsHttpServeMux *mux_;
+// clang-format off
+SRS_DECLARE_PRIVATE: // clang-format on
+ ISrsAppConfig *config_;
+
public:
SrsHttpStaticServer();
virtual ~SrsHttpStaticServer();
diff --git a/trunk/src/app/srs_app_latest_version.hpp b/trunk/src/app/srs_app_latest_version.hpp
index 22b8db06c2..99a356cd30 100644
--- a/trunk/src/app/srs_app_latest_version.hpp
+++ b/trunk/src/app/srs_app_latest_version.hpp
@@ -23,6 +23,8 @@ class ISrsAppFactory;
// Build features string for version query
extern void srs_build_features(std::stringstream &ss);
+// Deprecated: the latest version query is not used anymore, so do not add tests or
+// testability refactors for this class.
class SrsLatestVersion : public ISrsCoroutineHandler
{
// clang-format off
diff --git a/trunk/src/app/srs_app_log.cpp b/trunk/src/app/srs_app_log.cpp
index 22cecc133b..7a10352089 100644
--- a/trunk/src/app/srs_app_log.cpp
+++ b/trunk/src/app/srs_app_log.cpp
@@ -27,6 +27,49 @@
// reserved for the end of log data, it must be strlen(LOG_TAIL)
#define LOG_TAIL_SIZE 1
+ISrsLogWriter::ISrsLogWriter()
+{
+}
+
+ISrsLogWriter::~ISrsLogWriter()
+{
+}
+
+SrsLogWriter::SrsLogWriter()
+{
+}
+
+SrsLogWriter::~SrsLogWriter()
+{
+}
+
+int SrsLogWriter::open_file(const std::string &path)
+{
+ return ::open(path.c_str(),
+ O_RDWR | O_CREAT | O_APPEND,
+ S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH);
+}
+
+void SrsLogWriter::close_file(int fd)
+{
+ ::close(fd);
+}
+
+void SrsLogWriter::write_file(int fd, const char *str_log, int size)
+{
+ ::write(fd, str_log, size);
+}
+
+void SrsLogWriter::write_console(const char *color, const char *str_log, int size)
+{
+ if (!color || !*color) {
+ printf("%.*s", size, str_log);
+ } else {
+ printf("%s%.*s\033[0m", color, size, str_log);
+ }
+ fflush(stdout);
+}
+
SrsFileLog::SrsFileLog()
{
level_ = SrsLogLevelTrace;
@@ -35,43 +78,60 @@ SrsFileLog::SrsFileLog()
fd_ = -1;
log_to_file_tank_ = false;
utc_ = false;
+
+ // The config global does not exist yet when the logger is created, so it is captured by initialize().
+ config_ = NULL;
+
+ writer_ = new SrsLogWriter();
}
SrsFileLog::~SrsFileLog()
{
srs_freepa(log_data_);
- if (fd_ > 0) {
- ::close(fd_);
- fd_ = -1;
+ if (writer_ && fd_ >= 0) {
+ writer_->close_file(fd_);
}
+ fd_ = -1;
- if (_srs_config) {
- _srs_config->unsubscribe(this);
+ if (config_) {
+ config_->unsubscribe(this);
}
+
+ config_ = NULL;
+ srs_freep(writer_);
}
// LCOV_EXCL_START
srs_error_t SrsFileLog::initialize()
{
- if (_srs_config) {
- _srs_config->subscribe(this);
+ // Capture the config here rather than in the constructor: the logger is one of the first objects created, before
+ // the config global exists.
+ config_ = _srs_config;
- log_to_file_tank_ = _srs_config->get_log_tank_file();
- utc_ = _srs_config->get_utc_time();
+ if (config_) {
+ config_->subscribe(this);
- std::string level = _srs_config->get_log_level();
- std::string level_v2 = _srs_config->get_log_level_v2();
+ log_to_file_tank_ = config_->get_log_tank_file();
+ utc_ = config_->get_utc_time();
+
+ std::string level = config_->get_log_level();
+ std::string level_v2 = config_->get_log_level_v2();
level_ = level_v2.empty() ? srs_get_log_level(level) : srs_get_log_level_v2(level_v2);
}
return srs_success;
}
+// LCOV_EXCL_STOP
void SrsFileLog::reopen()
{
- if (fd_ > 0) {
- ::close(fd_);
+ // Clear the descriptor with the close. Every path below may leave without opening a new file, and write_log()
+ // opens one only when the descriptor is negative, so a closed descriptor left here would be written to after the
+ // number has been handed to another socket or file.
+ if (fd_ >= 0) {
+ writer_->close_file(fd_);
+ fd_ = -1;
}
if (!log_to_file_tank_) {
@@ -133,13 +193,12 @@ void SrsFileLog::write_log(int &fd, char *str_log, int size, int level)
// \033[33m : yellow text code in shell
// \033[0m : normal text code
if (level <= SrsLogLevelTrace) {
- printf("%.*s", size, str_log);
+ writer_->write_console("", str_log, size);
} else if (level == SrsLogLevelWarn) {
- printf("\033[33m%.*s\033[0m", size, str_log);
+ writer_->write_console("\033[33m", str_log, size);
} else {
- printf("\033[31m%.*s\033[0m", size, str_log);
+ writer_->write_console("\033[31m", str_log, size);
}
- fflush(stdout);
return;
}
@@ -149,26 +208,24 @@ void SrsFileLog::write_log(int &fd, char *str_log, int size, int level)
open_log_file();
}
- // write log to file.
- if (fd > 0) {
- ::write(fd, str_log, size);
+ // write log to file. A descriptor of 0 is a log file like any other: the process may have been started with its
+ // standard input closed, so open() hands out 0. Only the negative sentinel means there is no file.
+ if (fd >= 0) {
+ writer_->write_file(fd, str_log, size);
}
}
void SrsFileLog::open_log_file()
{
- if (!_srs_config) {
+ if (!config_) {
return;
}
- std::string filename = _srs_config->get_log_file();
+ std::string filename = config_->get_log_file();
if (filename.empty()) {
return;
}
- fd_ = ::open(filename.c_str(),
- O_RDWR | O_CREAT | O_APPEND,
- S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH);
+ fd_ = writer_->open_file(filename);
}
-// LCOV_EXCL_STOP
diff --git a/trunk/src/app/srs_app_log.hpp b/trunk/src/app/srs_app_log.hpp
index 79cf686d6f..b46bcfa810 100644
--- a/trunk/src/app/srs_app_log.hpp
+++ b/trunk/src/app/srs_app_log.hpp
@@ -15,6 +15,41 @@
#include
#include
+class ISrsAppConfig;
+
+// The output operations of the file logger, the log file and the console. A narrow seam over the system calls, so a
+// test can drive the logger and read back what it wrote without opening a file or writing to the terminal.
+class ISrsLogWriter
+{
+public:
+ ISrsLogWriter();
+ virtual ~ISrsLogWriter();
+
+public:
+ // Open path for append, return the descriptor, or a negative value when it cannot be opened.
+ virtual int open_file(const std::string &path) = 0;
+ // Close a descriptor returned by open_file().
+ virtual void close_file(int fd) = 0;
+ // Write size bytes of str_log to fd.
+ virtual void write_file(int fd, const char *str_log, int size) = 0;
+ // Write size bytes of str_log to the console, wrapped in the shell color code when color is not empty.
+ virtual void write_console(const char *color, const char *str_log, int size) = 0;
+};
+
+// Write the log to the real log file and the real console.
+class SrsLogWriter : public ISrsLogWriter
+{
+public:
+ SrsLogWriter();
+ virtual ~SrsLogWriter();
+ // Interface ISrsLogWriter
+public:
+ virtual int open_file(const std::string &path);
+ virtual void close_file(int fd);
+ virtual void write_file(int fd, const char *str_log, int size);
+ virtual void write_console(const char *color, const char *str_log, int size);
+};
+
// Use memory/disk cache and donot flush when write log.
// it's ok to use it without config, which will log to console, and default trace level.
// when you want to use different level, override this classs, set the protected _level.
@@ -35,6 +70,11 @@ class SrsFileLog : public ISrsLog, public ISrsReloadHandler
// Whether use utc time.
bool utc_;
+// clang-format off
+SRS_DECLARE_PRIVATE: // clang-format on
+ ISrsAppConfig *config_;
+ ISrsLogWriter *writer_;
+
public:
SrsFileLog();
virtual ~SrsFileLog();
diff --git a/trunk/src/app/srs_app_rtc_source.cpp b/trunk/src/app/srs_app_rtc_source.cpp
index bd3202a3cb..c14adff460 100644
--- a/trunk/src/app/srs_app_rtc_source.cpp
+++ b/trunk/src/app/srs_app_rtc_source.cpp
@@ -270,6 +270,7 @@ SrsRtcSourceManager::SrsRtcSourceManager()
{
lock_ = srs_mutex_new();
timer_ = new SrsHourGlass("sources", this, 1 * SRS_UTIME_SECONDS);
+ timer_->assemble();
}
SrsRtcSourceManager::~SrsRtcSourceManager()
diff --git a/trunk/src/app/srs_app_rtmp_conn.cpp b/trunk/src/app/srs_app_rtmp_conn.cpp
index 32fee7a175..be2f9bb4f1 100644
--- a/trunk/src/app/srs_app_rtmp_conn.cpp
+++ b/trunk/src/app/srs_app_rtmp_conn.cpp
@@ -207,22 +207,17 @@ const char *SrsRtmpsTransport::transport_type()
SrsRtmpConn::SrsRtmpConn(ISrsRtmpTransport *transport, string cip, int cport)
{
- // Create a identify for this client.
- _srs_context->set_id(_srs_context->generate_id());
-
transport_ = transport;
ip_ = cip;
port_ = cport;
- create_time_ = srsu2ms(srs_time_now_cached());
+ create_time_ = 0;
- trd_ = new SrsSTCoroutine("rtmp", this, _srs_context->get_id());
+ trd_ = NULL;
kbps_ = new SrsNetworkKbps();
- kbps_->set_io(transport_->io(), transport_->io());
delta_ = new SrsNetworkDelta();
- delta_->set_io(transport_->io(), transport_->io());
- rtmp_ = new SrsRtmpServer(transport_->io());
+ rtmp_ = NULL;
refer_ = new SrsRefer();
security_ = new SrsSecurity();
duration_ = 0;
@@ -239,6 +234,7 @@ SrsRtmpConn::SrsRtmpConn(ISrsRtmpTransport *transport, string cip, int cport)
publish_normal_timeout_ = 0;
app_factory_ = _srs_app_factory;
+ context_ = _srs_context;
config_ = _srs_config;
manager_ = _srs_conn_manager;
stream_publish_tokens_ = _srs_stream_publish_tokens;
@@ -254,6 +250,18 @@ SrsRtmpConn::SrsRtmpConn(ISrsRtmpTransport *transport, string cip, int cport)
void SrsRtmpConn::assemble()
{
+ // Create a identify for this client.
+ context_->set_id(context_->generate_id());
+
+ create_time_ = srsu2ms(srs_time_now_cached());
+
+ trd_ = app_factory_->create_coroutine("rtmp", this, context_->get_id());
+
+ kbps_->set_io(transport_->io(), transport_->io());
+ delta_->set_io(transport_->io(), transport_->io());
+
+ rtmp_ = new SrsRtmpServer(transport_->io());
+
config_->subscribe(this);
}
@@ -263,7 +271,9 @@ SrsRtmpConn::~SrsRtmpConn()
config_->unsubscribe(this);
}
- trd_->interrupt();
+ if (trd_) {
+ trd_->interrupt();
+ }
// wakeup the handler which need to notice.
if (wakable_) {
wakable_->wakeup();
@@ -280,6 +290,7 @@ SrsRtmpConn::~SrsRtmpConn()
srs_freep(security_);
app_factory_ = NULL;
+ context_ = NULL;
config_ = NULL;
manager_ = NULL;
stream_publish_tokens_ = NULL;
@@ -591,7 +602,7 @@ srs_error_t SrsRtmpConn::stream_service_cycle()
switch (info_->type_) {
case SrsRtmpConnPlay: {
// We must do stat the client before hooks, because hooks depends on it.
- if ((err = stat_->on_client(_srs_context->get_id().c_str(), req, this, info_->type_)) != srs_success) {
+ if ((err = stat_->on_client(context_->get_id().c_str(), req, this, info_->type_)) != srs_success) {
return srs_error_wrap(err, "rtmp: stat client");
}
@@ -732,7 +743,7 @@ srs_error_t SrsRtmpConn::playing(SrsSharedPtr source)
}
// Use receiving thread to receive packets from peer.
- SrsQueueRecvThread trd(consumer.get(), rtmp_, SRS_PERF_MW_SLEEP, _srs_context->get_id());
+ SrsQueueRecvThread trd(consumer.get(), rtmp_, SRS_PERF_MW_SLEEP, context_->get_id());
if ((err = trd.start()) != srs_success) {
return srs_error_wrap(err, "rtmp: start receive thread");
@@ -873,7 +884,7 @@ srs_error_t SrsRtmpConn::do_playing(SrsSharedPtr source, SrsLiveC
if (count <= 0) {
#ifndef SRS_PERF_QUEUE_COND_WAIT
- srs_usleep(mw_sleep);
+ srs_usleep(mw_sleep_);
#endif
// ignore when nothing got.
continue;
@@ -944,7 +955,7 @@ srs_error_t SrsRtmpConn::authorize_publish()
}
// We must do stat the client before hooks, because hooks depends on it.
- if ((err = stat_->on_client(_srs_context->get_id().c_str(), req, this, info_->type_)) != srs_success) {
+ if ((err = stat_->on_client(context_->get_id().c_str(), req, this, info_->type_)) != srs_success) {
return srs_error_wrap(err, "rtmp: stat client");
}
@@ -977,7 +988,7 @@ srs_error_t SrsRtmpConn::publishing(SrsSharedPtr source)
if ((err = acquire_err) == srs_success) {
// use isolate thread to recv,
// @see: https://github.com/ossrs/srs/issues/237
- SrsPublishRecvThread rtrd(rtmp_, req, transport_->osfd(), 0, this, source, _srs_context->get_id());
+ SrsPublishRecvThread rtrd(rtmp_, req, transport_->osfd(), 0, this, source, context_->get_id());
rtrd.assemble();
err = do_publishing(source, &rtrd);
diff --git a/trunk/src/app/srs_app_rtmp_conn.hpp b/trunk/src/app/srs_app_rtmp_conn.hpp
index f80391b229..094886074b 100644
--- a/trunk/src/app/srs_app_rtmp_conn.hpp
+++ b/trunk/src/app/srs_app_rtmp_conn.hpp
@@ -54,6 +54,7 @@ class ISrsRtspSourceManager;
class ISrsRtmpServer;
class ISrsRtmpTransport;
class ISrsSecurity;
+class ISrsContext;
// The simple rtmp client for SRS.
class SrsSimpleRtmpClient : public SrsBasicRtmpClient
@@ -176,6 +177,7 @@ class SrsRtmpConn : public ISrsConnection, // It's a resource.
// clang-format off
SRS_DECLARE_PRIVATE: // clang-format on
ISrsAppFactory *app_factory_;
+ ISrsContext *context_;
ISrsResourceManager *manager_;
ISrsAppConfig *config_;
ISrsStreamPublishTokenManager *stream_publish_tokens_;
@@ -235,7 +237,7 @@ class SrsRtmpConn : public ISrsConnection, // It's a resource.
public:
SrsRtmpConn(ISrsRtmpTransport *transport, std::string cip, int port);
- void assemble();
+ void assemble(); // Construct object, to avoid call function in constructor.
virtual ~SrsRtmpConn();
// Interface ISrsResource.
public:
diff --git a/trunk/src/app/srs_app_rtmp_source.cpp b/trunk/src/app/srs_app_rtmp_source.cpp
index 25b034fe40..2707a31f94 100644
--- a/trunk/src/app/srs_app_rtmp_source.cpp
+++ b/trunk/src/app/srs_app_rtmp_source.cpp
@@ -1622,7 +1622,9 @@ SrsLiveSourceManager *_srs_sources = NULL;
SrsLiveSourceManager::SrsLiveSourceManager()
{
lock_ = srs_mutex_new();
- timer_ = new SrsHourGlass("sources", this, 1 * SRS_UTIME_SECONDS);
+ SrsHourGlass *timer = new SrsHourGlass("sources", this, 1 * SRS_UTIME_SECONDS);
+ timer->assemble();
+ timer_ = timer;
app_factory_ = _srs_app_factory;
}
diff --git a/trunk/src/app/srs_app_rtsp_conn.cpp b/trunk/src/app/srs_app_rtsp_conn.cpp
index c005c4fa80..d0ed5f3be3 100644
--- a/trunk/src/app/srs_app_rtsp_conn.cpp
+++ b/trunk/src/app/srs_app_rtsp_conn.cpp
@@ -19,7 +19,6 @@ using namespace std;
#include
#endif
#include
-#include
#include
#include
#include
@@ -205,7 +204,7 @@ srs_error_t SrsRtspPlayStream::start()
}
srs_freep(trd_);
- trd_ = new SrsFastCoroutine("rtsp_sender", this, cid_);
+ trd_ = app_factory_->create_coroutine("rtsp_sender", this, cid_);
if ((err = trd_->start()) != srs_success) {
return srs_error_wrap(err, "rtsp_sender");
@@ -382,8 +381,6 @@ ISrsRtspConnection::~ISrsRtspConnection()
SrsRtspConnection::SrsRtspConnection(ISrsResourceManager *cm, ISrsProtocolReadWriter *skt, std::string cip, int port)
{
manager_ = cm;
- cid_ = _srs_context->generate_id();
- _srs_context->set_id(cid_);
// Initialize timeout management fields from SrsRtspConnection2
last_stun_time = 0;
@@ -395,7 +392,7 @@ SrsRtspConnection::SrsRtspConnection(ISrsResourceManager *cm, ISrsProtocolReadWr
ip_ = cip;
port_ = port;
rtsp_ = new SrsRtspStack(skt);
- trd_ = new SrsSTCoroutine("rtsp", this, _srs_context->get_id());
+ trd_ = NULL;
// Initialize merged SrsRtspSession members
skt_ = skt;
@@ -415,16 +412,26 @@ SrsRtspConnection::SrsRtspConnection(ISrsResourceManager *cm, ISrsProtocolReadWr
config_ = _srs_config;
rtsp_sources_ = _srs_rtsp_sources;
hooks_ = _srs_hooks;
+ app_factory_ = _srs_app_factory;
+ context_ = _srs_context;
}
void SrsRtspConnection::assemble()
{
+ // Create a identify for this session.
+ cid_ = context_->generate_id();
+ context_->set_id(cid_);
+
+ trd_ = app_factory_->create_coroutine("rtsp", this, context_->get_id());
+
rtsp_manager_->subscribe(this);
}
SrsRtspConnection::~SrsRtspConnection()
{
- rtsp_manager_->unsubscribe(this);
+ if (rtsp_manager_) {
+ rtsp_manager_->unsubscribe(this);
+ }
srs_freep(request_);
srs_freep(rtsp_);
@@ -457,6 +464,8 @@ SrsRtspConnection::~SrsRtspConnection()
config_ = NULL;
rtsp_sources_ = NULL;
hooks_ = NULL;
+ app_factory_ = NULL;
+ context_ = NULL;
}
// LCOV_EXCL_START
@@ -465,7 +474,10 @@ srs_error_t SrsRtspConnection::do_send_packet(SrsRtpPacket *pkt)
srs_error_t err = srs_success;
uint32_t ssrc = pkt->header_.get_ssrc();
- ISrsStreamWriter *network = networks_[ssrc];
+ // Look up without inserting: this runs for every RTP packet, and operator[] would default-insert
+ // a NULL entry on each miss.
+ std::map::iterator it = networks_.find(ssrc);
+ ISrsStreamWriter *network = (it == networks_.end()) ? NULL : it->second;
if (!network) {
return srs_error_new(ERROR_RTSP_NO_TRACK, "network not found for ssrc: %u", ssrc);
}
@@ -642,7 +654,20 @@ srs_error_t SrsRtspConnection::on_rtsp_request(SrsRtspRequest *req_raw)
std::string local_sdp_escaped = srs_strings_replace(sdp.c_str(), "\r\n", "\\r\\n");
srs_trace("RTSP: DESCRIBE cseq=%ld, session=%s, sdp: %s", req->seq_, session_id_.c_str(), local_sdp_escaped.c_str());
} else if (req->is_setup()) {
- srs_assert(req->transport_);
+ // SETUP carries its parameters in the Transport header, which the parser leaves NULL when the
+ // client omits it, while is_setup() only looks at the method. Refuse such a request here: the
+ // response and do_setup() below both dereference the transport, and asserting on it would end
+ // the process, and every other session with it, on input any client can send.
+ if (!req->transport_) {
+ SrsUniquePtr res(new SrsRtspResponse((int)req->seq_));
+ res->status_ = SRS_CONSTS_RTSP_BadRequest;
+ res->session_ = session_id_;
+ if ((err = rtsp_->send_message(res.get())) != srs_success) {
+ return srs_error_wrap(err, "response setup");
+ }
+ srs_warn("RTSP: SETUP cseq=%ld without transport, session=%s", req->seq_, session_id_.c_str());
+ return err;
+ }
SrsUniquePtr res(new SrsRtspSetupResponse((int)req->seq_));
res->session_ = session_id_;
@@ -712,7 +737,7 @@ void SrsRtspConnection::on_before_dispose(ISrsResource *c)
}
if (session && session == this) {
- _srs_context->set_id(cid_);
+ context_->set_id(cid_);
srs_trace("RTSP: session detach from [%s](%s), disposing=%d", c->get_id().c_str(),
c->desc().c_str(), disposing_);
}
@@ -727,7 +752,7 @@ void SrsRtspConnection::on_disposing(ISrsResource *c)
void SrsRtspConnection::switch_to_context()
{
- _srs_context->set_id(cid_);
+ context_->set_id(cid_);
}
const SrsContextId &SrsRtspConnection::context_id()
@@ -785,6 +810,14 @@ srs_error_t SrsRtspConnection::do_describe(SrsRtspRequest *req, std::string &sdp
local_sdp.control_ = req->uri_;
local_sdp.ice_lite_ = ""; // Disable this line.
+ // A client may DESCRIBE more than once, and each one rebuilds the track set from the source, so
+ // drop the previous one first. The track ids restart at 0 below, so a stale entry would collide
+ // on that id in get_ssrc_by_stream_id() and bind SETUP to an SSRC that is no longer published.
+ for (std::map::iterator it = tracks_.begin(); it != tracks_.end(); ++it) {
+ srs_freep(it->second);
+ }
+ tracks_.clear();
+
uint32_t track_id = 0;
SrsRtcTrackDescription *audio_desc = source_->audio_desc();
if (audio_desc) {
@@ -878,7 +911,11 @@ srs_error_t SrsRtspConnection::do_setup(SrsRtspRequest *req, uint32_t *pssrc)
"UDP transport not supported, only TCP/interleaved mode is supported");
}
+ // A client may re-SETUP a track, for instance to move it to another interleaved channel. The
+ // writer installed by the previous SETUP is owned here, so free it before taking the new one.
SrsRtspTcpNetwork *network = new SrsRtspTcpNetwork(skt_, req->transport_->interleaved_min_);
+ ISrsStreamWriter *&slot = networks_[ssrc];
+ srs_freep(slot);
networks_[ssrc] = network;
*pssrc = ssrc;
diff --git a/trunk/src/app/srs_app_rtsp_conn.hpp b/trunk/src/app/srs_app_rtsp_conn.hpp
index e66f0a2869..20b5348d27 100644
--- a/trunk/src/app/srs_app_rtsp_conn.hpp
+++ b/trunk/src/app/srs_app_rtsp_conn.hpp
@@ -41,6 +41,7 @@ class ISrsStatistic;
class ISrsRtspSourceManager;
class ISrsHttpHooks;
class ISrsAppConfig;
+class ISrsContext;
// The handler for RTSP play stream.
class ISrsRtspPlayStream
@@ -149,6 +150,8 @@ class SrsRtspConnection : public ISrsResource, // It's a resource.
ISrsStatistic *stat_;
ISrsAppConfig *config_;
ISrsHttpHooks *hooks_;
+ ISrsAppFactory *app_factory_;
+ ISrsContext *context_;
// clang-format off
SRS_DECLARE_PRIVATE: // clang-format on
diff --git a/trunk/src/app/srs_app_rtsp_source.cpp b/trunk/src/app/srs_app_rtsp_source.cpp
index 5c530c6294..2dcfa49fcf 100644
--- a/trunk/src/app/srs_app_rtsp_source.cpp
+++ b/trunk/src/app/srs_app_rtsp_source.cpp
@@ -123,6 +123,7 @@ SrsRtspSourceManager::SrsRtspSourceManager()
{
lock_ = srs_mutex_new();
timer_ = new SrsHourGlass("sources", this, 1 * SRS_UTIME_SECONDS);
+ timer_->assemble();
}
SrsRtspSourceManager::~SrsRtspSourceManager()
diff --git a/trunk/src/app/srs_app_server.cpp b/trunk/src/app/srs_app_server.cpp
index 140972f44a..8b20fedb0d 100644
--- a/trunk/src/app/srs_app_server.cpp
+++ b/trunk/src/app/srs_app_server.cpp
@@ -200,6 +200,7 @@ SrsServer::SrsServer()
#endif
http_server_ = new SrsHttpServer();
+ http_server_->assemble();
reuse_api_over_server_ = false;
reuse_rtc_over_server_ = false;
diff --git a/trunk/src/app/srs_app_srt_listener.cpp b/trunk/src/app/srs_app_srt_listener.cpp
index a9299f3310..5c9a0081ba 100644
--- a/trunk/src/app/srs_app_srt_listener.cpp
+++ b/trunk/src/app/srs_app_srt_listener.cpp
@@ -18,6 +18,14 @@ ISrsSrtHandler::~ISrsSrtHandler()
{
}
+ISrsSrtListener::ISrsSrtListener()
+{
+}
+
+ISrsSrtListener::~ISrsSrtListener()
+{
+}
+
SrsSrtListener::SrsSrtListener(ISrsSrtHandler *h, std::string i, int p)
{
handler_ = h;
diff --git a/trunk/src/app/srs_app_srt_listener.hpp b/trunk/src/app/srs_app_srt_listener.hpp
index 5cda7237ae..80f24a7cce 100644
--- a/trunk/src/app/srs_app_srt_listener.hpp
+++ b/trunk/src/app/srs_app_srt_listener.hpp
@@ -25,8 +25,21 @@ class ISrsSrtHandler
virtual srs_error_t on_srt_client(srs_srt_t srt_fd) = 0;
};
+// The SRT listener, which creates the socket, then listens after the options are set.
+class ISrsSrtListener
+{
+public:
+ ISrsSrtListener();
+ virtual ~ISrsSrtListener();
+
+public:
+ virtual srs_srt_t fd() = 0;
+ virtual srs_error_t create_socket() = 0;
+ virtual srs_error_t listen() = 0;
+};
+
// Bind and listen SRT(udp) port, use handler to process the client.
-class SrsSrtListener : public ISrsCoroutineHandler
+class SrsSrtListener : public ISrsSrtListener, public ISrsCoroutineHandler
{
// clang-format off
SRS_DECLARE_PRIVATE: // clang-format on
diff --git a/trunk/src/app/srs_app_srt_server.cpp b/trunk/src/app/srs_app_srt_server.cpp
index fef0faf9fd..a9cb422f0b 100644
--- a/trunk/src/app/srs_app_srt_server.cpp
+++ b/trunk/src/app/srs_app_srt_server.cpp
@@ -9,6 +9,7 @@
using namespace std;
#include
+#include
#include
#include
#include
@@ -27,17 +28,24 @@ ISrsSrtClientHandler::~ISrsSrtClientHandler()
{
}
-// LCOV_EXCL_START
SrsSrtAcceptor::SrsSrtAcceptor(ISrsSrtClientHandler *srt_handler)
{
port_ = 0;
srt_handler_ = srt_handler;
listener_ = NULL;
+
+ config_ = _srs_config;
+ app_factory_ = _srs_app_factory;
+ srt_options_ = new SrsSrtOptions();
}
SrsSrtAcceptor::~SrsSrtAcceptor()
{
srs_freep(listener_);
+ srs_freep(srt_options_);
+
+ config_ = NULL;
+ app_factory_ = NULL;
}
srs_error_t SrsSrtAcceptor::listen(std::string ip, int port)
@@ -48,7 +56,7 @@ srs_error_t SrsSrtAcceptor::listen(std::string ip, int port)
port_ = port;
srs_freep(listener_);
- listener_ = new SrsSrtListener(this, ip_, port_);
+ listener_ = app_factory_->create_srt_listener(this, ip_, port_);
// Create srt socket.
if ((err = listener_->create_socket()) != srs_success) {
@@ -74,62 +82,62 @@ srs_error_t SrsSrtAcceptor::set_srt_opt()
{
srs_error_t err = srs_success;
- if ((err = srs_srt_set_maxbw(listener_->fd(), _srs_config->get_srto_maxbw())) != srs_success) {
- return srs_error_wrap(err, "set opt maxbw=%" PRId64 " failed", _srs_config->get_srto_maxbw());
+ if ((err = srt_options_->set_maxbw(listener_->fd(), config_->get_srto_maxbw())) != srs_success) {
+ return srs_error_wrap(err, "set opt maxbw=%" PRId64 " failed", config_->get_srto_maxbw());
}
- if ((err = srs_srt_set_mss(listener_->fd(), _srs_config->get_srto_mss())) != srs_success) {
- return srs_error_wrap(err, "set opt mss=%d failed", _srs_config->get_srto_mss());
+ if ((err = srt_options_->set_mss(listener_->fd(), config_->get_srto_mss())) != srs_success) {
+ return srs_error_wrap(err, "set opt mss=%d failed", config_->get_srto_mss());
}
- if ((err = srs_srt_set_tsbpdmode(listener_->fd(), _srs_config->get_srto_tsbpdmode())) != srs_success) {
- return srs_error_wrap(err, "set opt tsbpdmode=%d failed", _srs_config->get_srto_tsbpdmode());
+ if ((err = srt_options_->set_tsbpdmode(listener_->fd(), config_->get_srto_tsbpdmode())) != srs_success) {
+ return srs_error_wrap(err, "set opt tsbpdmode=%d failed", config_->get_srto_tsbpdmode());
}
- if ((err = srs_srt_set_latency(listener_->fd(), _srs_config->get_srto_latency())) != srs_success) {
- return srs_error_wrap(err, "set opt latency=%d failed", _srs_config->get_srto_latency());
+ if ((err = srt_options_->set_latency(listener_->fd(), config_->get_srto_latency())) != srs_success) {
+ return srs_error_wrap(err, "set opt latency=%d failed", config_->get_srto_latency());
}
- if ((err = srs_srt_set_rcv_latency(listener_->fd(), _srs_config->get_srto_recv_latency())) != srs_success) {
- return srs_error_wrap(err, "set opt recvlatency=%d failed", _srs_config->get_srto_recv_latency());
+ if ((err = srt_options_->set_rcv_latency(listener_->fd(), config_->get_srto_recv_latency())) != srs_success) {
+ return srs_error_wrap(err, "set opt recvlatency=%d failed", config_->get_srto_recv_latency());
}
- if ((err = srs_srt_set_peer_latency(listener_->fd(), _srs_config->get_srto_peer_latency())) != srs_success) {
- return srs_error_wrap(err, "set opt peerlatency=%d failed", _srs_config->get_srto_peer_latency());
+ if ((err = srt_options_->set_peer_latency(listener_->fd(), config_->get_srto_peer_latency())) != srs_success) {
+ return srs_error_wrap(err, "set opt peerlatency=%d failed", config_->get_srto_peer_latency());
}
- if ((err = srs_srt_set_tlpktdrop(listener_->fd(), _srs_config->get_srto_tlpktdrop())) != srs_success) {
- return srs_error_wrap(err, "set opt tlpktdrop=%d failed", _srs_config->get_srto_tlpktdrop());
+ if ((err = srt_options_->set_tlpktdrop(listener_->fd(), config_->get_srto_tlpktdrop())) != srs_success) {
+ return srs_error_wrap(err, "set opt tlpktdrop=%d failed", config_->get_srto_tlpktdrop());
}
- if ((err = srs_srt_set_connect_timeout(listener_->fd(), srsu2msi(_srs_config->get_srto_conntimeout()))) != srs_success) {
- return srs_error_wrap(err, "set opt connect_timeout=%d failed", _srs_config->get_srto_conntimeout());
+ if ((err = srt_options_->set_connect_timeout(listener_->fd(), srsu2msi(config_->get_srto_conntimeout()))) != srs_success) {
+ return srs_error_wrap(err, "set opt connect_timeout=%d failed", config_->get_srto_conntimeout());
}
- if ((err = srs_srt_set_peer_idle_timeout(listener_->fd(), srsu2msi(_srs_config->get_srto_peeridletimeout()))) != srs_success) {
- return srs_error_wrap(err, "set opt peer_idle_timeout=%d failed", _srs_config->get_srto_peeridletimeout());
+ if ((err = srt_options_->set_peer_idle_timeout(listener_->fd(), srsu2msi(config_->get_srto_peeridletimeout()))) != srs_success) {
+ return srs_error_wrap(err, "set opt peer_idle_timeout=%d failed", config_->get_srto_peeridletimeout());
}
- if ((err = srs_srt_set_sndbuf(listener_->fd(), _srs_config->get_srto_sendbuf())) != srs_success) {
- return srs_error_wrap(err, "set opt sendbuf=%d failed", _srs_config->get_srto_sendbuf());
+ if ((err = srt_options_->set_sndbuf(listener_->fd(), config_->get_srto_sendbuf())) != srs_success) {
+ return srs_error_wrap(err, "set opt sendbuf=%d failed", config_->get_srto_sendbuf());
}
- if ((err = srs_srt_set_rcvbuf(listener_->fd(), _srs_config->get_srto_recvbuf())) != srs_success) {
- return srs_error_wrap(err, "set opt recvbuf=%d failed", _srs_config->get_srto_recvbuf());
+ if ((err = srt_options_->set_rcvbuf(listener_->fd(), config_->get_srto_recvbuf())) != srs_success) {
+ return srs_error_wrap(err, "set opt recvbuf=%d failed", config_->get_srto_recvbuf());
}
- if ((err = srs_srt_set_payload_size(listener_->fd(), _srs_config->get_srto_payloadsize())) != srs_success) {
- return srs_error_wrap(err, "set opt payload_size=%d failed", _srs_config->get_srto_payloadsize());
+ if ((err = srt_options_->set_payload_size(listener_->fd(), config_->get_srto_payloadsize())) != srs_success) {
+ return srs_error_wrap(err, "set opt payload_size=%d failed", config_->get_srto_payloadsize());
}
- string passphrase = _srs_config->get_srto_passphrase();
+ string passphrase = config_->get_srto_passphrase();
if (!passphrase.empty()) {
- if ((err = srs_srt_set_passphrase(listener_->fd(), passphrase)) != srs_success) {
+ if ((err = srt_options_->set_passphrase(listener_->fd(), passphrase)) != srs_success) {
return srs_error_wrap(err, "set opt passphrase=%s failed", passphrase.c_str());
}
- int pbkeylen = _srs_config->get_srto_pbkeylen();
- if ((err = srs_srt_set_pbkeylen(listener_->fd(), pbkeylen)) != srs_success) {
+ int pbkeylen = config_->get_srto_pbkeylen();
+ if ((err = srt_options_->set_pbkeylen(listener_->fd(), pbkeylen)) != srs_success) {
return srs_error_wrap(err, "set opt pbkeylen=%d failed", pbkeylen);
}
}
@@ -137,6 +145,7 @@ srs_error_t SrsSrtAcceptor::set_srt_opt()
return err;
}
+// LCOV_EXCL_START
srs_error_t SrsSrtAcceptor::on_srt_client(srs_srt_t srt_fd)
{
srs_error_t err = srs_success;
diff --git a/trunk/src/app/srs_app_srt_server.hpp b/trunk/src/app/srs_app_srt_server.hpp
index 5f26ccd453..3e8691d4b9 100644
--- a/trunk/src/app/srs_app_srt_server.hpp
+++ b/trunk/src/app/srs_app_srt_server.hpp
@@ -15,6 +15,8 @@
class SrsSrtServer;
class SrsHourGlass;
class ISrsSrtClientHandler;
+class ISrsAppConfig;
+class ISrsAppFactory;
// Interface for SRT client acceptance
class ISrsSrtClientHandler
@@ -38,7 +40,13 @@ class SrsSrtAcceptor : public ISrsSrtHandler
// clang-format off
SRS_DECLARE_PRIVATE: // clang-format on
- SrsSrtListener *listener_;
+ ISrsAppConfig *config_;
+ ISrsAppFactory *app_factory_;
+ ISrsSrtOptions *srt_options_;
+
+// clang-format off
+SRS_DECLARE_PRIVATE: // clang-format on
+ ISrsSrtListener *listener_;
public:
SrsSrtAcceptor(ISrsSrtClientHandler *srt_handler);
diff --git a/trunk/src/app/srs_app_srt_source.cpp b/trunk/src/app/srs_app_srt_source.cpp
index 318985ffcf..486591429e 100644
--- a/trunk/src/app/srs_app_srt_source.cpp
+++ b/trunk/src/app/srs_app_srt_source.cpp
@@ -109,6 +109,7 @@ SrsSrtSourceManager::SrsSrtSourceManager()
{
lock_ = srs_mutex_new();
timer_ = new SrsHourGlass("sources", this, 1 * SRS_UTIME_SECONDS);
+ timer_->assemble();
}
SrsSrtSourceManager::~SrsSrtSourceManager()
@@ -327,6 +328,7 @@ SrsSrtFrameBuilder::SrsSrtFrameBuilder(ISrsFrameTarget *target)
audio_streamid_ = 2;
pp_audio_duration_ = new SrsAlonePithyPrint();
+ pp_audio_duration_->assemble();
}
SrsSrtFrameBuilder::~SrsSrtFrameBuilder()
diff --git a/trunk/src/core/srs_core_version8.hpp b/trunk/src/core/srs_core_version8.hpp
index 7ff0071dd0..d0ad688e67 100644
--- a/trunk/src/core/srs_core_version8.hpp
+++ b/trunk/src/core/srs_core_version8.hpp
@@ -9,6 +9,6 @@
#define VERSION_MAJOR 8
#define VERSION_MINOR 0
-#define VERSION_REVISION 41
+#define VERSION_REVISION 44
#endif
diff --git a/trunk/src/kernel/srs_kernel_hourglass.cpp b/trunk/src/kernel/srs_kernel_hourglass.cpp
index 3aa9360b93..d760be41a2 100644
--- a/trunk/src/kernel/srs_kernel_hourglass.cpp
+++ b/trunk/src/kernel/srs_kernel_hourglass.cpp
@@ -52,14 +52,26 @@ SrsHourGlass::SrsHourGlass(string label, ISrsHourGlassHandler *h, srs_utime_t re
handler_ = h;
resolution_ = resolution;
total_elapse_ = 0;
- trd_ = _srs_kernel_factory->create_coroutine("timer-" + label, this, _srs_context->get_id());
- time_ = _srs_kernel_factory->create_time();
+ trd_ = NULL;
+ time_ = NULL;
+
+ factory_ = _srs_kernel_factory;
+ context_ = _srs_context;
}
SrsHourGlass::~SrsHourGlass()
{
srs_freep(trd_);
srs_freep(time_);
+
+ factory_ = NULL;
+ context_ = NULL;
+}
+
+void SrsHourGlass::assemble()
+{
+ trd_ = factory_->create_coroutine("timer-" + label_, this, context_->get_id());
+ time_ = factory_->create_time();
}
srs_error_t SrsHourGlass::start()
@@ -155,14 +167,27 @@ ISrsFastTimer::~ISrsFastTimer()
SrsFastTimer::SrsFastTimer(std::string label, srs_utime_t interval)
{
interval_ = interval;
- trd_ = _srs_kernel_factory->create_coroutine(label, this, _srs_context->get_id());
- time_ = _srs_kernel_factory->create_time();
+ label_ = label;
+ trd_ = NULL;
+ time_ = NULL;
+
+ factory_ = _srs_kernel_factory;
+ context_ = _srs_context;
}
SrsFastTimer::~SrsFastTimer()
{
srs_freep(trd_);
srs_freep(time_);
+
+ factory_ = NULL;
+ context_ = NULL;
+}
+
+void SrsFastTimer::assemble()
+{
+ trd_ = factory_->create_coroutine(label_, this, context_->get_id());
+ time_ = factory_->create_time();
}
srs_error_t SrsFastTimer::start()
@@ -189,6 +214,12 @@ void SrsFastTimer::unsubscribe(ISrsFastTimerHandler *timer)
if (it != handlers_.end()) {
handlers_.erase(it);
}
+
+ // Also take it out of the round in progress, so it is not notified after it left.
+ deque::iterator p = std::find(pending_.begin(), pending_.end(), timer);
+ if (p != pending_.end()) {
+ pending_.erase(p);
+ }
}
srs_error_t SrsFastTimer::cycle()
@@ -202,8 +233,14 @@ srs_error_t SrsFastTimer::cycle()
++_srs_pps_timer->sugar_;
- for (int i = 0; i < (int)handlers_.size(); i++) {
- ISrsFastTimerHandler *timer = handlers_.at(i);
+ // Notify from a queue rather than walking handlers_, because a handler may unsubscribe during
+ // its callback, or from another coroutine while the callback yields, and erasing from the
+ // vector under the walk would skip the handler after it. Popping from the queue does not
+ // depend on positions, so unsubscribe() can remove from it freely.
+ pending_.assign(handlers_.begin(), handlers_.end());
+ while (!pending_.empty()) {
+ ISrsFastTimerHandler *timer = pending_.front();
+ pending_.pop_front();
if ((err = timer->on_timer(interval_)) != srs_success) {
srs_freep(err); // Ignore any error for shared timer.
@@ -296,9 +333,13 @@ srs_error_t SrsSharedTimer::initialize()
// Initialize global shared timers
timer20ms_ = new SrsFastTimer("shared", 20 * SRS_UTIME_MILLISECONDS);
+ timer20ms_->assemble();
timer100ms_ = new SrsFastTimer("shared", 100 * SRS_UTIME_MILLISECONDS);
+ timer100ms_->assemble();
timer1s_ = new SrsFastTimer("shared", 1 * SRS_UTIME_SECONDS);
+ timer1s_->assemble();
timer5s_ = new SrsFastTimer("shared", 5 * SRS_UTIME_SECONDS);
+ timer5s_->assemble();
clock_monitor_ = new SrsClockWallMonitor();
// Start all timers
diff --git a/trunk/src/kernel/srs_kernel_hourglass.hpp b/trunk/src/kernel/srs_kernel_hourglass.hpp
index 5c89f60e2e..6f8c830837 100644
--- a/trunk/src/kernel/srs_kernel_hourglass.hpp
+++ b/trunk/src/kernel/srs_kernel_hourglass.hpp
@@ -11,11 +11,14 @@
#include
+#include
#include