Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
33 changes: 28 additions & 5 deletions components/StatePublisher/StatePublisher.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -245,6 +245,15 @@ void StatePublisher::loop()
}
}

// Remove subscriptions whose channel has no connection: the client that subscribed has gone, and the next
// connection on the channel (e.g. a websocket in the same slot) mustn't get its publishes - a client
// subscribes again when it connects
_subscriptions.remove_if([](const Subscription& sub) {
if (sub._noConn)
LOG_I(MODULE_PREFIX, "loop removed subscription topic %s channelID %d (no connection)",
sub._pubTopic.c_str(), sub._channelID);
return sub._noConn;
});

#ifdef DEBUG_STATEPUB_OUTPUT_PUBLISH_STATS
uint64_t loopTimeUs = micros() - loopStartUs;
Expand Down Expand Up @@ -497,6 +506,14 @@ RaftRetCode StatePublisher::apiSubscription(const String &reqStr, String& respSt
continue;
}

// A rate of 0 disables the subscription (it isn't kept with a rate of 0, which would publish at
// every check)
if (pubRateHz <= 0)
{
removeSubscription(pubTopic, channelID);
continue;
}

#ifdef DEBUG_API_SUBSCRIPTION
LOG_I(MODULE_PREFIX, "apiSubscription topic %s rateHz %.2f minMs %d trigger %s channelID %d",
pubTopic.c_str(), pubRateHz, minMs, triggerStr.c_str(), channelID);
Expand Down Expand Up @@ -613,6 +630,13 @@ bool StatePublisher::createSubscription(const String& pubTopic, uint32_t channel
return false;
}

// A rate of 0 disables the subscription
if (rateHz <= 0)
{
removeSubscription(pubTopic, channelID);
return true;
}

// Find or create subscription for this topic+channel
Subscription* pSub = findSubscription(pubTopic, channelID);
if (pSub)
Expand Down Expand Up @@ -752,10 +776,9 @@ bool StatePublisher::attemptPublish(Subscription& sub)
#ifdef DEBUG_NO_PUBLISH_IF_CANNOT_ACCEPT_OUTBOUND
LOG_I(MODULE_PREFIX, "attemptPublish cannot accept outbound channelID %d noConn %d", sub._channelID, noConn);
#endif
#ifdef ENABLE_CONNECTION_BACKOFF
// No connection on the channel: the subscription is removed (after the loop)
if (noConn)
handleConnectionLost(sub);
#endif
sub._noConn = true;
return false;
}

Expand All @@ -781,12 +804,12 @@ bool StatePublisher::attemptPublish(Subscription& sub)
#endif
return true;
}
#ifdef ENABLE_CONNECTION_BACKOFF
else if (retc == COMMS_CORE_RET_NO_CONN)
{
handleConnectionLost(sub);
sub._noConn = true;
return false;
}
#ifdef ENABLE_CONNECTION_BACKOFF
else
{
// Other failure
Expand Down
1 change: 1 addition & 0 deletions components/StatePublisher/StatePublisher.h
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,7 @@ class StatePublisher : public RaftSysMod
uint32_t _lastCheckMs = 0; // Last time we checked for a publish trigger
uint32_t _lastPublishMs = 0; // Last time a message was actually published
bool _isPending = false; // Publish is due but not sent yet
bool _noConn = false; // No connection on the channel (removed after the loop)

// Callbacks (copied from PubSource for performance - avoids lookup in loop)
SysMod_publishMsgGenFn _msgGenFn = nullptr;
Expand Down
Loading