SubscriptionMgrDevice.h

Go to the documentation of this file.

#pragma once
 
#include <functional>
 
#include "basic/EscapingPrint.h"
#include "basic/Logger.h"
#include "basic/NullPrint.h"
#include "basic/Str.h"
#include "basic/StrView.h"
#include "basic/Url.h"
#include "basic/Vector.h"
#include "concurrency/ListLockFree.h"
#include "dlna/devices/ISubscriptionMgrDevice.h"
#include "http/Http.h"
#include "http/Server/HttpRequest.h"
#include "http/Server/IHttpServer.h"
 
namespace tiny_dlna {
 
// forward declaration for friend
class DLNAService;
template <typename ClientType>
class DLNADevice;
 
struct Subscription {
  Str sid;
  Str callback_url;  // as provided in CALLBACK header
  uint32_t timeout_sec = 1800;
  uint32_t seq = 0;
  uint64_t expires_at = 0;
  DLNAServiceInfo* service = nullptr;
};
struct PendingNotification {
  Subscription* p_subscription =
      nullptr; 
  std::function<size_t(Print&, void*)> writer; 
  void* ref = nullptr; 
  int error_count = 0; 
  int seq = 0;
};
 
template <typename ClientType>
class SubscriptionMgrDevice : public ISubscriptionMgrDevice {
  template <typename>
  friend class DLNADevice;
 
 public:
  SubscriptionMgrDevice() = default;
 
  ~SubscriptionMgrDevice() { clear(); }
 
  Str subscribe(DLNAServiceInfo& service, const char* callbackUrl,
                const char* sid = nullptr,
                uint32_t timeoutSec = 1800) override {
    // simple SID generation
    DlnaLogger.log(DlnaLogLevel::Info, "subscribe: %s %s",
                   StrView(service.service_id).c_str(),
                   StrView(callbackUrl).c_str());
 
    bool hasSid = !StrView(sid).isEmpty();
 
    // NEW subscription without a callback is invalid (GENA requirement)
    if (!hasSid) {
      if (StrView(callbackUrl).isEmpty()) {
        DlnaLogger.log(
            DlnaLogLevel::Warning,
            "subscribe: missing CALLBACK header for new subscription");
        return Str();
      }
    }
 
    // If sid provided, attempt to renew existing subscription for service
    if (hasSid) {
      Str renewed = renewSubscription(service, sid, callbackUrl, timeoutSec);
      if (!renewed.isEmpty()) return renewed;
      // not found: fall through and create new subscription
    }
 
    // Make sure that we have an url
    if (StrView(callbackUrl).isEmpty()) {
      DlnaLogger.log(DlnaLogLevel::Warning,
                     "subscribe: missing CALLBACK header for new subscription");
      return Str();
    }
 
    // generate uuid-based SID
    char buffer[64];
    snprintf(buffer, sizeof(buffer), "uuid:%lu", millis());
 
    // fill subscription
    Subscription* s = new Subscription();
    s->sid = buffer;
    s->callback_url = callbackUrl;
    s->timeout_sec = timeoutSec;
    s->seq = 0;
    s->expires_at = millis() + (uint64_t)timeoutSec * 1000ULL;
    s->service = &service;
 
    for (auto it = subscriptions.begin(); it != subscriptions.end(); it++) {
      auto existing_sub = *it;
      if (existing_sub->service == &service &&
          StrView(existing_sub->callback_url).equals(callbackUrl)) {
        // Found existing subscription for same service and callback URL
        DlnaLogger.log(DlnaLogLevel::Info,
                       "subscribe: found existing subscription for service "
                       "'%s' and callback '%s', renewing SID '%s'",
                       StrView(service.service_id).c_str(),
                       StrView(callbackUrl).c_str(), existing_sub->sid.c_str());
        // Renew existing subscription
        existing_sub->timeout_sec = timeoutSec;
        existing_sub->expires_at = millis() + (uint64_t)timeoutSec * 1000ULL;
        return existing_sub->sid;
      }
    }
 
    // add subscription
    subscriptions.push_back(s);
    return s->sid;
  }
 
