holoscan::InMemoryPubSubSession

Beta

Process-global shared state for a group of in-memory pub/sub participants.

InMemoryPubSubSession consolidates the SharedInMemoryPubSubBackend and SharedInMemoryPubSubSession into a single, reusable class. It owns a shared InMemoryDiscovery and InMemoryTransport, and provides per-participant discovery/transport frontends that multiplex callbacks so that each PubSubContext’s init_context() wiring is independent. This was implemented to enable test cases that involve launching multiple GXF contexts from the same process.

#include <holoscan/in_memory_pubsub_session.hpp>

Inherits from: std::enable_shared_from_this< InMemoryPubSubSession > (public)

Constructors

InMemoryPubSubSession

holoscan::InMemoryPubSubSession::InMemoryPubSubSession(
    const std::string &session_id
)

Destructor

~InMemoryPubSubSession

holoscan::InMemoryPubSubSession::~InMemoryPubSubSession()

Methods

create_frontends

Frontends holoscan::InMemoryPubSubSession::create_frontends()

join

void holoscan::InMemoryPubSubSession::join(
    gxf_context_t context
)

Join this session from a specific GXF context.

leave

void holoscan::InMemoryPubSubSession::leave(
    gxf_context_t context
)

Leave this session from a specific GXF context.

is_multi_context

bool holoscan::InMemoryPubSubSession::is_multi_context() const

Returns true if more than one distinct GXF context has joined.

session_id

const std::string & holoscan::InMemoryPubSubSession::session_id() const

ensure_backends

nvidia::gxf::Expected<void> holoscan::InMemoryPubSubSession::ensure_backends()

retain_participant

nvidia::gxf::Expected<void> holoscan::InMemoryPubSubSession::retain_participant(
    const std::shared_ptr<InMemorySessionParticipant> &participant
)

release_participant

nvidia::gxf::Expected<void> holoscan::InMemoryPubSubSession::release_participant(
    const std::shared_ptr<InMemorySessionParticipant> &participant
)

replay_existing_publishers

nvidia::gxf::Expected<void> holoscan::InMemoryPubSubSession::replay_existing_publishers(
    const std::shared_ptr<InMemorySessionParticipant> &participant
) const

replay_existing_subscribers

nvidia::gxf::Expected<void> holoscan::InMemoryPubSubSession::replay_existing_subscribers(
    const std::shared_ptr<InMemorySessionParticipant> &participant
) const

announce_publisher

nvidia::gxf::Expected<void> holoscan::InMemoryPubSubSession::announce_publisher(
    const std::shared_ptr<InMemorySessionParticipant> &participant,
    const nvidia::gxf::PublisherInfo &info
)

announce_subscriber

nvidia::gxf::Expected<void> holoscan::InMemoryPubSubSession::announce_subscriber(
    const std::shared_ptr<InMemorySessionParticipant> &participant,
    const nvidia::gxf::SubscriberInfo &info
)

remove_publisher

nvidia::gxf::Expected<void> holoscan::InMemoryPubSubSession::remove_publisher(
    const std::shared_ptr<InMemorySessionParticipant> &participant,
    const nvidia::gxf::PublisherGid &gid
)

remove_subscriber

nvidia::gxf::Expected<void> holoscan::InMemoryPubSubSession::remove_subscriber(
    const std::shared_ptr<InMemorySessionParticipant> &participant,
    const nvidia::gxf::SubscriberGid &gid
)

query_publishers

nvidia::gxf::Expected<std::vector<nvidia::gxf::PublisherInfo>> holoscan::InMemoryPubSubSession::query_publishers(
    const std::string &topic_name
) const

query_subscribers

nvidia::gxf::Expected<std::vector<nvidia::gxf::SubscriberInfo>> holoscan::InMemoryPubSubSession::query_subscribers(
    const std::string &topic_name
) const

get_all_topics

