mirror/zoneminder
1
0 Fork 0
mirror of https://github.com/ZoneMinder/zoneminder.git synced 2026-10-02 07:25:02 -04:00
Code Issues Packages Projects Releases Wiki Activity
Files
f69700eb6aad28e9f2de3a2fc1096c5896235308
zoneminder/tests/zm_stream_socket.cpp
T
SteveGilvarryandClaude Fable 5.1 f69700eb6a feat: serve monitor lifecycle events on the stream socket 
Add SendMonitorEvent() to broadcast an EVENT frame to every connected
consumer, framed with a per-monitor event sequence (independent of the
media sequences, so it is not reset by a media generation bump) and the
current media generation for correlation. Events are control messages and
are never dropped from a client queue; the sequence still advances when no
consumer is connected, so a late joiner sees the loss as a gap.

Add SetSnapshotEvent() to cache the current-status snapshot replayed to
each new consumer on connect, the events analogue of the cached keyframe.
AcceptClient now enqueues HELLO(s), the snapshot, then the keyframe.

Tests: a broadcast EVENT round-trips with the right type/stream/sequence;
the event sequence advances across a clientless gap; the snapshot is
replayed after HELLO on connect. Ran ./tests/tests '[stream_socket]':
130 assertions in 11 cases pass.

refs #2875

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-16 05:02:24 +10:00

484 lines
15 KiB
C++
Raw Blame History

