RTSPServer.h

/*
 * Author: Phil Schatzmann
 *
 * Based on Micro-RTSP library:
 * https://github.com/geeksville/Micro-RTSP
 * https://github.com/Tomp0801/Micro-RTSP-Audio
 *
 */
 
#pragma once
#include "RTSPSession.h"
#ifdef ESP32
#include <WiFi.h>
#include <esp_wifi.h>
#include "AudioTools/Concurrency/RTOS.h"
#endif
 
namespace audio_tools {
 
template <typename Platform>
class RTSPServer {
 public:
  using streamer_t = RTSPAudioStreamerBase<Platform>;
 
  RTSPServer(streamer_t& streamer, int port = 8554, int core = 1) {
    this->streamer = &streamer;
    this->port = port;
    this->core = core;
    // Create tasks with specified core
    serverTask.create("RTSPServerThread", 10000, 5, core);
    sessionTask.create("RTSPSessionTask", 8000, 8, core);
  }
 
  ~RTSPServer() { stop(); }
 
  void setOnSessionPath(bool (*cb)(const char* path, void* ref), void* ref = nullptr) {
    onSessionPathCb = cb;
    onSessionPathRef = ref;
  }
 
#if defined(USE_RTSP_LOGIN)
 
  bool begin(const char* ssid, const char* password) {
    // Start Wifi
    WiFi.begin(ssid, password);
    while (WiFi.status() != WL_CONNECTED) {
      delay(500);
      Serial.print(".");
    }
    WiFi.setSleep(WIFI_PS_NONE);
 
    Serial.println();
    Serial.print("connect to rtsp://");
    Serial.print(WiFi.localIP());
    Serial.print(":");
    Serial.println(port);
    Serial.println();
    return begin();
  }
  
#endif
 
  bool begin() {
    bool ok = runAsync();
    if (!ok) {
      LOGE("Couldn't start RTSP server");
    }
    return ok;
  }
 
 
  audio_tools::Task& getTaskHandle() { return serverTask; };
 
  int clientCount() { return client_count;}
 
  operator bool() { return client_count > 0;}
 
  void setSessionTimoutMs(unsigned long ms){ sessionTimeoutMs = ms;}
 
 protected:
  int port;  // port that the RTSP Server listens on
  int core;  // the core number the RTSP runs on (platform-specific)
  audio_tools::Task serverTask{"RTSPServerThread", 15000, 5, core};
  audio_tools::Task sessionTask{"RTSPSessionTask", 15000, 8, core};
  typename Platform::TcpServerType* server = nullptr;  // platform server
  typename Platform::TcpClientType client;  // RTSP client connection (value)
  int client_count = 0;  // number of connected clients
  streamer_t* streamer = nullptr;  // RTSPAudioStreamer object that acts as a
                                   // source for data streams
  bool (*onSessionPathCb)(const char*, void*) = nullptr; // session path callback
  void* onSessionPathRef = nullptr;
  unsigned long sessionTimeoutMs = 20000; // 20 seconds
 
  bool runAsync() {
    LOGI("Running RTSP server on port %d", port);
    streamer->initAudioSource();
    if (server == nullptr) {
      server = Platform::createServer(port);
      LOGI("RTSP server started on port %d", port);
    }
    if (!serverTask.begin([this]() { serverThreadLoop(); })) {
      LOGE("Couldn't start server thread");
      return false;
    }
    return true;
  }
 
  void serverThreadLoop() {
    unsigned long lastCheck = millis();
 
    LOGD("Server thread listening... (numClients: %d)", client_count);
 
    // only allow one client at a time
    if (client_count == 0) {
      auto newClient = Platform::getAvailableClient(server);
      if (newClient.connected()) {
        client = newClient;  // copy/move assign
        LOGI("Client connected");
        if (!sessionTask.begin([this]() { sessionThreadLoop(); })) {
          LOGE("Couldn't start sessionThread");
          Platform::closeSocket(&client);
        } else {
          client_count++;
          LOGI("Number of clients: %d", client_count);
        }
      }
    } else {
      LOGD("Waiting for current session to end (numClients: %d)", client_count);
    }
    int time = 200 - (millis() - lastCheck);
    if (time < 0) time = 0;
    delay(time);
  }
 
  void stop() {
    LOGI("Stopping RTSP server");
 
    // Stop tasks
    sessionTask.end();
    serverTask.end();
 
    // Close sockets
    if (server) {
      // server->stop(); // not supported by EthernetServer
      delete server;
      server = nullptr;
    }
 
    client_count = 0;
    LOGI("RTSP server stopped");
  }
 
  void sessionThreadLoop() {
    typename Platform::TcpClientType* s = &client;
    unsigned long lastCheck = millis();
 
    // TODO check if everything is ok to go
    LOGD("RTSP Task running");
 
    // our threads RTSP session and state
    RtspSession<Platform>* rtsp = new RtspSession<Platform>(*s, *streamer);
    if (onSessionPathCb) {
      rtsp->setOnSessionPath(onSessionPathCb, onSessionPathRef);
    }
    assert(rtsp != nullptr);
 
    LOGI("Session ready");
 
 
    // Session timeout logic
    unsigned long lastRequestTime = millis();
 
    while (rtsp->isSessionOpen()) {
      uint32_t timeout = 30;
      bool gotRequest = rtsp->handleRequests(timeout);
      if (gotRequest) {
        lastRequestTime = millis();
        LOGD("Request handling successful");
      } else {
        LOGI("Request handling timed out or no data yet");
      }
 
      // If streaming, check for session timeout
      if (rtsp->isStreaming()) {
        if ((millis() - lastRequestTime) > sessionTimeoutMs) {
          LOGW("Session timeout: no client request received for 20 seconds, closing session");
          break;
        }
      }
 
      int time = timeout - (millis() - lastCheck);
      if (time < 0) time = 0;
      delay(time);
    }
 
    LOGI("Session loop exited - session no longer open");
 
    // cleanup when session ends
    LOGI("sessionThread stopped, cleaning up");
    delete rtsp;
 
    // Clean up client socket
    if (client.connected()) {
      Platform::closeSocket(&client);
    }
 
    // Add delay to ensure complete cleanup before accepting new connections
    delay(500);  
 
    client_count--;
    LOGI("Session cleaned up: (numClients: %d)", client_count);
 
    // end the task - this should be the last thing we do
    sessionTask.end();
  }
};
 
}  // namespace audio_tools