Skip to content

Feat/stream room closure ttl #459

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Uriagat wants to merge 2 commits into
base: refactor-streams
Choose a base branch
from
Open

Feat/stream room closure ttl #459

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
91d4bf4
feat: stream room closure ttl
Uriagat Jun 24, 2026
3261835
test: stream room closure ttl
Uriagat Jun 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions endpoint/stream/stream/include/privmx/endpoint/stream/ServerTypes.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,15 +29,16 @@ JSON_STRUCT_EXT(StreamRoomDataEntry, core::server::ContainerDataEntry, STREAM_RO

#define STREAM_ROOM_INFO_EXTRA_FIELDS(F) \
F(data, std::vector<StreamRoomDataEntry>) \
F(state, std::optional<std::string>)
F(state, std::optional<std::string>) \
F(streamRoomTtl, std::optional<int64_t>)
JSON_STRUCT_EXT(StreamRoomInfo, core::server::ContainerInfoBase, STREAM_ROOM_INFO_EXTRA_FIELDS);

#define SESSION_DESCRIPTION_FIELDS(F) \
F(sdp, std::string) \
F(type, std::string)
JSON_STRUCT(SessionDescription, SESSION_DESCRIPTION_FIELDS);

#define STREAM_ROOM_CREATE_MODEL_EXTRA_FIELDS(F)
#define STREAM_ROOM_CREATE_MODEL_EXTRA_FIELDS(F) F(streamRoomTtl, std::optional<int64_t>)
JSON_STRUCT_EXT(StreamRoomCreateModel, core::server::ContainerCreateModelBase, STREAM_ROOM_CREATE_MODEL_EXTRA_FIELDS);

#define STREAM_ROOM_CREATE_RESULT_FIELDS(F) F(streamRoomId, std::string)
Expand Down
1 change: 1 addition & 0 deletions endpoint/stream/stream/include/privmx/endpoint/stream/StreamApiLowImpl.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@ class StreamApiLowImpl : public privmx::utils::ManualManagedClass<StreamApiLowIm
const core::Buffer& publicMeta,
const core::Buffer& privateMeta,
const std::optional<core::ContainerPolicyWithoutItem>& policies,
const std::optional<int64_t>& streamRoomTtl,
const std::string& type = STREAM_TYPE_FILTER_FLAG
);

Expand Down
3 changes: 2 additions & 1 deletion endpoint/stream/stream/include_pub/privmx/endpoint/stream/StreamApiLow.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,8 @@ class StreamApiLow : public privmx::endpoint::core::ExtendedPointer<StreamApiLow
const std::vector<core::UserWithPubKey>& managers,
const core::Buffer& publicMeta,
const core::Buffer& privateMeta,
const std::optional<core::ContainerPolicyWithoutItem>& policies
const std::optional<core::ContainerPolicyWithoutItem>& policies,
const std::optional<int64_t>& streamRoomTtl = std::nullopt
);

