diff --git a/include/asterisk/stasis.h b/include/asterisk/stasis.h index 14ab7d93b24d348b54d4f2db7fe1e5ba218a7301..62ed1ed1a00f3c966b39bcf31adeca9190d044fb 100644 --- a/include/asterisk/stasis.h +++ b/include/asterisk/stasis.h @@ -590,6 +590,20 @@ struct stasis_subscription *stasis_subscribe_pool(struct stasis_topic *topic, struct stasis_subscription *stasis_unsubscribe( struct stasis_subscription *subscription); +/*! + * \brief Set the high and low alert water marks of the stasis subscription. + * \since 13.10.0 + * + * \param subscription Pointer to a stasis subscription + * \param low_water New queue low water mark. (-1 to set as 90% of high_water) + * \param high_water New queue high water mark. + * + * \retval 0 on success. + * \retval -1 on error (water marks not changed). + */ +int stasis_subscription_set_congestion_limits(struct stasis_subscription *subscription, + long low_water, long high_water); + /*! * \brief Block until the last message is processed on a subscription. * diff --git a/include/asterisk/stasis_message_router.h b/include/asterisk/stasis_message_router.h index 89657a5ee5eddd10dfe653cd8c72162af18e1093..50270a788b4c89537d5ce9ad3e26f863ff1408d5 100644 --- a/include/asterisk/stasis_message_router.h +++ b/include/asterisk/stasis_message_router.h @@ -126,6 +126,20 @@ int stasis_message_router_is_done(struct stasis_message_router *router); void stasis_message_router_publish_sync(struct stasis_message_router *router, struct stasis_message *message); +/*! + * \brief Set the high and low alert water marks of the stasis message router. + * \since 13.10.0 + * + * \param router Pointer to a stasis message router + * \param low_water New queue low water mark. (-1 to set as 90% of high_water) + * \param high_water New queue high water mark. + * + * \retval 0 on success. + * \retval -1 on error (water marks not changed). + */ +int stasis_message_router_set_congestion_limits(struct stasis_message_router *router, + long low_water, long high_water); + /*! * \brief Add a route to a message router. * diff --git a/main/cdr.c b/main/cdr.c index 8658710d523d6235396d36b03231754a6cd43b0c..586a10684a35355e7c2bfa388b48b86ce07f5e6d 100644 --- a/main/cdr.c +++ b/main/cdr.c @@ -71,6 +71,7 @@ ASTERISK_REGISTER_FILE() #include "asterisk/stasis_bridges.h" #include "asterisk/stasis_message_router.h" #include "asterisk/astobj2.h" +#include "asterisk/taskprocessor.h" /*** DOCUMENTATION <configInfo name="cdr" language="en_US"> @@ -4219,6 +4220,8 @@ int ast_cdr_engine_init(void) if (!stasis_router) { return -1; } + stasis_message_router_set_congestion_limits(stasis_router, -1, + 10 * AST_TASKPROCESSOR_HIGH_WATER_LEVEL); if (STASIS_MESSAGE_TYPE_INIT(cdr_sync_message_type)) { return -1; diff --git a/main/cel.c b/main/cel.c index a0d0ad7235e69d0a6e72e7df328ede64c1f7ac02..887a9e6a56a9725d5b6ec7af375812e4f5cf1b99 100644 --- a/main/cel.c +++ b/main/cel.c @@ -59,6 +59,7 @@ ASTERISK_REGISTER_FILE() #include "asterisk/parking.h" #include "asterisk/pickup.h" #include "asterisk/core_local.h" +#include "asterisk/taskprocessor.h" /*** DOCUMENTATION <configInfo name="cel" language="en_US"> @@ -1575,6 +1576,8 @@ static int create_routes(void) if (!cel_state_router) { return -1; } + stasis_message_router_set_congestion_limits(cel_state_router, -1, + 6 * AST_TASKPROCESSOR_HIGH_WATER_LEVEL); ret |= stasis_message_router_add(cel_state_router, stasis_cache_update_type(), diff --git a/main/manager.c b/main/manager.c index 94415b7a0d8f7ef870c7139ed6cc55d5bb129c40..029da70f730c7373cc29e6d3644f5e5a0aa356a2 100644 --- a/main/manager.c +++ b/main/manager.c @@ -100,6 +100,7 @@ ASTERISK_REGISTER_FILE() #include "asterisk/rtp_engine.h" #include "asterisk/format_cache.h" #include "asterisk/translate.h" +#include "asterisk/taskprocessor.h" /*** DOCUMENTATION <manager name="Ping" language="en_US"> @@ -8692,6 +8693,8 @@ static int manager_subscriptions_init(void) if (!stasis_router) { return -1; } + stasis_message_router_set_congestion_limits(stasis_router, -1, + 6 * AST_TASKPROCESSOR_HIGH_WATER_LEVEL); res |= stasis_message_router_set_default(stasis_router, manager_default_msg_cb, NULL); diff --git a/main/stasis.c b/main/stasis.c index 9fe3a2aae591b0a71dba2ce5919732e39e3e703b..91ad94e7664c804430df4f300ff1110131b923b5 100644 --- a/main/stasis.c +++ b/main/stasis.c @@ -564,6 +564,18 @@ struct stasis_subscription *stasis_unsubscribe(struct stasis_subscription *sub) return NULL; } +int stasis_subscription_set_congestion_limits(struct stasis_subscription *subscription, + long low_water, long high_water) +{ + int res = -1; + + if (subscription) { + res = ast_taskprocessor_alert_set_levels(subscription->mailbox, + low_water, high_water); + } + return res; +} + void stasis_subscription_join(struct stasis_subscription *subscription) { if (subscription) { diff --git a/main/stasis_message_router.c b/main/stasis_message_router.c index f60180d68bb9ed006eaf0c342fe2e0a8f4c5c575..85034bcf958e5a74ae5137c088edef8cd7168ef2 100644 --- a/main/stasis_message_router.c +++ b/main/stasis_message_router.c @@ -289,6 +289,18 @@ void stasis_message_router_publish_sync(struct stasis_message_router *router, ao2_cleanup(router); } +int stasis_message_router_set_congestion_limits(struct stasis_message_router *router, + long low_water, long high_water) +{ + int res = -1; + + if (router) { + res = stasis_subscription_set_congestion_limits(router->subscription, + low_water, high_water); + } + return res; +} + int stasis_message_router_add(struct stasis_message_router *router, struct stasis_message_type *message_type, stasis_subscription_cb callback, void *data)