BLESerialStream.h

Go to the documentation of this file.

#pragma once
 
#include "AudioToolsConfig.h"
#include "AudioTools/CoreAudio/BaseStream.h"
#include "AudioTools/CoreAudio/Buffers.h"
#include "AudioTools/CoreAudio/AudioBasic/StrView.h"
#include "AudioTools/Concurrency/Mutex.h"
#include "AudioTools/Concurrency/LockGuard.h"
 
#ifndef BLE_SERIAL_MTU_OVERHEAD
#define BLE_SERIAL_MTU_OVERHEAD 5
#endif
 
#define BLE_SERIAL_HEADER_SIZE 2
 
#define BLE_SERIAL_MAX_VALUE_LEN 512
 
#ifndef BLE_SERIAL_BUFFER_SIZE
#define BLE_SERIAL_BUFFER_SIZE 4096
#endif
 
#ifndef BLE_SERIAL_PACKET_COUNT
#define BLE_SERIAL_PACKET_COUNT 100
#endif
 
namespace audio_tools {
 
class BLESerialPacketBuffer {
 public:
  void setMutex(MutexBase *mutex) { p_mutex = mutex; }
 
  void resize(int bytes, int packets, bool framed) {
    LockGuard guard(p_mutex);
    is_framed = framed;
    if (data.size() != (size_t)bytes) data.resize(bytes);
    if (is_framed && sizes.size() != (size_t)packets) sizes.resize(packets);
  }
 
  bool isAllocated() { return data.size() > 0; }
 
  void clear() {
    LockGuard guard(p_mutex);
    data.reset();
    sizes.reset();
  }
 
  int write(const uint8_t *packet, int len) {
    LockGuard guard(p_mutex);
    if (!is_framed) return data.writeArray(packet, len);
    if (sizes.availableForWrite() == 0 || data.availableForWrite() < len)
      return 0;
    sizes.write(len);
    return data.writeArray(packet, len);
  }
 
  int read(uint8_t *result, int len) {
    LockGuard guard(p_mutex);
    if (!is_framed) return data.readArray(result, len);
    uint16_t size = 0;
    if (!sizes.peek(size)) return 0;
    if (size > len) {
      LOGE("BLESerial: read size %d too small: it must be >= %d", len, size);
      return 0;
    }
    sizes.read(size);
    return data.readArray(result, size);
  }
 
  int available() {
    LockGuard guard(p_mutex);
    if (!is_framed) return data.available();
    uint16_t size = 0;
    return sizes.peek(size) ? size : 0;
  }
 
  int availableForWrite() {
    LockGuard guard(p_mutex);
    if (is_framed && sizes.availableForWrite() == 0) return 0;
    return data.availableForWrite();
  }
 
 protected:
  RingBuffer<uint8_t> data{0};
  RingBuffer<uint16_t> sizes{0};
  MutexBase *p_mutex = nullptr;
  bool is_framed = false;
};
 
class BLESerialStream : public AudioStream {
 public:
  BLESerialStream(int mtu) { max_transfer_size = mtu; }
 
  virtual void end() = 0;
 
  virtual bool connected() = 0;
 
  operator bool() override { return connected(); }
 
  void setAudioInfo(AudioInfo newInfo) override {
    is_audio_info_defined = true;
    AudioStream::setAudioInfo(newInfo);
    if (is_audio_info_active) {
      writeAudioInfo(toStr(newInfo));
    }
  }
 
  void setServiceUUID(const char *uuid) { BLE_SERIAL_SERVICE_UUID = uuid; }
 
  void setRxUUID(const char *uuid) { BLE_CH2_UUID = uuid; }
 
  void setTxUUID(const char *uuid) { BLE_CH1_UUID = uuid; }
 
  void setAudioInfoUUID(const char *uuid) { BLE_INFO_UUID = uuid; }
 
  void setAudioInfoActive(bool flag) { is_audio_info_active = flag; }
 
  void setFramed(bool flag) { is_framed = flag; }
 
  void setMTU(int mtu) { max_transfer_size = mtu; }
 
  void setRxBufferSize(int bytes, int packets = BLE_SERIAL_PACKET_COUNT) {
    rx_buffer_size = bytes;
    rx_packet_count = packets;
  }
 
  void setTxBufferSize(int bytes, int packets = BLE_SERIAL_PACKET_COUNT) {
    tx_buffer_size = bytes;
    tx_packet_count = packets;
  }
 
  int packetSize() {
    int result = getMTU() - BLE_SERIAL_MTU_OVERHEAD;
    int max = BLE_SERIAL_MAX_VALUE_LEN - BLE_SERIAL_HEADER_SIZE;
    return result < max ? result : max;
  }
 
  StrView toStr(AudioInfo info) {
    snprintf(audio_info_str, sizeof(audio_info_str), "%d:%d:%d",
             (int)info.sample_rate, (int)info.channels,
             (int)info.bits_per_sample);
    return StrView(audio_info_str);
  }
 