  bool unsubscribe(DLNAServiceInfo& service, const char* sid) override {
    for (auto it = subscriptions.begin(); it != subscriptions.end(); ++it) {
      Subscription* s = *it;
      if (s->service == &service && StrView(s->sid).equals(sid)) {
        // remove pending notifications that reference this subscription
        for (auto pit = pending_list.begin(); pit != pending_list.end();) {
          if (pit->p_subscription == s) {
            pending_list.erase(pit);
            pit = pending_list
                      .begin();  // restart after erase (iterator invalidated)
          } else {
            ++pit;
          }
        }
        delete s;
        subscriptions.erase(it);
        return true;
      }
    }
    return false;
  }
 
  void addChange(DLNAServiceInfo& service,
                 std::function<size_t(Print&, void*)> changeWriter,
                 void* ref) override {
    bool any = false;
    // do not enqueue if subscriptions are inactive
    if (!is_active) return;
 
    // enqueue notifications to be posted later by post()
    for (auto it = subscriptions.begin(); it != subscriptions.end(); ++it) {
      Subscription* sub = *it;
      if (sub->service != &service) continue;
      any = true;
      NullPrint np;
      PendingNotification pn;
      // store pointer to the subscription entry (no snapshot)
      pn.p_subscription = sub;
      pn.ref = ref;
      pn.writer = changeWriter;
      pn.seq = sub->seq;
      pending_list.push_back(pn);
      // increment seq AFTER queuing (so first notification uses SEQ=0)
      sub->seq++;
    }
    if (!any) {
      DlnaLogger.log(DlnaLogLevel::Info, "service '%s' has no subscriptions",
                     service.service_id.c_str());
    }
  }
 
  int publish() override {
    if (!is_active) {
      // clear any queued notifications when subscriptions are turned off
      clear();
      return 0;
    }
    // First remove expired subscriptions so we don't deliver to them.
    removeExpired();
 
    if (pending_list.empty()) return 0;
    // Attempt to process each pending notification once. If processing
    // fails (non-200), leave the entry in the queue for a later retry.
    int processed = 0;
    for (auto it = pending_list.begin(); it != pending_list.end();) {
      PendingNotification pn = *it;
      // Ensure the subscription pointer is valid; if not, drop the entry.
      if (pn.p_subscription == nullptr) {
        DlnaLogger.log(DlnaLogLevel::Warning,
                       "pending notification dropped: missing subscription");
        pending_list.erase(it);
        it = pending_list.begin();  // restart after erase
        continue;
      }
      Subscription& sub = *pn.p_subscription;
      char seqBuf[32];
      snprintf(seqBuf, sizeof(seqBuf), "%d", pn.seq);
      Url cbUrl(sub.callback_url.c_str());
      HttpRequest<ClientType> http;
      http.setTimeout(DLNA_HTTP_REQUEST_TIMEOUT_MS);
      http.setHost(cbUrl.host());
      http.setAgent("tiny-dlna-notify");
      http.request().put("NT", "upnp:event");
      http.request().put("NTS", "upnp:propchange");
      http.request().put("SEQ", seqBuf);
      http.request().put("SID", sub.sid.c_str());
      DlnaLogger.log(DlnaLogLevel::Info, "Notify %s", cbUrl.url());
      DlnaLogger.log(DlnaLogLevel::Info, "- SEQ: %s", seqBuf);
      DlnaLogger.log(DlnaLogLevel::Info, "- SID: %s", sub.sid.c_str());
      int rc = http.notify(cbUrl, createXML, "text/xml", &pn);
      DlnaLogger.log(DlnaLogLevel::Info, "Notify %s -> %d", cbUrl.url(), rc);
      if (rc == 200) {
        pending_list.erase(it);
        it = pending_list.begin();
        processed++;
      } else {
        // Increment error count on the stored entry (not the local copy)
        it->error_count++;
        if (it->error_count > MAX_NOTIFY_RETIES) {
          DlnaLogger.log(DlnaLogLevel::Warning,
                         "dropping notify to %s after %d errors with rc=%d %s",
                         cbUrl.url(), it->error_count, rc,
                         http.reply().statusMessage());
          pending_list.erase(it);
          it = pending_list.begin();
        } else {
          ++it;
        }
      }
    }
 
    DlnaLogger.log(
        DlnaLogLevel::Info,
        "Published: %d notifications, %d remaining (for %d subscriptions)",
        processed, pendingCount(), subscriptionsCount());
    return processed;
  }
 
