diff --git a/src/flow/channel/AlgChannelDemux.cc b/src/flow/channel/AlgChannelDemux.cc index 876a22110..d19bcda7e 100644 --- a/src/flow/channel/AlgChannelDemux.cc +++ b/src/flow/channel/AlgChannelDemux.cc @@ -4,6 +4,7 @@ #include +#include "flow/channel/VideoEofPolicy.h" #include "service/detail/ServiceRegistry.h" #include "service/event/IEventNotifier.h" #include "service/system/IConfigReadService.h" @@ -218,36 +219,36 @@ void AlgChannelDemux::HandleStream() { duration_stat_.EndSample(); if (media::ReadFrameStatus::Success != ret) { if (media::ReadFrameStatus::StreamEnd == ret) { - if ((video_read_count_ < video_repeat_count_) || (video_repeat_count_ <= 0)) { - LOG_INFO("{}:{} Stream End Now RepeatCount:{} Max RepeatCount:{}", kTag, channel_id_, - video_read_count_, video_repeat_count_); - is_need_repeat_ = true; - } else { - // Do not report completion for live streams. - if ((!is_have_report_) && (!IsLiveStream())) { - is_have_report_ = true; - NotifyOnComplete(); - } - } - if (IsLiveStream()) { - LOG_INFO("{}:{} Live Stream End Need ReOpen Or Request Url", kTag, channel_id_); - // Live stream has no definitive end — request new URL or reopen. - SetStatusInfo(service::camera::AlgDemuxStatus::AlgDemuxReadFailed); + const bool is_live_stream = IsLiveStream(); + const auto disposition = + flow::DecideVideoEof(is_live_stream, video_read_count_, video_repeat_count_); + if (flow::VideoEofDisposition::Reopen == disposition) { is_need_repeat_ = true; - action_status_ = util::ErrorEnum::DemuxReadStreamFail; - } else { - SetStatusInfo(service::camera::AlgDemuxStatus::AlgDemuxReadEnd); - if (!is_need_repeat_) { - action_status_ = util::ErrorEnum::DemuxStreamClosed; + if (is_live_stream) { + LOG_INFO("{}:{} Live Stream End Need ReOpen Or Request Url", kTag, channel_id_); + SetStatusInfo(service::camera::AlgDemuxStatus::AlgDemuxReadFailed); + action_status_ = util::ErrorEnum::DemuxReadStreamFail; + return; } - // When is_need_repeat_ is true the stream is about to loop — - // keep action_status_ as Success from the last frame read to - // prevent transient "取流无数据" during the loop transition. - // Signal unfinished recording tasks to finalize. + LOG_INFO("{}:{} Stream End Now RepeatCount:{} Max RepeatCount:{}", kTag, channel_id_, + video_read_count_, video_repeat_count_); + // A loop boundary is not terminal. Keep the last Reading state + // and successful action status until OpenStream publishes the + // next Opened/Reading transition. frame_packet->index = -1; recorder_->TaskFrame(frame_packet); + return; + } + + if (!is_have_report_) { + is_have_report_ = true; + NotifyOnComplete(); } + SetStatusInfo(service::camera::AlgDemuxStatus::AlgDemuxReadEnd); + action_status_ = util::ErrorEnum::DemuxStreamClosed; + frame_packet->index = -1; + recorder_->TaskFrame(frame_packet); return; } SetStatusInfo(service::camera::AlgDemuxStatus::AlgDemuxReadFailed); diff --git a/src/flow/channel/AlgChannelDemuxStream.cc b/src/flow/channel/AlgChannelDemuxStream.cc index c64c7830e..2325df608 100644 --- a/src/flow/channel/AlgChannelDemuxStream.cc +++ b/src/flow/channel/AlgChannelDemuxStream.cc @@ -145,7 +145,8 @@ bool AlgChannelDemux::GetAttr(MsgCameraAttr& attr) { attr.channelStatus = ChannelStatus::ChannelStatusResolusionUnSupport; } } - attr.dataStatus = static_cast(status_.status); + attr.dataStatus = static_cast(status_.status); + attr.repeatPending = is_need_repeat_.load(); return true; } diff --git a/src/flow/channel/VideoEofPolicy.h b/src/flow/channel/VideoEofPolicy.h new file mode 100644 index 000000000..121166b78 --- /dev/null +++ b/src/flow/channel/VideoEofPolicy.h @@ -0,0 +1,28 @@ +// Local-video EOF lifecycle policy shared by the demuxer and monitor guard. + +#pragma once + +namespace cosmo::flow { + +enum class VideoEofDisposition { + Reopen, + Complete, +}; + +// video_read_count is the number of successful opens, including the current +// playback. A non-positive repeat count means infinite playback. +constexpr VideoEofDisposition DecideVideoEof(bool is_live_stream, int video_read_count, + int video_repeat_count) { + if (is_live_stream || video_repeat_count <= 0 || video_read_count < video_repeat_count) { + return VideoEofDisposition::Reopen; + } + return VideoEofDisposition::Complete; +} + +// Defense for the camera monitor: a transient ReadEnd can never be terminal +// while the demuxer has already committed to reopening the stream. +constexpr bool IsTerminalOfflineReadEnd(bool is_read_end, bool repeat_pending) { + return is_read_end && !repeat_pending; +} + +} // namespace cosmo::flow diff --git a/src/service/camera/impl/CameraServiceImpl.cc b/src/service/camera/impl/CameraServiceImpl.cc index 6b04d662b..20633b7a5 100644 --- a/src/service/camera/impl/CameraServiceImpl.cc +++ b/src/service/camera/impl/CameraServiceImpl.cc @@ -20,6 +20,7 @@ #include #include "flow/channel/AlgChannel.h" +#include "flow/channel/VideoEofPolicy.h" #include "flow/common/AlgDataRecord.h" #include "flow/common/FlowTaskUtil.h" #include "service/algorithm/IAlgorithmQuery.h" @@ -581,11 +582,17 @@ void CameraServiceImpl::MonitorCameraEntity(const CameraEntityPtr& camera, bool MsgCameraAttr attr; if (ServiceRegistry::Instance().Get().GetChannelAttr(camera->videoChannelId, attr)) { - bool isReadEnd = + const bool isReadEnd = (attr.dataStatus == static_cast(camera::AlgDemuxStatus::AlgDemuxReadEnd)); + const bool isTerminalReadEnd = + flow::IsTerminalOfflineReadEnd(isReadEnd, attr.repeatPending); + if (isReadEnd && attr.repeatPending) { + LOG_ERRO("[{}/{}] Refusing task auto-stop: local video is reopening", + camera->videoChannelId, task->task_id_); + } // Channel has finished reading and no active data remains (queue fully consumed) - if (isReadEnd && !ServiceRegistry::Instance().Get().TaskDataActive( - camera->videoChannelId)) { + if (isTerminalReadEnd && !ServiceRegistry::Instance().Get().TaskDataActive( + camera->videoChannelId)) { LOG_INFO("[{}/{}] Offline video completed, auto-stopping task to release resources", camera->videoChannelId, task->task_id_); ServiceRegistry::Instance().Get().TaskStop(task->task_id_); diff --git a/src/util/dto/CameraMsgTypes.h b/src/util/dto/CameraMsgTypes.h index 5245f820c..5f939e9a1 100644 --- a/src/util/dto/CameraMsgTypes.h +++ b/src/util/dto/CameraMsgTypes.h @@ -31,6 +31,9 @@ struct MsgCameraAttr { float fps{0.0}; ChannelStatus channelStatus{ChannelStatus::ChannelStatusOffline}; // 0: offline, 1: online, 2: auth error int dataStatus{0}; // service::camera::AlgDemuxStatus status + // Runtime-only guard. It is intentionally not serialized into the public + // camera DTO because it only coordinates the demuxer and task monitor. + bool repeatPending{false}; friend void to_json(nlohmann::json& j, const MsgCameraAttr& v); friend void from_json(const nlohmann::json& j, MsgCameraAttr& v); }; diff --git a/test/test_video_eof_policy.cc b/test/test_video_eof_policy.cc new file mode 100644 index 000000000..897ac1854 --- /dev/null +++ b/test/test_video_eof_policy.cc @@ -0,0 +1,33 @@ +#include "catch_amalgamated.hpp" +#include "flow/channel/VideoEofPolicy.h" + +using cosmo::flow::DecideVideoEof; +using cosmo::flow::IsTerminalOfflineReadEnd; +using cosmo::flow::VideoEofDisposition; + +TEST_CASE("Infinite local video EOF always reopens", "[video-eof][repeat]") { + for (int read_count = 1; read_count <= 1000; ++read_count) { + CHECK(DecideVideoEof(false, read_count, 0) == VideoEofDisposition::Reopen); + } +} + +TEST_CASE("Finite local video EOF completes only after the configured playback count", + "[video-eof][repeat]") { + CHECK(DecideVideoEof(false, 1, 1) == VideoEofDisposition::Complete); + + CHECK(DecideVideoEof(false, 1, 3) == VideoEofDisposition::Reopen); + CHECK(DecideVideoEof(false, 2, 3) == VideoEofDisposition::Reopen); + CHECK(DecideVideoEof(false, 3, 3) == VideoEofDisposition::Complete); +} + +TEST_CASE("Live stream EOF remains a reopen condition", "[video-eof][live]") { + CHECK(DecideVideoEof(true, 1, 1) == VideoEofDisposition::Reopen); + CHECK(DecideVideoEof(true, 1000, 3) == VideoEofDisposition::Reopen); +} + +TEST_CASE("Camera monitor rejects terminal interpretation while reopen is pending", "[video-eof][monitor]") { + CHECK_FALSE(IsTerminalOfflineReadEnd(false, false)); + CHECK_FALSE(IsTerminalOfflineReadEnd(false, true)); + CHECK_FALSE(IsTerminalOfflineReadEnd(true, true)); + CHECK(IsTerminalOfflineReadEnd(true, false)); +}