nvidia::gxf::Expected<std::vector<std::string>> holoscan::InMemoryPubSubSession::get_all_topics() const

wait_for_subscribers

bool holoscan::InMemoryPubSubSession::wait_for_subscribers(
    const std::string &topic_name,
    size_t expected_count,
    std::chrono::milliseconds timeout
) const

send

Const (1)

Const (2)

nvidia::gxf::Expected<void> holoscan::InMemoryPubSubSession::send(
    const nvidia::gxf::Gid &destination_gid,
    const std::vector<uint8_t> &payload,
    const nvidia::gxf::MessageMetadata &metadata
) const

send_queue_size

size_t holoscan::InMemoryPubSubSession::send_queue_size() const

discovery

std::shared_ptr<nvidia::gxf::InMemoryDiscovery> holoscan::InMemoryPubSubSession::discovery() const

transport

std::shared_ptr<nvidia::gxf::InMemoryTransport> holoscan::InMemoryPubSubSession::transport() const

snapshot_participants

std::vector<std::shared_ptr<InMemorySessionParticipant>> holoscan::InMemoryPubSubSession::snapshot_participants()

cleanup_expired_participants_locked

void holoscan::InMemoryPubSubSession::cleanup_expired_participants_locked()

reset_locked

void holoscan::InMemoryPubSubSession::reset_locked()

notify_publisher_discovered

void holoscan::InMemoryPubSubSession::notify_publisher_discovered(
    const nvidia::gxf::PublisherInfo &info
)

notify_subscriber_discovered

void holoscan::InMemoryPubSubSession::notify_subscriber_discovered(
    const nvidia::gxf::SubscriberInfo &info
)

notify_publisher_lost

void holoscan::InMemoryPubSubSession::notify_publisher_lost(
    const nvidia::gxf::PublisherGid &gid
)

notify_subscriber_lost

void holoscan::InMemoryPubSubSession::notify_subscriber_lost(
    const nvidia::gxf::SubscriberGid &gid
)

make_receive_callback

static nvidia::gxf::PubSubTransport::ReceiveCallback holoscan::InMemoryPubSubSession::make_receive_callback(
    const std::shared_ptr<InMemorySessionParticipant> &participant
)

registry

static SessionRegistry & holoscan::InMemoryPubSubSession::registry()

Static methods

get_or_create

static std::shared_ptr<InMemoryPubSubSession> holoscan::InMemoryPubSubSession::get_or_create(
    const std::string &session_id
)

Get or create a session by ID.

Thread-safe. Returns the same shared_ptr for the same session_id within the process.

reset_all_for_testing

static void holoscan::InMemoryPubSubSession::reset_all_for_testing()

Release all sessions and reset shared state for tests.

Must only be called when no session frontends are active.

Member variables

Name	Type	Description
`session_id_`	`std::string`
`mutex_`	`std::mutex`
`subscriber_wait_mutex_`	`std::mutex`
`subscriber_wait_cv_`	`std::condition_variable`
`participants_`	`std::unordered_map< uint64_t, std::weak_ptr< InMemorySessionParticipant > >`
`discovery_`	`std::shared_ptr< nvidia::gxf::InMemoryDiscovery >`
`transport_`	`std::shared_ptr< nvidia::gxf::InMemoryTransport >`
`next_participant_id_`	`uint64_t`
`context_mutex_`	`std::mutex`
`contexts_`	`std::set< gxf_context_t >`

Inner classes

Frontends

struct holoscan::InMemoryPubSubSession::Frontends

Create a paired discovery + transport frontend that share the same participant state.

The discovery and transport frontends must be used together for a single PubSubContext — they share per-participant callback state so that PubSubContext::init_context() wiring is independent per context.

Name	Type	Description
`discovery`	`std::shared_ptr< nvidia::gxf::PubSubDiscovery >`
`transport`	`std::shared_ptr< nvidia::gxf::PubSubTransport >`