#./fio-stl/420 pubsub.h

#define FIO_PUBSUB_FUTURE_LIMIT_MS 60000ULL

#define FIO_PUBSUB_HISTORY_DEFAULT_CACHE_SIZE_LIMIT (1ULL << 28)

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

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

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);
}

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);
}

void fio_pubsub_subscribe(fio_pubsub_subscribe_args_s args)

#define fio_pubsub_subscribe(...)   \
  fio_pubsub_subscribe((fio_pubsub_subscribe_args_s){__VA_ARGS__})

int fio_pubsub_unsubscribe(fio_pubsub_subscribe_args_s args)

#define fio_pubsub_unsubscribe(...)   \
  fio_pubsub_unsubscribe((fio_pubsub_subscribe_args_s){__VA_ARGS__})

void fio_pubsub_publish(fio_pubsub_publish_args_s args)

#define fio_pubsub_publish(...)   \
  fio_pubsub_publish((fio_pubsub_publish_args_s){__VA_ARGS__})

void fio_pubsub_defer(fio_pubsub_msg_s *msg)

void fio_pubsub_match_fn_set(uint8_t (*match_cb)(fio_str_info_s, fio_str_info_s))

fio_ipc_s *fio_pubsub_msg2ipc(fio_pubsub_msg_s *msg)

fio_pubsub_msg_s fio_pubsub_ipc2msg(fio_ipc_s *ipc)

void fio_pubsub_engine_attach(fio_pubsub_engine_s *engine)

void fio_pubsub_engine_detach(fio_pubsub_engine_s *engine)

fio_pubsub_engine_s const *fio_pubsub_engine_ipc(void)

fio_pubsub_engine_s const *fio_pubsub_engine_cluster(void)

fio_pubsub_engine_s const *fio_pubsub_engine_default(void)

fio_pubsub_engine_s const *fio_pubsub_engine_default_set( fio_pubsub_engine_s const *engine)

int fio_pubsub_history_attach(const fio_pubsub_history_s *manager, uint8_t priority)

void fio_pubsub_history_detach(const fio_pubsub_history_s *manager)

void fio_pubsub_history_push_all(fio_pubsub_msg_s *msg)

fio_pubsub_history_s const *fio_pubsub_history_cache(size_t size_limit)