CBorStream.h

Go to the documentation of this file.

/****************************************************************************
 * Copyright (C) 2024 Davide Costa                                          *
 *                                                                          *
 * This file is part of RMW desert.                                         *
 *                                                                          *
 *   RMW desert is free software: you can redistribute it and/or modify it  *
 *   under the terms of the GNU General Public License as published by the  *
 *   Free Software Foundation, either version 3 of the License, or any      *
 *   later version.                                                         *
 *                                                                          *
 *   RMW desert is distributed in the hope that it will be useful,          *
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of         *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the          *
 *   GNU General Public License for more details.                           *
 *                                                                          *
 *   You should have received a copy of the GNU General Public License      *
 *   along with RMW desert.  If not, see <http://www.gnu.org/licenses/>.    *
 ****************************************************************************/

 
#ifndef CBORSTREAM_H_
#define CBORSTREAM_H_
 
#include "TcpDaemon.h"
#include "TopicsConfig.h"

 
#include <map>
#include <queue>
#include <utility>
#include <vector>
#include <string>
#include <locale>
#include <codecvt>
#include <cstdint>
#include <cstdio>
#include <mutex>

 
#include "half.hpp"
 
#define PUBLISHER_TYPE  0
#define SUBSCRIBER_TYPE 1
#define CLIENT_TYPE     2
#define SERVICE_TYPE    3
 
#define MAX_BUFFER_CAPACITY 100
 
template <typename T, int MaxLen, typename Container=std::deque<T>>
class CircularQueue : public std::queue<T, Container> {
  public:
    void push(const T& value)
    {
        if (this->size() == MaxLen)
        {
           this->c.pop_front();
        }
        std::queue<T, Container>::push(value);
    }
};
 
namespace cbor
{
 
class TxStream
{
  public:
    TxStream(uint8_t stream_type, std::string stream_name, uint8_t stream_identifier);
    
    void start_transmission(uint64_t sequence_id);
    void start_transmission();
    void end_transmission();

    TxStream & operator<<(const uint64_t n);
    TxStream & operator<<(const uint32_t n);
    TxStream & operator<<(const uint16_t n);
    TxStream & operator<<(const uint8_t n);
    TxStream & operator<<(const int64_t n);
    TxStream & operator<<(const int32_t n);
    TxStream & operator<<(const int16_t n);
    TxStream & operator<<(const int8_t n);
    TxStream & operator<<(const char n);
    TxStream & operator<<(const float f);
    TxStream & operator<<(const double d);
    TxStream & operator<<(const std::string s);
    TxStream & operator<<(const std::u16string s);
    TxStream & operator<<(const bool b);
    
    template<typename T>
    inline TxStream & operator<<(const std::vector<T> v)
    {
      *this << static_cast<const uint32_t>(v.size());
      return serialize_sequence(v.data(), v.size());
    }
    
    TxStream & operator<<(const std::vector<bool> v);
    
    template<typename T>
    inline TxStream & serialize_sequence(const T * items, size_t size)
    {
      for (size_t i = 0; i < size; ++i)
      {
        *this << items[i];
      }
      return *this;
    }
 
  private:
    // Forward declarations to abstract the implementation
    struct WRITER;
    struct ERROR;
    
    uint8_t _stream_type;
    std::string _stream_name;
    uint8_t _stream_identifier;
    
    bool _overflow;
    uint8_t * _packet;
    WRITER * _writer;
    
    void new_packet();
    void handle_overrun(ERROR result);
    
    std::string toUTF8(const std::u16string source);
  
};
 
class RxStream
{
  public:
    RxStream(uint8_t stream_type, std::string stream_name, uint8_t stream_identifier);
    
    ~RxStream();
    
    bool data_available(int64_t sequence_id = 0);
    
    void clear_buffer();
    
    RxStream & operator>>(uint64_t & n);
    RxStream & operator>>(uint32_t & n);
    RxStream & operator>>(uint16_t & n);
    RxStream & operator>>(uint8_t & n);
    RxStream & operator>>(int64_t & n);
    RxStream & operator>>(int32_t & n);
    RxStream & operator>>(int16_t & n);
    RxStream & operator>>(int8_t & n);
    
    RxStream & operator>>(char & n);
    RxStream & operator>>(float & f);
    RxStream & operator>>(double & d);
    RxStream & operator>>(std::string & s);
    RxStream & operator>>(std::u16string & s);
    RxStream & operator>>(bool & b);
    
    template<typename T>
    inline RxStream & operator>>(std::vector<T> & v)
    {
      uint32_t size;
      *this >> size;
      v.resize(size);
      
      return deserialize_sequence(v.data(), size);
    }
    
    RxStream & operator>>(std::vector<bool> & v);
    
    template<typename T>
    inline RxStream & deserialize_sequence(T * items, size_t size)
    {
      for (size_t i = 0; i < size; ++i)
      {
        *this >> items[i];
      }
      return *this;
    }
    
    
    uint8_t get_type() const;
    std::string get_name() const;
    uint8_t get_identifier() const;
    
    void push_packet(std::vector<std::pair<void *, int>> packet);
    
    static void interpret_packets();
  
  private:
    // Forward declarations to abstract the implementation
    struct ITEM;
    
    uint8_t _stream_type;
    std::string _stream_name;
    uint8_t _stream_identifier;
    
    size_t _buffered_iterator;
    
    // packets: <packet <field, field_type>>
    std::vector<std::pair<void *, int>> _buffered_packet;
    CircularQueue<std::vector<std::pair<void *, int>>, MAX_BUFFER_CAPACITY> _received_packets;
    
    static const std::map<int, int> _stream_type_match_map;
    static std::vector<RxStream *> _listening_streams;
    
    union _cbor_value {
    int8_t i8;
    int16_t i16;
    int32_t i32;
    int64_t i64;
    float f32;
    double f64;
    uint8_t *bin;
    char *str;
    uint8_t str_copy[128];
    };
    
    static std::mutex _rx_mutex;
    
    static std::pair<void *, int> interpret_field(ITEM * items, size_t i, union _cbor_value & val);
    static void destroy_interpreted_field(std::pair<void *, int>& field);
    std::u16string toUTF16(const std::string source);
};
 
}  // namespace cbor
 
 
#endif