  AudioInfo toInfo(const uint8_t *data, size_t len) {
    AudioInfo result;
    char str[40] = {0};
    memcpy(str, data, len < sizeof(str) - 1 ? len : sizeof(str) - 1);
    int sample_rate = 0, channels = 0, bits = 0;
    if (sscanf(str, "%d:%d:%d", &sample_rate, &channels, &bits) == 3) {
      result.sample_rate = sample_rate;
      result.channels = channels;
      result.bits_per_sample = bits;
    }
    return result;
  }
 
 protected:
  // disable copy constructor
  BLESerialStream(BLESerialStream const &other) = delete;
  // disable assign constructor
  void operator=(BLESerialStream const &other) = delete;
  int max_transfer_size = 0;
  int rx_buffer_size = BLE_SERIAL_BUFFER_SIZE;
  int rx_packet_count = BLE_SERIAL_PACKET_COUNT;
  int tx_buffer_size = BLE_SERIAL_BUFFER_SIZE;
  int tx_packet_count = BLE_SERIAL_PACKET_COUNT;
  bool is_audio_info_active = false;
  bool is_audio_info_defined = false;
  bool is_framed = false;
  char audio_info_str[40];
 
  // Bluetooth LE GATT UUIDs: change them with the setters if required
  const char *BLE_SERIAL_SERVICE_UUID = "6e400001-b5a3-f393-e0a9-e50e24dcca9e";
  // read by the client: server -> client
  const char *BLE_CH1_UUID = "6e400002-b5a3-f393-e0a9-e50e24dcca9e";
  // written by the client: client -> server
  const char *BLE_CH2_UUID = "6e400003-b5a3-f393-e0a9-e50e24dcca9e";
  const char *BLE_INFO_UUID = "6e400004-b5a3-f393-e0a9-e50e24dcca9e";
 
  virtual int getMTU() { return max_transfer_size; }
 
  virtual void writeAudioInfo(StrView info) = 0;
 
  void receiveAudioInfo(const uint8_t *data, size_t size) {
    if (!is_audio_info_active || size == 0) return;
    AudioInfo ai = toInfo(data, size);
    if (!ai) {
      LOGW("BLESerial: invalid audio info");
      return;
    }
    // no echo to the other side
    AudioStream::setAudioInfo(ai);
  }
 
  static void setHeader(uint8_t *packet, int len) {
    packet[0] = len & 0xFF;
    packet[1] = (len >> 8) & 0xFF;
  }
 
  static int getHeader(const uint8_t *packet, int packetLen) {
    if (packetLen < BLE_SERIAL_HEADER_SIZE) return 0;
    int len = packet[0] | (packet[1] << 8);
    int max = packetLen - BLE_SERIAL_HEADER_SIZE;
    if (len > max) {
      LOGW("BLESerial: incomplete packet: %d of %d bytes", max, len);
      len = max;
    }
    return len;
  }
};
 
class BLESerialServerBase : public BLESerialStream {
 public:
  BLESerialServerBase(int mtu) : BLESerialStream(mtu) {}
 
  virtual bool begin(const char *name) = 0;
 
  size_t readBytes(uint8_t *data, size_t len) override {
    poll();
    return rx_buffer.read(data, len);
  }
 
  int available() override {
    poll();
    return rx_buffer.available();
  }
 
  size_t write(const uint8_t *data, size_t len) override {
    LOGD("BLESerialServer::write: %d", (int)len);
    if (!connected()) return 0;
    setupBuffers();
    if (is_framed && (int)len > packetSize()) {
      LOGE("BLESerial: frame of %d bytes > max %d", (int)len, packetSize());
      return 0;
    }
    return tx_buffer.write(data, len);
  }
 
  int availableForWrite() override {
    if (!connected()) return 0;
    setupBuffers();
    int result = tx_buffer.availableForWrite();
    if (is_framed) return result < packetSize() ? 0 : packetSize();
    // make sure we copy always a consistent amount of data
    return result < DEFAULT_BUFFER_SIZE ? 0 : result;
  }
 
 protected:
  BLESerialPacketBuffer rx_buffer;
  BLESerialPacketBuffer tx_buffer;
  int receive_timeout_ms = 0;
 
  virtual void poll() {}
 
  void setupBuffers() {
    if (!rx_buffer.isAllocated()) {
      LOGI("BLESerial: buffers %d / %d bytes", rx_buffer_size, tx_buffer_size);
      rx_buffer.resize(rx_buffer_size, rx_packet_count, is_framed);
      tx_buffer.resize(tx_buffer_size, tx_packet_count, is_framed);
    }
  }
 