/*
* This file is part of the ZoneMinder Project. See AUTHORS file for Copyright information
*
* This program is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License as published by the
* Free Software Foundation; either version 2 of the License, or (at your
* option) any later version.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
* more details.
*
* You should have received a copy of the GNU General Public License along
* with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "zm_catch2.h"
#include "zm_comms.h"
#include "zm_stream_socket.h"
#include "zm_stream_socket_protocol.h"
#include <chrono>
#include <cstring>
#include <memory>
#include <sys/stat.h>
#include <sys/time.h>
#include <thread>
#include <unistd.h>
using namespace zm::stream_socket;
namespace {
constexpr char kSockPath[] = "/tmp/zm.stream_socket.unittest.sock";
struct AVCodecParametersDeleter {
void operator()(AVCodecParameters *par) const { avcodec_parameters_free(&par); }
};
using codec_parameters_ptr = std::unique_ptr<AVCodecParameters, AVCodecParametersDeleter>;
struct ReceivedMessage {
Header header;
std::vector<uint8_t> payload;
};
// Blocking exact-size reads against the non-blocking-server socket
class TestClient {
public:
bool Connect() {
return sock_.connect(kSockPath);
}
// Returns false on timeout/close instead of blocking forever
bool ReadMessage(ReceivedMessage &message,
std::chrono::milliseconds timeout = std::chrono::seconds(5)) {
uint8_t header_bytes[kHeaderSize];
if (!ReadExact(header_bytes, kHeaderSize, timeout))
return false;
if (!ParseHeader(header_bytes, message.header))
return false;
message.payload.resize(message.header.payload_size());
if (message.payload.empty())
return true;
return ReadExact(message.payload.data(), message.payload.size(), timeout);
}
zm::TcpUnixClient sock_;
private:
bool ReadExact(uint8_t *out, size_t len, std::chrono::milliseconds timeout) {
timeval tv = {};
tv.tv_sec = timeout.count() / 1000;
tv.tv_usec = (timeout.count() % 1000) * 1000;
setsockopt(sock_.getDesc(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
size_t done = 0;
while (done < len) {
ssize_t bytes = ::recv(sock_.getDesc(), out + done, len - done, 0);
if (bytes <= 0)
return false;
done += bytes;
}
return true;
}
};
av_packet_ptr make_packet(size_t size, uint8_t fill) {
av_packet_ptr packet{av_packet_alloc()};
REQUIRE(av_new_packet(packet.get(), size) == 0);
memset(packet->data, fill, size);
return packet;
}
codec_parameters_ptr make_h264_parameters() {
codec_parameters_ptr par{avcodec_parameters_alloc()};
par->codec_type = AVMEDIA_TYPE_VIDEO;
par->codec_id = AV_CODEC_ID_H264;
par->width = 640;
par->height = 480;
return par;
}
} // namespace
TEST_CASE("StreamSocket lifecycle", "[stream_socket]") {
StreamSocket server(1, kSockPath);
REQUIRE_FALSE(server.IsRunning());
REQUIRE(server.Start());
REQUIRE(server.IsRunning());
SECTION("socket file exists with 0660 permissions") {
struct stat st = {};
REQUIRE(stat(kSockPath, &st) == 0);
REQUIRE(S_ISSOCK(st.st_mode));
REQUIRE((st.st_mode & 0777) == 0660);
}
SECTION("client can connect") {
TestClient client;
REQUIRE(client.Connect());
}
server.Stop();
REQUIRE_FALSE(server.IsRunning());
// socket file removed on Stop
struct stat st = {};
REQUIRE(stat(kSockPath, &st) != 0);
}
TEST_CASE("StreamSocket sends HELLO first, then media", "[stream_socket]") {
StreamSocket server(1, kSockPath);
REQUIRE(server.Start());
codec_parameters_ptr par = make_h264_parameters();
server.SetVideoParams(par.get(), {30, 1});
TestClient client;
REQUIRE(client.Connect());
ReceivedMessage hello;
REQUIRE(client.ReadMessage(hello));
REQUIRE(hello.header.type == static_cast<uint8_t>(MessageType::Hello));
REQUIRE(hello.header.stream == static_cast<uint8_t>(StreamId::Video));
HelloInfo info;
REQUIRE(ParseHello(hello.payload.data(), hello.payload.size(), info));
REQUIRE(info.codec_id == AV_CODEC_ID_H264);
REQUIRE(info.width == 640);
REQUIRE(info.height == 480);
REQUIRE(info.fps_num == 30);
// wait for the server to register the client before sending
std::this_thread::sleep_for(std::chrono::milliseconds(50));
av_packet_ptr packet = make_packet(1000, 0xAB);
server.SendMedia(packet.get(), StreamId::Video, true, 123456);
ReceivedMessage media;
REQUIRE(client.ReadMessage(media));
REQUIRE(media.header.type == static_cast<uint8_t>(MessageType::Media));
REQUIRE(media.header.flags == kFlagKeyframe);
REQUIRE(media.header.pts_us == 123456);
REQUIRE(media.payload.size() == 1000);
REQUIRE(media.payload[0] == 0xAB);
REQUIRE(media.payload[999] == 0xAB);
server.Stop();
}
TEST_CASE("StreamSocket late joiner receives cached keyframe", "[stream_socket]") {
StreamSocket server(1, kSockPath);
REQUIRE(server.Start());
codec_parameters_ptr par = make_h264_parameters();
server.SetVideoParams(par.get(), {0, 0});
av_packet_ptr keyframe = make_packet(2000, 0x5A);
server.SendMedia(keyframe.get(), StreamId::Video, true, 1000);
av_packet_ptr delta = make_packet(100, 0x11);
server.SendMedia(delta.get(), StreamId::Video, false, 2000);
TestClient client;
REQUIRE(client.Connect());
ReceivedMessage hello;
REQUIRE(client.ReadMessage(hello));
REQUIRE(hello.header.type == static_cast<uint8_t>(MessageType::Hello));
ReceivedMessage cached;
REQUIRE(client.ReadMessage(cached));
REQUIRE(cached.header.type == static_cast<uint8_t>(MessageType::Keyframe));
REQUIRE(cached.header.pts_us == 1000);
REQUIRE(cached.payload.size() == 2000);
REQUIRE(cached.payload[0] == 0x5A);
server.Stop();
}
TEST_CASE("StreamSocket queue overflow drops media but never HELLO", "[stream_socket]") {
StreamSocket::Config config;
config.queue_max_bytes = 64 * 1024;
config.queue_max_msgs = 16;
config.stats_interval = std::chrono::milliseconds(200);
config.stall_timeout = std::chrono::seconds(60); // not under test here
StreamSocket server(1, kSockPath, config);
REQUIRE(server.Start());
codec_parameters_ptr par = make_h264_parameters();
server.SetVideoParams(par.get(), {0, 0});
TestClient client;
REQUIRE(client.Connect());
std::this_thread::sleep_for(std::chrono::milliseconds(100));
// Push much more than the queue + kernel buffers can hold while the client
// is not reading.
constexpr size_t kPacketSize = 16 * 1024;
constexpr size_t kPackets = 200;
av_packet_ptr packet = make_packet(kPacketSize, 0x42);
for (size_t i = 0; i < kPackets; ++i) {
server.SendMedia(packet.get(), StreamId::Video, false, i);
}
// Now drain everything; track sequence gaps and stats
uint64_t received_media = 0;
uint64_t reported_dropped = 0;
bool got_stats_with_drops = false;
uint32_t last_sequence = 0;
bool first_media = true;
uint64_t gap_total = 0;
// The server emits STATS continuously, so bound the drain by a streak of
// non-media messages rather than a read timeout.
int non_media_streak = 0;
ReceivedMessage message;
while (non_media_streak < 5
and client.ReadMessage(message, std::chrono::milliseconds(1000))) {
++non_media_streak;
if (message.header.type == static_cast<uint8_t>(MessageType::Hello))
continue;
if (message.header.type == static_cast<uint8_t>(MessageType::Stats)) {
uint64_t sent = 0;
REQUIRE(ParseStats(message.payload.data(), message.payload.size(), sent, reported_dropped));
if (reported_dropped > 0)
got_stats_with_drops = true;
continue;
}
if (message.header.type == static_cast<uint8_t>(MessageType::Media)) {
non_media_streak = 0;
if (!first_media) {
REQUIRE(message.header.sequence > last_sequence); // monotonic
gap_total += message.header.sequence - last_sequence - 1;
} else {
gap_total += message.header.sequence; // drops before the first received
first_media = false;
}
last_sequence = message.header.sequence;
++received_media;
}
}
// The client never saw some packets, and the loss is observable both ways
REQUIRE(received_media < kPackets);
REQUIRE(gap_total > 0);
REQUIRE(got_stats_with_drops);
REQUIRE(gap_total + received_media == kPackets);
server.Stop();
}
TEST_CASE("StreamSocket stalled client does not affect a live one", "[stream_socket]") {
StreamSocket::Config config;
config.queue_max_bytes = 32 * 1024;
config.queue_max_msgs = 8;
StreamSocket server(1, kSockPath, config);
REQUIRE(server.Start());
codec_parameters_ptr par = make_h264_parameters();
server.SetVideoParams(par.get(), {0, 0});
TestClient stalled;
REQUIRE(stalled.Connect());
TestClient live;
REQUIRE(live.Connect());
std::this_thread::sleep_for(std::chrono::milliseconds(100));
ReceivedMessage message;
REQUIRE(live.ReadMessage(message)); // HELLO
REQUIRE(message.header.type == static_cast<uint8_t>(MessageType::Hello));
constexpr size_t kPackets = 50;
av_packet_ptr packet = make_packet(8 * 1024, 0x99);
uint64_t live_received = 0;
bool live_gap = false;
uint32_t expected_sequence = 0;
for (size_t i = 0; i < kPackets; ++i) {
server.SendMedia(packet.get(), StreamId::Video, false, i);
// live client reads continuously; stalled never reads
if (live.ReadMessage(message)) {
if (message.header.type == static_cast<uint8_t>(MessageType::Media)) {
if (message.header.sequence != expected_sequence)
live_gap = true;
expected_sequence = message.header.sequence + 1;
++live_received;
}
}
}
REQUIRE(live_received == kPackets);
REQUIRE_FALSE(live_gap);
server.Stop();
}
TEST_CASE("StreamSocket parameter change bumps generation and resends HELLO", "[stream_socket]") {
StreamSocket server(1, kSockPath);
REQUIRE(server.Start());
codec_parameters_ptr par = make_h264_parameters();
server.SetVideoParams(par.get(), {0, 0});
TestClient client;
REQUIRE(client.Connect());
ReceivedMessage hello;
REQUIRE(client.ReadMessage(hello));
REQUIRE(hello.header.generation == 0);
std::this_thread::sleep_for(std::chrono::milliseconds(50));
// Same parameters: no new HELLO, no generation bump
server.SetVideoParams(par.get(), {0, 0});
// Changed parameters: generation bump + new HELLO
par->width = 1920;
par->height = 1080;
server.SetVideoParams(par.get(), {0, 0});
ReceivedMessage hello2;
REQUIRE(client.ReadMessage(hello2));
REQUIRE(hello2.header.type == static_cast<uint8_t>(MessageType::Hello));
REQUIRE(hello2.header.generation == 1);
HelloInfo info;
REQUIRE(ParseHello(hello2.payload.data(), hello2.payload.size(), info));
REQUIRE(info.width == 1920);
// Media now carries the new generation, sequence restarted
std::this_thread::sleep_for(std::chrono::milliseconds(50));
av_packet_ptr packet = make_packet(100, 0x77);
server.SendMedia(packet.get(), StreamId::Video, false, 5000);
ReceivedMessage media;
REQUIRE(client.ReadMessage(media));
REQUIRE(media.header.type == static_cast<uint8_t>(MessageType::Media));
REQUIRE(media.header.generation == 1);
REQUIRE(media.header.sequence == 0);
server.Stop();
}
TEST_CASE("StreamSocket sends BYE on stop", "[stream_socket]") {
StreamSocket server(1, kSockPath);
REQUIRE(server.Start());
TestClient client;
REQUIRE(client.Connect());
std::this_thread::sleep_for(std::chrono::milliseconds(100));
server.Stop();
ReceivedMessage message;
REQUIRE(client.ReadMessage(message));
REQUIRE(message.header.type == static_cast<uint8_t>(MessageType::Bye));
}
namespace {
std::vector<uint8_t> make_state_event(uint16_t code, uint32_t state_id,
uint32_t prev_state_id, const char *name) {
MonitorEvent ev;
ev.code = code;
ev.state_id = state_id; ev.has_state_id = true;
ev.prev_state_id = prev_state_id; ev.has_prev_state_id = true;
ev.state_name = name;
ev.wall_clock_us = 1718355103501000ULL;
ev.has_wall_clock = true;
return BuildEvent(ev);
}
} // namespace
TEST_CASE("StreamSocket broadcasts a monitor EVENT", "[stream_socket]") {
StreamSocket server(1, kSockPath);
REQUIRE(server.Start());
TestClient client;
REQUIRE(client.Connect());
std::this_thread::sleep_for(std::chrono::milliseconds(50));
server.SendMonitorEvent(make_state_event(kEventStateChanged, 2, 0, "Alarm"));
ReceivedMessage message;
REQUIRE(client.ReadMessage(message));
REQUIRE(message.header.type == static_cast<uint8_t>(MessageType::Event));
REQUIRE(message.header.stream == static_cast<uint8_t>(StreamId::Monitor));
REQUIRE(message.header.sequence == 0);
MonitorEvent ev;
REQUIRE(ParseEvent(message.payload.data(), message.payload.size(), ev));
REQUIRE(ev.code == kEventStateChanged);
REQUIRE(ev.state_id == 2);
REQUIRE(ev.prev_state_id == 0);
REQUIRE(ev.state_name == "Alarm");
REQUIRE(ev.has_wall_clock);
server.Stop();
}
TEST_CASE("StreamSocket event sequence advances without clients", "[stream_socket]") {
StreamSocket server(1, kSockPath);
REQUIRE(server.Start());
// Two events produced before any consumer connects; the sequence still
// advances so the gap is observable to a later joiner.
server.SendMonitorEvent(make_state_event(kEventConnectionFailed, 0, 0, ""));
server.SendMonitorEvent(make_state_event(kEventConnectionRestored, 0, 0, ""));
TestClient client;
REQUIRE(client.Connect());
std::this_thread::sleep_for(std::chrono::milliseconds(50));
server.SendMonitorEvent(make_state_event(kEventStateChanged, 1, 0, "PreAlarm"));
ReceivedMessage message;
REQUIRE(client.ReadMessage(message));
REQUIRE(message.header.type == static_cast<uint8_t>(MessageType::Event));
REQUIRE(message.header.sequence == 2); // 0 and 1 dropped before connect
server.Stop();
}
TEST_CASE("StreamSocket replays snapshot on connect", "[stream_socket]") {
StreamSocket server(1, kSockPath);
REQUIRE(server.Start());
codec_parameters_ptr par = make_h264_parameters();
server.SetVideoParams(par.get(), {30, 1});
server.SetSnapshotEvent(make_state_event(kEventSnapshot, 0, 0, "Idle"));
TestClient client;
REQUIRE(client.Connect());
// HELLO first
ReceivedMessage hello;
REQUIRE(client.ReadMessage(hello));
REQUIRE(hello.header.type == static_cast<uint8_t>(MessageType::Hello));
// then the snapshot EVENT
ReceivedMessage snap;
REQUIRE(client.ReadMessage(snap));
REQUIRE(snap.header.type == static_cast<uint8_t>(MessageType::Event));
REQUIRE(snap.header.stream == static_cast<uint8_t>(StreamId::Monitor));
MonitorEvent ev;
REQUIRE(ParseEvent(snap.payload.data(), snap.payload.size(), ev));
REQUIRE(ev.code == kEventSnapshot);
REQUIRE(ev.state_id == 0);
REQUIRE(ev.state_name == "Idle");
server.Stop();
}
TEST_CASE("StreamSocket::ParseAllowedUids", "[stream_socket]") {
REQUIRE(StreamSocket::ParseAllowedUids("").empty());
REQUIRE(StreamSocket::ParseAllowedUids("33") == std::vector<uid_t>{33});
REQUIRE(StreamSocket::ParseAllowedUids("33,1000") == std::vector<uid_t>{33, 1000});
REQUIRE(StreamSocket::ParseAllowedUids(" 33 , 1000 ") == std::vector<uid_t>{33, 1000});
REQUIRE(StreamSocket::ParseAllowedUids("33,,1000") == std::vector<uid_t>{33, 1000});
REQUIRE(StreamSocket::ParseAllowedUids("33,bogus,1000") == std::vector<uid_t>{33, 1000});
}
Reference in new issue View Git Blame Copy Permalink