void updateStreamRoom(
Expand Down
1 change: 1 addition & 0 deletions endpoint/stream/stream/include_pub/privmx/endpoint/stream/Types.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ struct StreamRoom {
int64_t statusCode;
int64_t schemaVersion;
std::string state; // "created" | "open" | "closed"
int64_t streamRoomTtl;
};

struct Stream {
Expand Down
5 changes: 3 additions & 2 deletions endpoint/stream/stream/src/StreamApiLow.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,14 +67,15 @@ std::string StreamApiLow::createStreamRoom(
const std::vector<core::UserWithPubKey>& managers,
const core::Buffer& publicMeta,
const core::Buffer& privateMeta,
const std::optional<core::ContainerPolicyWithoutItem>& policies
const std::optional<core::ContainerPolicyWithoutItem>& policies,
const std::optional<int64_t>& streamRoomTtl
) {
auto impl = getImpl();
core::Validator::validateId(contextId, "field:contextId ");
core::Validator::validateClass<std::vector<core::UserWithPubKey>>(users, "field:users ");
core::Validator::validateClass<std::vector<core::UserWithPubKey>>(managers, "field:managers ");
try {
return impl->createStreamRoom(contextId, users, managers, publicMeta, privateMeta, policies);
return impl->createStreamRoom(contextId, users, managers, publicMeta, privateMeta, policies, streamRoomTtl);
} catch (const privmx::utils::PrivmxException& e) {
core::ExceptionConverter::rethrowAsCoreException(e);
throw core::Exception("ExceptionConverter rethrow error");
Expand Down
2 changes: 2 additions & 0 deletions endpoint/stream/stream/src/StreamApiLowImpl.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -576,6 +576,7 @@ std::string StreamApiLowImpl::createStreamRoom(
const core::Buffer& publicMeta,
const core::Buffer& privateMeta,
const std::optional<core::ContainerPolicyWithoutItem>& policies,
const std::optional<int64_t>& streamRoomTtl,
const std::string& type
) {
auto ctx = prepareContainerCreate(contextId, users, managers);
Expand All @@ -595,6 +596,7 @@ std::string StreamApiLowImpl::createStreamRoom(
if (policies.has_value()) {
createStreamRoomModel.policy = privmx::endpoint::core::Factory::createPolicyServerObject(policies.value());
}
createStreamRoomModel.streamRoomTtl = streamRoomTtl;
auto result = _serverApi->streamRoomCreate(createStreamRoomModel);
return result.streamRoomId;
}
Expand Down
1 change: 1 addition & 0 deletions endpoint/stream/stream/src/StreamVarSerializer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,7 @@ Poco::Dynamic::Var VarSerializer::serialize<stream::StreamRoom>(const stream::St
obj->set("statusCode", serialize(val.statusCode));
obj->set("schemaVersion", serialize(val.schemaVersion));
obj->set("state", serialize(val.state));
obj->set("streamRoomTtl", serialize(val.streamRoomTtl));
return obj;
}

Expand Down
3 changes: 2 additions & 1 deletion endpoint/stream/stream/src/encryptors/streamRoom/StreamRoomDataSchemaMapper.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,7 @@ StreamRoom StreamRoomDataSchemaMapper::toLibStreamRoom(
.policy = core::Factory::parsePolicyServerObject(info.policy),
.statusCode = statusCode,
.schemaVersion = schemaVersion,
.state = info.state.value_or("closed")
.state = info.state.value_or("closed"),
.streamRoomTtl = info.streamRoomTtl.value_or(0)
};
}
5 changes: 3 additions & 2 deletions endpoint/stream/stream/src/varinterface/StreamApiLowVarInterface.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,14 +89,15 @@ Poco::Dynamic::Var StreamApiLowVarInterface::buildSubscriptionQuery(const Poco::
}

Poco::Dynamic::Var StreamApiLowVarInterface::createStreamRoom(const Poco::Dynamic::Var& args) {
auto argsArr = core::VarInterfaceUtil::validateAndExtractArray(args, 6);
auto argsArr = core::VarInterfaceUtil::validateAndExtractArray(args, 7);
auto contextId = _deserializer.deserialize<std::string>(argsArr->get(0), "contextId");
auto users = _deserializer.deserializeVector<core::UserWithPubKey>(argsArr->get(1), "users");
auto managers = _deserializer.deserializeVector<core::UserWithPubKey>(argsArr->get(2), "managers");
auto publicMeta = _deserializer.deserialize<core::Buffer>(argsArr->get(3), "publicMeta");
auto privateMeta = _deserializer.deserialize<core::Buffer>(argsArr->get(4), "privateMeta");
auto policies = _deserializer.deserializeOptional<core::ContainerPolicyWithoutItem>(argsArr->get(5), "policies");
auto result = _streamApi.createStreamRoom(contextId, users, managers, publicMeta, privateMeta, policies);
auto streamRoomTtl = _deserializer.deserializeOptional<int64_t>(argsArr->get(6), "streamRoomTtl");
auto result = _streamApi.createStreamRoom(contextId, users, managers, publicMeta, privateMeta, policies, streamRoomTtl);
return _serializer.serialize(result);
}

Expand Down
3 changes: 2 additions & 1 deletion endpoint/stream/webrtc/include/privmx/endpoint/stream/StreamApiImpl.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,8 @@ class StreamApiImpl {
const std::vector<core::UserWithPubKey>& managers,
const core::Buffer& publicMeta,
const core::Buffer& privateMeta,
const std::optional<core::ContainerPolicyWithoutItem>& policies
const std::optional<core::ContainerPolicyWithoutItem>& policies,
const std::optional<int64_t>& streamRoomTtl
);

void updateStreamRoom(
Expand Down
5 changes: 4 additions & 1 deletion endpoint/stream/webrtc/include_pub/privmx/endpoint/stream/StreamApi.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,8 @@ class StreamApi {
* @param publicMeta public (unencrypted) metadata
* @param privateMeta private (encrypted) metadata
* @param policies Stream Room's policies (pass std::nullopt to use defaults)
* @param streamRoomTtl grace period (ms) the room stays open after the last participant leaves before being
* closed; 0 closes it immediately (pass std::nullopt to use the server default)
*
* @return created Stream Room ID
*/
Expand All @@ -67,7 +69,8 @@ class StreamApi {
const std::vector<core::UserWithPubKey>& managers,
const core::Buffer& publicMeta,
const core::Buffer& privateMeta,
const std::optional<core::ContainerPolicyWithoutItem>& policies
const std::optional<core::ContainerPolicyWithoutItem>& policies,
const std::optional<int64_t>& streamRoomTtl = std::nullopt
);

/**
Expand Down
5 changes: 3 additions & 2 deletions endpoint/stream/webrtc/src/StreamApi.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,14 +41,15 @@ std::string StreamApi::createStreamRoom(
const std::vector<core::UserWithPubKey>& managers,
const core::Buffer& publicMeta,
const core::Buffer& privateMeta,
const std::optional<core::ContainerPolicyWithoutItem>& policies
const std::optional<core::ContainerPolicyWithoutItem>& policies,
const std::optional<int64_t>& streamRoomTtl
) {
validateEndpoint();
core::Validator::validateId(contextId, "field:contextId ");
core::Validator::validateClass<std::vector<core::UserWithPubKey>>(users, "field:users ");
core::Validator::validateClass<std::vector<core::UserWithPubKey>>(managers, "field:managers ");
try {
return _impl->createStreamRoom(contextId, users, managers, publicMeta, privateMeta, policies);
return _impl->createStreamRoom(contextId, users, managers, publicMeta, privateMeta, policies, streamRoomTtl);
} catch (const privmx::utils::PrivmxException& e) {
core::ExceptionConverter::rethrowAsCoreException(e);
throw core::Exception("ExceptionConverter rethrow error");
Expand Down
5 changes: 3 additions & 2 deletions endpoint/stream/webrtc/src/StreamApiImpl.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -560,9 +560,10 @@ std::string StreamApiImpl::createStreamRoom(
const std::vector<core::UserWithPubKey>& managers,
const core::Buffer& publicMeta,
const core::Buffer& privateMeta,
const std::optional<core::ContainerPolicyWithoutItem>& policies
const std::optional<core::ContainerPolicyWithoutItem>& policies,
const std::optional<int64_t>& streamRoomTtl
) {
return _api->createStreamRoom(contextId, users, managers, publicMeta, privateMeta, policies);
return _api->createStreamRoom(contextId, users, managers, publicMeta, privateMeta, policies, streamRoomTtl);
}

void StreamApiImpl::updateStreamRoom(
Expand Down
60 changes: 59 additions & 1 deletion test/tests/StreamTest.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -279,7 +279,8 @@ TEST_F(StreamTest, createStreamRoom) {
.updatePolicy="all",
.updaterCanBeRemovedFromManagers="yes",
.ownerCanBeRemovedFromManagers="yes"
}
},
60000
);
});
if(streamRoomId.empty()) {
Expand Down Expand Up @@ -308,6 +309,7 @@ TEST_F(StreamTest, createStreamRoom) {
EXPECT_EQ(streamRoom.policy.updatePolicy, std::optional<std::string>("all"));
EXPECT_EQ(streamRoom.policy.updaterCanBeRemovedFromManagers, std::optional<std::string>("yes"));
EXPECT_EQ(streamRoom.policy.ownerCanBeRemovedFromManagers, std::optional<std::string>("yes"));
EXPECT_EQ(streamRoom.streamRoomTtl, 60000);
// same users and managers
EXPECT_NO_THROW({
streamRoomId = streamApi->createStreamRoom(
Expand Down Expand Up @@ -345,6 +347,62 @@ TEST_F(StreamTest, createStreamRoom) {
if(streamRoom.managers.size() == 1) {
EXPECT_EQ(streamRoom.managers[0], reader->getString("Login.user_1_id"));
}
EXPECT_EQ(streamRoom.streamRoomTtl, 0);
}

TEST_F(StreamTest, streamRoomTtl_allowsRejoinAndRepublishAfterLeave) {
std::string streamRoomId;
EXPECT_NO_THROW({
streamRoomId = streamApi->createStreamRoom(
reader->getString("Context_1.contextId"),
std::vector<core::UserWithPubKey>{core::UserWithPubKey{
.userId=reader->getString("Login.user_1_id"),
.pubKey=reader->getString("Login.user_1_pubKey")
}},
std::vector<core::UserWithPubKey>{core::UserWithPubKey{
.userId=reader->getString("Login.user_1_id"),
.pubKey=reader->getString("Login.user_1_pubKey")
}},
core::Buffer::from("public"),
core::Buffer::from("private"),
std::nullopt,
60000
);
});
ASSERT_FALSE(streamRoomId.empty());

stream::StreamHandle handle;

// first session
EXPECT_NO_THROW({
streamApi->joinStreamRoom(streamRoomId);
});
EXPECT_NO_THROW({
handle = streamApi->createStream(streamRoomId);
streamApi->getImpl()->addFakeVideoTrack(handle);
streamApi->publishStream(handle);
});
std::this_thread::sleep_for(std::chrono::milliseconds(1500));
EXPECT_NO_THROW({
streamApi->removeStream(handle);
});
EXPECT_NO_THROW({
streamApi->leaveStreamRoom(streamRoomId);
});

// second session
EXPECT_NO_THROW({
streamApi->joinStreamRoom(streamRoomId);
});
EXPECT_NO_THROW({
handle = streamApi->createStream(streamRoomId);
streamApi->getImpl()->addFakeVideoTrack(handle);
streamApi->publishStream(handle);
});
std::this_thread::sleep_for(std::chrono::milliseconds(1500));
EXPECT_NO_THROW({
streamApi->leaveStreamRoom(streamRoomId);
});
}

TEST_F(StreamTest, listStreamRooms_incorrect_input_data) {
Expand Down