  void receiveData(const uint8_t *data, size_t len) {
    if (len == 0) return;
    setupBuffers();
    uint32_t start = millis();
    while (rx_buffer.availableForWrite() < (int)len) {
      if (millis() - start >= (uint32_t)receive_timeout_ms) {
        LOGW("BLESerial: receive buffer full: %d bytes lost", (int)len);
        return;
      }
      // wait for the buffer to get freed up
      delay(5);
    }
    rx_buffer.write(data, len);
  }
 
  int provideData(uint8_t *packet, int maxLen) {
    setupBuffers();
    int max = maxLen - BLE_SERIAL_HEADER_SIZE;
    if (max > packetSize()) max = packetSize();
    int len = 0;
    if (max > 0 && tx_buffer.available() > 0) {
      len = tx_buffer.read(packet + BLE_SERIAL_HEADER_SIZE, max);
    }
    setHeader(packet, len);
    return len + BLE_SERIAL_HEADER_SIZE;
  }
 
  void onDisconnected() {
    // do not send outdated data to the next client
    tx_buffer.clear();
  }
};
 
class BLESerialClientBase : public BLESerialStream {
 public:
  BLESerialClientBase(int mtu) : BLESerialStream(mtu) {}
 
  bool begin(const char *localName, int seconds) {
    TRACEI();
    if (!startClient(localName)) {
      LOGE("BLESerial: starting BLE failed");
      return false;
    }
    is_started = true;
    uint32_t start = millis();
    while (!connected() && millis() - start < 1000ul * seconds) {
      delay(10);
    }
    return seconds == 0 || connected();
  }
 
  size_t readBytes(uint8_t *data, size_t len) override {
    TRACED();
    if (rx_buffer.available() == 0) {
      if (!connected()) return 0;
      receivePacket();
    }
    return rx_buffer.read(data, len);
  }
 
  int available() override {
    int result = rx_buffer.available();
    if (result > 0) return result;
    // we need to read from the server to know more
    return connected() ? packetSize() : 0;
  }
 
  size_t write(const uint8_t *data, size_t len) override {
    TRACED();
    if (!connected()) return 0;
    if (is_framed) {
      if ((int)len > packetSize()) {
        LOGE("BLESerial: frame of %d bytes > max %d", (int)len, packetSize());
        return 0;
      }
      return sendPacket(data, len) ? len : 0;
    }
    // send only packets with the max size
    for (size_t j = 0; j < len; j++) {
      write_buffer.write(data[j]);
      if (write_buffer.isFull()) {
        if (!sendPacket(write_buffer.data(), write_buffer.available())) {
          write_buffer.reset();
          return j + 1;
        }
        write_buffer.reset();
      }
    }
    return len;
  }
 
  void flush() override {
    if (write_buffer.available() > 0 && connected()) {
      sendPacket(write_buffer.data(), write_buffer.available());
    }
    write_buffer.reset();
  }
 
  int availableForWrite() override {
    if (!connected()) return 0;
    return is_framed ? packetSize() : DEFAULT_BUFFER_SIZE;
  }
 
  bool connected() override {
    if (!is_started) return false;
    if (isConnected()) return true;
    if (!connect()) return false;
    onConnected();
    return true;
  }
 
  void setWriteThrottle(int ms) { write_throttle = ms; }
 
  void setConfirmWrite(bool flag) { write_confirmation_flag = flag; }
 
 protected:
  BLESerialPacketBuffer rx_buffer;
  SingleBuffer<uint8_t> write_buffer{0};
  Vector<uint8_t> read_packet{0};
  int write_throttle = 0;
  bool write_confirmation_flag = false;
  bool is_started = false;
 
  virtual bool startClient(const char *localName) = 0;
  virtual bool isConnected() = 0;
  virtual bool connect() = 0;
  virtual int readPacket(uint8_t *data, int maxLen) = 0;
  virtual bool writePacket(const uint8_t *data, int len) = 0;
  virtual bool readAudioInfo() = 0;
 
  void onConnected() {
    TRACEI();
    rx_buffer.resize(rx_buffer_size, rx_packet_count, is_framed);
    rx_buffer.clear();
    write_buffer.resize(packetSize());
    write_buffer.reset();
    // the server might use a bigger MTU than we requested
    read_packet.resize(BLE_SERIAL_MAX_VALUE_LEN);
    if (is_audio_info_active) {
      // the side that defined the audio info is the source
      if (is_audio_info_defined) {
        writeAudioInfo(toStr(audioInfo()));
      } else {
        readAudioInfo();
      }
    }
  }
 
  void receivePacket() {
    int len = readPacket(read_packet.data(), read_packet.size());
    int payload = getHeader(read_packet.data(), len);
    if (payload > 0) {
      rx_buffer.write(read_packet.data() + BLE_SERIAL_HEADER_SIZE, payload);
    }
  }
 
  bool sendPacket(const uint8_t *data, size_t len) {
    bool rc = writePacket(data, len);
    if (write_throttle > 0) delay(write_throttle);
    return rc;
  }
};
 
}  // namespace audio_tools