RTSPMediaSourceMTS.h

Go to the documentation of this file.

#pragma once
 
#include "AudioTools/Concurrency/LockFree.h"
#include "IMediaSource.h"
#include "RTSPFormat.h"
 
namespace audio_tools {

class RTSPMediaSourceMTS : public IMediaSource, public Print {
 public:
  static constexpr int kPacketSize = 188;
  static constexpr int kPacketsPerFragment = 7;

  RTSPMediaSourceMTS(int bufferPackets = 128, float sendRate = 100)
      : buffer(bufferPackets * kPacketSize) {
    capacity = buffer.availableForWrite();
    format.setFramerate(sendRate);
    format.setFragmentSize(kPacketsPerFragment * kPacketSize);
  }
 
  RTSPFormat &getFormat() override { return format; }

  void start() override {
    uint8_t tmp[kPacketSize];
    while (buffer.readArray(tmp, kPacketSize) > 0) {
    }
    is_active = true;
  }

  void stop() override { is_active = false; }
 
  size_t write(uint8_t c) override { return write(&c, 1); }

  size_t write(const uint8_t *data, size_t len) override {
    size_t pos = 0;
    while (pos < len) {
      size_t n = kPacketSize - packet_len;
      if (n > len - pos) n = len - pos;
      memcpy(packet + packet_len, data + pos, n);
      packet_len += n;
      pos += n;
      if (packet_len == kPacketSize) {
        addPacket();
        packet_len = 0;
      }
    }
    return len;
  }

  int packetSize() override {
    int packets = buffer.available() / kPacketSize;
    if (packets > kPacketsPerFragment) packets = kPacketsPerFragment;
    return packets * kPacketSize;
  }

  int readBytes(void *dest, int maxBytes) override {
    int len = buffer.available() < maxBytes ? buffer.available() : maxBytes;
    len = len / kPacketSize * kPacketSize;
    return buffer.readArray((uint8_t *)dest, len);
  }

  uint32_t droppedPackets() { return dropped; }

  uint32_t droppedVideoPackets() { return dropped_video; }
 
 protected:
  RTSPFormatMTS format;
  RingBufferSPSC<uint8_t> buffer;
  uint8_t packet[kPacketSize];
  size_t packet_len = 0;
  volatile bool is_active = false;
  int capacity = 0;
  uint32_t dropped = 0;
  uint32_t dropped_video = 0;
  int video_pid = -1;  // learned from the PES stream ids
  bool dropping_video = false;
 
  void addPacket() {
    if (!is_active) return;  // no client is streaming
    int pid = ((packet[1] & 0x1F) << 8) | packet[2];
    bool pusi = packet[1] & 0x40;
    if (pusi && isVideoPes()) video_pid = pid;
    bool is_video = pid == video_pid;
    int free = buffer.availableForWrite();
 
    // too slow: drop video from the start of the next frame on, until the
    // buffer has recovered and the next keyframe (preceded by the PAT)
    if (dropping_video) {
      if (pid == 0 && free >= capacity / 2) {
        dropping_video = false;
        LOGI("RTSPMediaSourceMTS: resuming video at the next keyframe");
      }
    } else if (is_video && pusi && free < capacity / 4) {
      dropping_video = true;
      LOGW("RTSPMediaSourceMTS: network too slow - dropping video");
    }
    if (dropping_video && is_video) {
      dropped_video++;
      return;
    }
 
    // only whole packets: the free space can only grow while we write
    if (free < kPacketSize) {
      if (dropped++ % 100 == 0) {
        LOGW("RTSPMediaSourceMTS: buffer full, %u packets dropped",
             (unsigned)dropped);
      }
      return;
    }
    buffer.writeArray(packet, kPacketSize);
  }

  bool isVideoPes() {
    int afc = (packet[3] >> 4) & 3;
    if (!(afc & 1)) return false;  // no payload
    int pos = (afc & 2) ? 5 + packet[4] : 4;
    if (pos + 4 > kPacketSize) return false;
    return packet[pos] == 0 && packet[pos + 1] == 0 && packet[pos + 2] == 1 &&
           (packet[pos + 3] & 0xF0) == 0xE0;
  }
};
 
}  // namespace audio_tools