rpn: rfbridge operators and mqtt fixes (#2302)

- cache received rfbridge codes in the internal list, allow to operate on it via the rpn operators
- add `<N> <proto> <code> rfb_match`, matching when we receive specified protocol + code string at least N times
- add `<proto> <code> <proto> <code> rfb_sequence`, checking if specified protocol + code pairs happen in sequence
- add `<TIME> <N> <proto> <code> rfb_match_wait` - similar to `rfb_match`, but waiting for at least `TIME` (ms) via oneshot runner
- add `<proto> <code> rfb_info`, pushes code's latest timestamp and it's counter on the stack
- add `<proto> <code> rfb_pop`, which removes the specified protocol + code from the internal cache
- fix MQTT skip setting making RPN variables absent on initial connection
- default to no skip when receiving MQTT
(small issue still stands with us having non-clean MQTT session, broker will persist variable subscriptions even after unsubscribe event)

code/espurna/config/general.h

@@ -1098,12 +1098,8 @@
 #endif
 
 
-#ifndef MQTT_SKIP_RETAINED
-#define MQTT_SKIP_RETAINED 1 // Skip retained messages on connection
-#endif
-
 #ifndef MQTT_SKIP_TIME
-#define MQTT_SKIP_TIME 1000 // Skip messages for 1 second anter connection
+#define MQTT_SKIP_TIME 0  // Skip messages for N ms after connection. Disabled by default
 #endif
 
 #ifndef MQTT_USE_JSON

code/espurna/mqtt.cpp

@@ -70,12 +70,13 @@ Updated secure client support by Niek van der Maas < mail at niekvandermaas dot
 #endif // MQTT_LIBRARY == MQTT_ASYNCMQTTCLIENT
 
 
-bool _mqtt_enabled = MQTT_ENABLED;
-bool _mqtt_use_json = false;
-unsigned long _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MIN;
 unsigned long _mqtt_last_connection = 0;
 AsyncClientState _mqtt_state = AsyncClientState::Disconnected;
-bool _mqtt_retain_skipped = false;
+bool _mqtt_skip_messages = false;
+unsigned long _mqtt_skip_time = MQTT_SKIP_TIME;
+unsigned long _mqtt_reconnect_delay = MQTT_RECONNECT_DELAY_MIN;
+bool _mqtt_enabled = MQTT_ENABLED;
+bool _mqtt_use_json = false;
 bool _mqtt_retain = MQTT_RETAIN;
 int _mqtt_qos = MQTT_QOS;
 int _mqtt_keepalive = MQTT_KEEPALIVE;
@@ -343,6 +344,9 @@ void _mqttConfigure() {
  _mqttApplyTopic(_mqtt_topic_json, MQTT_TOPIC_JSON);
  }
 
+ // Skip messages in a small window right after the connection
+ _mqtt_skip_time = getSetting("mqttSkipTime", MQTT_SKIP_TIME);
+
  // Custom payload strings
  settingsProcessConfig({
  {_mqtt_payload_online, "mqttPayloadOnline", MQTT_STATUS_ONLINE},
@@ -433,6 +437,7 @@ void _mqttInfo() {
  );
  }
  }
+
 }
 
 // -----------------------------------------------------------------------------
