# `./fio-stl/420 pubsub.h` 26 public symbols. ### Macros #### `FIO_PUBSUB_FUTURE_LIMIT_MS` ```c #define FIO_PUBSUB_FUTURE_LIMIT_MS 60000ULL ``` Maximum time in milliseconds to allow "future" messages to be delivered. If the module detects someone publishing a future message, it will refuse to deliver the message to subscribers. Instead, the module will notify the history engines, allowing them to buffer future messages for future delivery. _Symbol type:_ `macro` #### `FIO_PUBSUB_HISTORY_DEFAULT_CACHE_SIZE_LIMIT` ```c #define FIO_PUBSUB_HISTORY_DEFAULT_CACHE_SIZE_LIMIT (1ULL << 28) ``` The default cache limit - 256Mb _Symbol type:_ `macro` ### Types #### `fio_pubsub_subscribe_args_s` ```c typedef struct { /** * The subscription owner - if none, the subscription is owned by the system. * * Note: * * Both the system and the `io` objects each manage channel listing * which allows only a single subscription to the same channel. * * This means a single subscription per channel per IO and a single * subscription per channel for the global system unless managing the * subscription handle manually. */ fio_io_s *io; /** * A named `channel` to which the message was sent. * * Subscriptions require a match by both channel name and namespace filter. */ fio_buf_info_s channel; /** * The callback to be called for each message forwarded to the subscription. */ void (*on_message)(fio_pubsub_msg_s *msg); /** An optional callback for when a subscription is canceled. */ void (*on_unsubscribe)(void *udata); /** The opaque udata value is ignored and made available to the callbacks. */ void *udata; /** The queue to which the callbacks should be routed. May be NULL. */ fio_queue_s *queue; /** * OPTIONAL: subscription handle return value - should be NULL when using * automatic memory management with the IO or global environment. * * When set, the `io` pointer will be ignored and the subscription object * handle will be written to the `subscription_handle_ptr` which MUST be * used when unsubscribing. * * NOTE: this could cause subscriptions and memory leaks unless properly * handled. */ uintptr_t *subscription_handle_ptr; /** Replay cached messages (if any) since supplied time in milliseconds. */ uint64_t replay_since; /** A numerical namespace `filter` subscribers need to match. */ int16_t filter; /** If set, pattern matching will be used (name is a pattern). */ uint8_t is_pattern; /** If set, subscription will be limited to the root / master process. */ uint8_t master_only; } fio_pubsub_subscribe_args_s ``` _Symbol type:_ `type` #### `fio_pubsub_publish_args_s` ```c typedef struct { struct fio_pubsub_engine_s const *engine; fio_io_s *from; uint64_t id; uint64_t timestamp; fio_buf_info_s channel; fio_buf_info_s message; int16_t filter; } fio_pubsub_publish_args_s ``` Publish arguments _Symbol type:_ `type` #### `fio_pubsub_engine_s` ```c struct fio_pubsub_engine_s { /** Called when engine is detached */ void (*detached)(const struct fio_pubsub_engine_s *eng); /** Called when a subscription is created */ void (*subscribe)(const struct fio_pubsub_engine_s *eng, const fio_buf_info_s channel, int16_t filter); /** Called when a pattern subscription is created */ void (*psubscribe)(const struct fio_pubsub_engine_s *eng, const fio_buf_info_s channel, int16_t filter); /** Called when a subscription is removed */ void (*unsubscribe)(const struct fio_pubsub_engine_s *eng, const fio_buf_info_s channel, int16_t filter); /** Called when a pattern subscription is removed */ void (*punsubscribe)(const struct fio_pubsub_engine_s *eng, const fio_buf_info_s channel, int16_t filter); /** Called when a message is published */ void (*publish)(const struct fio_pubsub_engine_s *eng, const fio_pubsub_msg_s *msg); } ``` Engine structure for external pub/sub backends. EXECUTION CONTEXT: - Publish callback can be called from any thread / process. - Subscription callbacks are called from the MASTER process only - Subscription callbacks are called from the main event loop thread - Callbacks MUST NOT block (defer long operations) _Symbol type:_ `type` #### `fio_pubsub_history_s` ```c struct fio_pubsub_history_s { /** Cleanup callback - called when history manager is detached */ void (*detached)(const struct fio_pubsub_history_s *hist); /** * Stores a message in history. * * Returns 0 on success, -1 on error. * Called when a message is published (master only). */ int (*push)(const struct fio_pubsub_history_s *hist, fio_pubsub_msg_s *msg); /** * Replay messages since timestamp by calling callback for each. * * Returns 0 if replay was handled, -1 if this manager cannot replay. * * MUST call the `on_done` callback to handle possible cleanup. */ int (*replay)(const struct fio_pubsub_history_s *hist, fio_buf_info_s channel, int16_t filter, uint64_t since, void (*on_message)(fio_pubsub_msg_s *msg, void *udata), void (*on_done)(void *udata), void *udata); /** * Get oldest available timestamp for a channel. * * Returns UINT64_MAX if no history available. */ uint64_t (*oldest)(const struct fio_pubsub_history_s *hist, fio_buf_info_s channel, int16_t filter); } ``` History storage interface - completely separate from engines. EXECUTION CONTEXT: - All callbacks are called from the MASTER process only - Callbacks are called from the main event loop thread - Callbacks MUST NOT block (defer long operations) - Callbacks SHOULD NOT be called directly by the user _Symbol type:_ `type` ### Functions #### `fio_pubsub_subscribe` ```c void fio_pubsub_subscribe(fio_pubsub_subscribe_args_s args) ``` Subscribe to a channel. In worker processes, this uses IPC to notify the master. _Symbol type:_ `function` #### `fio_pubsub_subscribe` ```c #define fio_pubsub_subscribe(...) \ fio_pubsub_subscribe((fio_pubsub_subscribe_args_s){__VA_ARGS__}) ``` _Note:_ this may be a macro only / macro wrapper for a function. _Symbol type:_ `macro` #### `fio_pubsub_unsubscribe` ```c int fio_pubsub_unsubscribe(fio_pubsub_subscribe_args_s args) ``` Unsubscribe from a channel. Returns 0 on success, -1 if subscription not found. _Symbol type:_ `function` #### `fio_pubsub_unsubscribe` ```c #define fio_pubsub_unsubscribe(...) \ fio_pubsub_unsubscribe((fio_pubsub_subscribe_args_s){__VA_ARGS__}) ``` _Note:_ this may be a macro only / macro wrapper for a function. _Symbol type:_ `macro` #### `fio_pubsub_publish` ```c void fio_pubsub_publish(fio_pubsub_publish_args_s args) ``` Publish a message to a channel. In worker processes, this uses IPC to notify the master. _Symbol type:_ `function` #### `fio_pubsub_publish` ```c #define fio_pubsub_publish(...) \ fio_pubsub_publish((fio_pubsub_publish_args_s){__VA_ARGS__}) ``` _Note:_ this may be a macro only / macro wrapper for a function. _Symbol type:_ `macro` #### `fio_pubsub_defer` ```c void fio_pubsub_defer(fio_pubsub_msg_s *msg) ``` Pushes execution of the on_message callback to the end of the queue. _Symbol type:_ `function` #### `fio_pubsub_match_fn_set` ```c void fio_pubsub_match_fn_set(uint8_t (*match_cb)(fio_str_info_s, fio_str_info_s)) ``` _Symbol type:_ `function` #### `fio_pubsub_msg2ipc` ```c fio_ipc_s *fio_pubsub_msg2ipc(fio_pubsub_msg_s *msg) ``` Returns the underlying IPC message buffer carrying the message data. This allows message deferral (use fio_ipc_dup) and tighter control over the message's lifetime. _Symbol type:_ `function` #### `fio_pubsub_ipc2msg` ```c fio_pubsub_msg_s fio_pubsub_ipc2msg(fio_ipc_s *ipc) ``` Extract pub/sub message from IPC message _Symbol type:_ `function` #### `fio_pubsub_engine_attach` ```c void fio_pubsub_engine_attach(fio_pubsub_engine_s *engine) ``` Attach an engine to the pub/sub system - moves ownership to the system. _Symbol type:_ `function` #### `fio_pubsub_engine_detach` ```c void fio_pubsub_engine_detach(fio_pubsub_engine_s *engine) ``` Detach an engine from the pub/sub system - frees system's reference. _Symbol type:_ `function` #### `fio_pubsub_engine_ipc` ```c fio_pubsub_engine_s const *fio_pubsub_engine_ipc(void) ``` Returns the builtin engine for publishing to the process group (IPC). _Symbol type:_ `function` #### `fio_pubsub_engine_cluster` ```c fio_pubsub_engine_s const *fio_pubsub_engine_cluster(void) ``` Returns the builtin engine for multi-machine cluster publishing (RPC). _Symbol type:_ `function` #### `fio_pubsub_engine_default` ```c fio_pubsub_engine_s const *fio_pubsub_engine_default(void) ``` Returns the current default engine associated with the pub/sub system. _Symbol type:_ `function` #### `fio_pubsub_engine_default_set` ```c fio_pubsub_engine_s const *fio_pubsub_engine_default_set( fio_pubsub_engine_s const *engine) ``` Sets the current default engine associated with the pub/sub system. _Symbol type:_ `function` #### `fio_pubsub_history_attach` ```c int fio_pubsub_history_attach(const fio_pubsub_history_s *manager, uint8_t priority) ``` Attach a history manager with the given priority. Multiple history managers can be attached. All managers receive push() calls when messages are published. For replay(), managers are tried in priority order (highest first) until one can handle the request. _Symbol type:_ `function` #### `fio_pubsub_history_detach` ```c void fio_pubsub_history_detach(const fio_pubsub_history_s *manager) ``` Detach a history manager. _Symbol type:_ `function` #### `fio_pubsub_history_push_all` ```c void fio_pubsub_history_push_all(fio_pubsub_msg_s *msg) ``` Pushes a pub/sub message to all history containers. Use ONLY by an engine, if the message needs to be saved to the history managers in the process but is never delivered locally. _Symbol type:_ `function` #### `fio_pubsub_history_cache` ```c fio_pubsub_history_s const *fio_pubsub_history_cache(size_t size_limit) ``` Get the built-in in-memory history manager and set it's byte-size limit. A zero value size_limit will be replaced with the following default: - Environment value of WEBSITE_MEMORY_LIMIT_MB * 1024 * 1024 - Environment value of WEBSITE_MEMORY_LIMIT_KB - Environment value of WEBSITE_MEMORY_LIMIT - FIO_PUBSUB_HISTORY_DEFAULT_CACHE_SIZE_LIMIT _Symbol type:_ `function` -----------------------------------------------------