  size_t subscriptionsCount() override { return subscriptions.size(); }
 
  size_t pendingCount() override { return pending_list.size(); }
 
  void setSubscriptionsActive(bool flag) override { is_active = flag; }
 
  void end() override { setSubscriptionsActive(false); }
 
  bool isSubscriptionsActive() override { return is_active; }
 
 protected:
  // store pointers to heap-allocated Subscription objects. This keeps
  // references stable (we manage lifetimes explicitly) and avoids
  // large copies when enqueuing notifications.
  // Vector<Subscription*> subscriptions;
  // Vector<PendingNotification> pending_list;
  ListLockFree<Subscription*> subscriptions;
  ListLockFree<PendingNotification> pending_list;
  bool is_active = true;
 
  void removeExpired() {
    uint64_t now = millis();
    auto it = subscriptions.begin();
    while (it != subscriptions.end()) {
      Subscription* s = *it;
      if (s->expires_at != 0 && s->expires_at <= now) {
        DlnaLogger.log(DlnaLogLevel::Info, "removing expired subscription %s",
                       s->sid.c_str());
        unsubscribe(*s->service, s->sid.c_str());
        it = subscriptions.begin();  // restart after erase
      } else {
        ++it;
      }
    }
  }
 
  Str renewSubscription(DLNAServiceInfo& service, const char* sid,
                        const char* callbackUrl, uint32_t timeoutSec) {
    if (StrView(sid).isEmpty()) return Str();
    for (auto it = subscriptions.begin(); it != subscriptions.end(); ++it) {
      Subscription* ex = *it;
      if (ex->service == &service && StrView(ex->sid).equals(sid)) {
        // renew: update timeout and expiry, do not change SID
        ex->timeout_sec = timeoutSec;
        ex->expires_at = millis() + (uint64_t)timeoutSec * 1000ULL;
        // if a new callback URL was provided, update it
        if (callbackUrl != nullptr && !StrView(callbackUrl).isEmpty()) {
          ex->callback_url = callbackUrl;
        }
        DlnaLogger.log(DlnaLogLevel::Info, "renewed subscription %s",
                       ex->sid.c_str());
        return ex->sid;
      }
    }
    return Str();
  }
 
  void clear() {
    // free any remaining heap-allocated subscriptions
    for (int i = 0; i < subscriptions.size(); ++i) {
      delete subscriptions[i];
    }
    subscriptions.clear();
    // pending_list entries do not own subscription pointers
    pending_list.clear();
  }
 
  static size_t createXML(Print& out, void* ref) {
    EscapingPrint out_esc{out};
    PendingNotification& pn = *(PendingNotification*)ref;
    const char* service_abbrev =
        pn.p_subscription->service->subscription_namespace_abbrev;
    int instance_id = pn.p_subscription->service->instance_id;
 
    size_t written = 0;
    written += out.println("<?xml version=\"1.0\"?>");
    written += out.println(
        "<e:propertyset "
        "xmlns:e=\"urn:schemas-upnp-org:metadata-1-0/events\">");
    written += out.println("<e:property>");
    written += out.println("<LastChange>");
    // below should be escaped
    written +=
        out_esc.print("<Event xmlns=\"urn:schemas-upnp-org:metadata-1-0/");
    written += out_esc.print(service_abbrev);
    written += out_esc.print("/\">");
    written += out_esc.print("<InstanceID val=\"");
    written += out_esc.print(instance_id);
    written += out_esc.print("\">");
    if (pn.writer) {
      written += pn.writer(out_esc, pn.ref);
    }
    written += out_esc.print("</InstanceID>");
    written += out_esc.print("</Event>");
    // end of escaped
    written += out.println();
    written += out.println("</LastChange>");
    written += out.println("</e:property>");
    written += out.println("</e:propertyset>");
    return written;
  }
 