@@ -535,7 +540,6 @@ void _mqttOnConnect() {
 
  _mqtt_last_connection = millis();
  _mqtt_state = AsyncClientState::Connected;
- _mqtt_retain_skipped = false;
 
  DEBUG_MSG_P(PSTR("[MQTT] Connected!\n"));
 
@@ -554,7 +558,6 @@ void _mqttOnDisconnect() {
  // Reset reconnection delay
  _mqtt_last_connection = millis();
  _mqtt_state = AsyncClientState::Disconnected;
- _mqtt_retain_skipped = false;
 
  DEBUG_MSG_P(PSTR("[MQTT] Disconnected!\n"));
 
@@ -568,14 +571,12 @@ void _mqttOnDisconnect() {
 // Force-skip everything received in a short window right after connecting to avoid syncronization issues.
 
 bool _mqttMaybeSkipRetained(char* topic) {
- #if MQTT_SKIP_RETAINED
- if (!_mqtt_retain_skipped && (millis() - _mqtt_last_connection < MQTT_SKIP_TIME)) {
- DEBUG_MSG_P(PSTR("[MQTT] Received %s - SKIPPED\n"), topic);
- return true;
- }
- #endif
+ if (_mqtt_skip_messages && (millis() - _mqtt_last_connection < _mqtt_skip_time)) {
+ DEBUG_MSG_P(PSTR("[MQTT] Received %s - SKIPPED\n"), topic);
+ return true;
+ }
 
- _mqtt_retain_skipped = true;
+ _mqtt_skip_messages = false;
  return false;
 }
 
@@ -1006,6 +1007,8 @@ void _mqttConnect() {
 
  _mqtt_state = AsyncClientState::Connecting;
 
+ _mqtt_skip_messages = (_mqtt_skip_time > 0);
+
  #if SECURE_CLIENT != SECURE_CLIENT_NONE
  const bool secure = getSetting("mqttUseSSL", 1 == MQTT_SSL_ENABLED);
  #else

code/espurna/rpnrules.cpp

@@ -16,10 +16,13 @@ Copyright (C) 2019 by Xose Pérez <xose dot perez at gmail dot com>
 #include "relay.h"
 #include "rpc.h"
 #include "sensor.h"
+#include "rfbridge.h"
 #include "terminal.h"
 #include "ws.h"
 
+#include <list>
 #include <vector>
+
 // -----------------------------------------------------------------------------
 // Custom commands
 // -----------------------------------------------------------------------------
@@ -236,6 +239,214 @@ rpn_error _rpnRelayStatus(rpn_context & ctxt, bool force) {
 
 #endif // RELAY_SUPPORT
 
+#if RFB_SUPPORT
+
+struct rpn_rfbridge_code {
+ unsigned char protocol;
+ String raw;
+ size_t count;
+ decltype(millis()) last;
+};
+
+// TODO: in theory, we could do with forward_list. however, this would require a more complicated removal process,
+// as we would no longer know the previous element and would need to track 2 elements at a time
+static std::list<rpn_rfbridge_code> _rfb_codes;
+
+static uint32_t _rfb_code_repeat_window;
+static uint32_t _rfb_code_stale_delay;
+
+static uint32_t _rfb_code_match_window;
+
+struct rpn_rfbridge_match {
+ unsigned char protocol;
+ String raw;
+};
+
+rpn_error _rpnRfbSequence(rpn_context& ctxt) {
+ auto raw_second = rpn_stack_pop(ctxt);
+ auto proto_second = rpn_stack_pop(ctxt);
+
+ auto raw_first = rpn_stack_pop(ctxt);
+ auto proto_first = rpn_stack_pop(ctxt);
+
+ // find 2 codes in the same order and save pointers
+ rpn_rfbridge_match match[2] {
+ {static_cast<unsigned char>(proto_first.toUint()), raw_first.toString()},
+ {static_cast<unsigned char>(proto_second.toUint()), raw_second.toString()}
+ };
+ rpn_rfbridge_code* refs[2] {nullptr, nullptr};
+
+ for (auto& recent : _rfb_codes) {
+ if ((refs[0] != nullptr) && (refs[1] != nullptr)) {
+ break;
+ }
+ for (int index = 0; index < 2; ++index) {
+ if ((refs[index] == nullptr)
+ && (match[index].protocol == recent.protocol)
+ && (match[index].raw == recent.raw)) {
+ refs[index] = &recent;
+ }
+ }
+ }
+
+ if ((refs[0] == nullptr) || (refs[1] == nullptr)) {
+ return rpn_operator_error::CannotContinue;
+ }
+
+ // purge codes to avoid matching again on the next rules run
+ if ((millis() - refs[0]->last) > (millis() - refs[1]->last)) {
+ _rfb_codes.remove_if([&refs](rpn_rfbridge_code& code) {
+ return (refs[0] == &code) || (refs[1] == &code);
+ });
+ return rpn_operator_error::Ok;
+ }
+
+ return rpn_operator_error::CannotContinue;
+}
+
+decltype(_rfb_codes)::iterator _rpnRfbFindCode(unsigned char protocol, const String& match) {
+ return std::find_if(_rfb_codes.begin(), _rfb_codes.end(), [protocol, &match](const rpn_rfbridge_code& code) {
+ return (code.protocol == protocol) && (code.raw == match);
+ });
+}
+
+rpn_error _rpnRfbPop(rpn_context& ctxt) {
+ auto code = rpn_stack_pop(ctxt);
+ auto proto = rpn_stack_pop(ctxt);
+
+ auto result = _rpnRfbFindCode(proto.toUint(), code.toString());
+ if (result == _rfb_codes.end()) {
+ return rpn_operator_error::CannotContinue;
+ }
+
+ _rfb_codes.erase(result);
+ return rpn_operator_error::Ok;
+}
+
+rpn_error _rpnRfbInfo(rpn_context& ctxt) {
+ auto code = rpn_stack_pop(ctxt);
+ auto proto = rpn_stack_pop(ctxt);
+
+ auto result = _rpnRfbFindCode(proto.toUint(), code.toString());
+ if (result == _rfb_codes.end()) {
+ return rpn_operator_error::CannotContinue;
+ }
+
+ rpn_stack_push(ctxt, rpn_value(
+ static_cast<rpn_uint>((*result).count)));
+ rpn_stack_push(ctxt, rpn_value(
+ static_cast<rpn_uint>((*result).last)));
+
+ return rpn_operator_error::Ok;
+}
+
+rpn_error _rpnRfbWaitMatch(rpn_context& ctxt) {
+ auto code = rpn_stack_pop(ctxt);
+ auto proto = rpn_stack_pop(ctxt);
+ auto count = rpn_stack_pop(ctxt);
+ auto time = rpn_stack_pop(ctxt);
+
+ auto result = _rpnRfbFindCode(proto.toUint(), code.toString());
+ if (result == _rfb_codes.end()) {
+ return rpn_operator_error::CannotContinue;
+ }
+
+ if ((*result).count < count.toUint()) {
+ return rpn_operator_error::CannotContinue;
+ }
+
+ // purge code to avoid matching again on the next rules run
+ if (rpn_operator_error::Ok == _rpnRunnerHandler(ctxt, RpnRunner::Policy::OneShot, time.toUint())) {
+ _rfb_codes.erase(result);
+ return rpn_operator_error::Ok;
+ }
+
+ return rpn_operator_error::CannotContinue;
+}
+
+rpn_error _rpnRfbMatcher(rpn_context& ctxt) {
+ auto code = rpn_stack_pop(ctxt);
+ auto proto = rpn_stack_pop(ctxt);
+ auto count = rpn_stack_pop(ctxt);
+
+ auto result = _rpnRfbFindCode(proto.toUint(), code.toString());
+ if (result == _rfb_codes.end()) {
+ return rpn_operator_error::CannotContinue;
+ }
+
+ // only process recent codes, ignore when rule is processing outside of this small window
+ if (millis() - (*result).last >= _rfb_code_match_window) {
+ return rpn_operator_error::CannotContinue;
+ }
+
+ // purge code to avoid matching again on the next rules run
+ if ((*result).count == count.toUint()) {
+ _rfb_codes.erase(result);
+ return rpn_operator_error::Ok;
+ }
+
+ return rpn_operator_error::CannotContinue;
+}
+
+void _rpnBrokerRfbridgeCallback(unsigned char protocol, const char* raw_code) {
+
+ // remove really old codes that we have not seen in a while to avoid memory exhaustion
+ auto ts = millis();
+ auto old = std::remove_if(_rfb_codes.begin(), _rfb_codes.end(), [ts](rpn_rfbridge_code& code) {
+ return (ts - code.last) >= _rfb_code_stale_delay;
+ });
+
+ if (old != _rfb_codes.end()) {
+ _rfb_codes.erase(old, _rfb_codes.end());
+ }
+
+ auto result = _rpnRfbFindCode(protocol, raw_code);
+ if (result != _rfb_codes.end()) {
+ // we also need to reset the counter at a certain point to allow next batch of repeats to go through
+ if (millis() - (*result).last >= _rfb_code_repeat_window) {
+ (*result).count = 0;
+ }
+ (*result).last = millis();
+ (*result).count += 1u;
+ } else {
+ _rfb_codes.push_back({protocol, raw_code, 1u, millis()});
+ }
+
+ _rpn_run = true;
+}
+
+void _rpnRfbSetup() {
+ // - Repeat window is an arbitrary time, just about 3-4 more times it takes for
+ // a code to be sent again when holding a generic remote button
+ // Code counter is reset to 0 when outside of the window.
+ // - Stale delay allows broker callback to remove really old codes.
+ // (TODO: can this happen in loop() cb instead?)
+ _rfb_code_repeat_window = getSetting("rfbRepeatWindow", 2000ul);
+ _rfb_code_match_window = getSetting("rfbMatchWindow", 2000ul);
+ _rfb_code_stale_delay = getSetting("rfbStaleDelay", 10000ul);
+
+#if TERMINAL_SUPPORT
+ terminalRegisterCommand(F("RFB.CODES"), [](const terminal::CommandContext& ctx) {
+ for (auto& code : _rfb_codes) {
+ char buffer[128] = {0};
+ snprintf_P(buffer, sizeof(buffer),
+ PSTR("\"%s\" proto=%u count=%u last=%u"),
+ code.protocol,
+ code.raw.c_str(),
+ code.count,
+ code.last
+ );
+ ctx.output.println(buffer);
+ }
+ });
+#endif
+
+ // Main bulk of the processing goes on in here
+ RfbridgeBroker::Register(_rpnBrokerRfbridgeCallback);
+}
+
+#endif // RFB_SUPPORT
+
 void _rpnShowStack(Print& print) {
  print.println(F("Stack:"));
 
@@ -369,6 +580,14 @@ void _rpnInit() {
 
  #endif
 
+ #if RFB_SUPPORT
+ rpn_operator_set(_rpn_ctxt, "rfb_pop", 2, _rpnRfbPop);
+ rpn_operator_set(_rpn_ctxt, "rfb_info", 2, _rpnRfbInfo);
+ rpn_operator_set(_rpn_ctxt, "rfb_sequence", 4, _rpnRfbSequence);
+ rpn_operator_set(_rpn_ctxt, "rfb_match", 3, _rpnRfbMatcher);
+ rpn_operator_set(_rpn_ctxt, "rfb_match_wait", 4, _rpnRfbWaitMatch);
+ #endif
+
  #if MQTT_SUPPORT
  rpn_operator_set(_rpn_ctxt, "mqtt_send", 2, [](rpn_context & ctxt) -> rpn_error {
  rpn_value message;
@@ -384,10 +603,12 @@ void _rpnInit() {
  #endif
 
  // Some debugging. Dump stack contents
- rpn_operator_set(_rpn_ctxt, "showstack", 0, [](rpn_context & ctxt) -> rpn_error {
- _rpnShowStack(terminalDefaultStream());
- return 0;
- });
+ #if TERMINAL_SUPPORT
+ rpn_operator_set(_rpn_ctxt, "showstack", 0, [](rpn_context & ctxt) -> rpn_error {
+ _rpnShowStack(terminalDefaultStream());
+ return 0;
+ });
+ #endif
 
  // And, simple string logging
  #if DEBUG_SUPPORT
@@ -591,6 +812,10 @@ void rpnSetup() {
 
  StatusBroker::Register(_rpnBrokerStatus);
 
+ #if RFB_SUPPORT
+ _rpnRfbSetup();
+ #endif
+
  #if SENSOR_SUPPORT
  SensorReadBroker::Register(_rpnBrokerCallback);
  #endif