  bool processSubscribeRequest(IHttpServer& server, DLNAServiceInfo& service,
                               IClientHandler* client) {
    // Get headers from request
    const char* callbackHeader =
        client ? client->requestHeader().get("CALLBACK") : nullptr;
    const char* timeoutHeader =
        client ? client->requestHeader().get("TIMEOUT") : nullptr;
    const char* sidHeader =
        client ? client->requestHeader().get("SID") : nullptr;
 
    // Log the incoming headers
    DlnaLogger.log(DlnaLogLevel::Info, "- SUBSCRIBE CALLBACK: %s",
                   callbackHeader ? callbackHeader : "null");
    DlnaLogger.log(DlnaLogLevel::Info, "- SUBSCRIBE TIMEOUT: %s",
                   timeoutHeader ? timeoutHeader : "null");
    DlnaLogger.log(DlnaLogLevel::Info, "- SUBSCRIBE SID: %s",
                   sidHeader ? sidHeader : "null");
 
    // Parse and clean callback URL (remove angle brackets)
    Str cbStr;
    if (callbackHeader) {
      cbStr = callbackHeader;
      cbStr.replace("<", "");
      cbStr.replace(">", "");
      cbStr.trim();
    }
 
    // Sanitize SID (remove optional angle brackets)
    Str sidStr;
    if (sidHeader) {
      sidStr = sidHeader;
      sidStr.replace("<", "");
      sidStr.replace(">", "");
      sidStr.trim();
    }
 
    // Parse timeout (extract seconds from "Second-1800" format)
    uint32_t tsec = 1800;
    if (timeoutHeader && StrView(timeoutHeader).startsWith("Second-")) {
      tsec = atoi(timeoutHeader + 7);
    }
 
    // Create or renew subscription
    const char* callbackPtr = cbStr.c_str();
    const char* sidPtr = sidStr.c_str();
 
    Str sid = subscribe(service, callbackPtr, sidPtr, tsec);
    if (sid.isEmpty()) {
      DlnaLogger.log(DlnaLogLevel::Warning,
                     "subscribe request rejected (missing data)");
      if (client) {
        client->replyHeader().setValues(412, "Precondition Failed");
        client->replyHeader().put("Content-Length", 0);
        if (client->client()) client->replyHeader().write(*client->client());
        client->endClient();
      }
      return false;
    }
    DlnaLogger.log(DlnaLogLevel::Info, "- SID: %s", sid.c_str());
 
    // Send SUBSCRIBE response
    if (client) {
      client->replyHeader().setValues(200, "OK");
      client->replyHeader().put("SID", sid.c_str());
      client->replyHeader().put("TIMEOUT", "Second-1800");
      client->replyHeader().put("Content-Length", 0);
      if (client->client()) client->replyHeader().write(*client->client());
      client->endClient();
    }
 
    return true;
  }
 
  bool processUnsubscribeRequest(IHttpServer& server, DLNAServiceInfo& service,
                                 IClientHandler* client) {
    const char* sid = client ? client->requestHeader().get("SID") : nullptr;
 
    DlnaLogger.log(DlnaLogLevel::Info, "- UNSUBSCRIBE SID: %s",
                   sid ? sid : "null");
 
    if (sid) {
      bool ok = unsubscribe(service, sid);
      if (ok) {
        DlnaLogger.log(DlnaLogLevel::Info, "Unsubscribed: %s", sid);
        if (client) client->replyOK();
        return true;
      }
    }
 
    // SID not found or invalid
    DlnaLogger.log(DlnaLogLevel::Warning,
                   "handleUnsubscribeRequest: failed for SID %s",
                   sid ? sid : "(null)");
    if (client) client->replyOK();
    return false;
  }
};
 
}  // namespace tiny_dlna