shm_pub_sub_vector.hpp

 
#ifndef __SHM_PS_VECTOR_LIB_H__
#define __SHM_PS_VECTOR_LIB_H__
 
#include <iostream>
#include <limits>
#include <string>
#include <regex>
#include <stdexcept>
#include <mutex>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/time.h>
#include <pthread.h>
 
#include "shm_base.hpp"
#include "shm_pub_sub.hpp"
 
namespace irlab
{
 
namespace shm
{
 
// ****************************************************************************
// ****************************************************************************
template <class T>
class Publisher<std::vector<T>>
{
public:
  Publisher(std::string name = "", int buffer_num = 3, PERM perm = DEFAULT_PERM);
  ~Publisher();
  
  void publish(const std::vector<T>& data);
  void _publish(const std::vector<T> data);
 
private:
  std::string shm_name;
  int shm_buf_num;
  PERM shm_perm;
  SharedMemory *shared_memory;
  RingBuffer *ring_buffer;
 
  size_t vector_size;
};
 
// ****************************************************************************
// ****************************************************************************
template <typename T>
class Subscriber<std::vector<T>>
{
public:
  Subscriber(std::string name = "");
  ~Subscriber();
  
  const std::vector<T> subscribe(bool *is_success);
  bool waitFor(uint64_t timeout_usec);
  void setDataExpiryTime_us(uint64_t time_us);
  
private:
  std::string shm_name;
  SharedMemory *shared_memory;
  RingBuffer *ring_buffer;
  int current_reading_buffer;
  uint64_t data_expiry_time_us;
 
  size_t vector_size;
};
 
 
// ****************************************************************************
// 関数定義
// （テンプレートクラス内の関数の定義はコンパイル時に実体化するのでヘッダに書く）
// ****************************************************************************
 
template <typename T>
Publisher<std::vector<T>>::Publisher(std::string name, int buffer_num, PERM perm)
: shm_name(name)
, shm_buf_num(buffer_num)
, shm_perm(perm)
, shared_memory(nullptr)
, ring_buffer(nullptr)
, vector_size(0)
{
  if (!std::is_standard_layout<T>::value)
  {
    throw std::runtime_error("shm::Publisher: Be setted not POD class in vector!");
  }
 
  shared_memory = new SharedMemoryPosix(shm_name, O_RDWR|O_CREAT, shm_perm);
  shared_memory->connect(RingBuffer::getSize(vector_size, shm_buf_num));
  if (shared_memory->isDisconnected())
  {
    throw std::runtime_error("shm::Publisher: Cannot get memory!");
  }
 
  ring_buffer = new RingBuffer(shared_memory->getPtr(), vector_size, shm_buf_num);
}
 
template <typename T>
Publisher<std::vector<T>>::~Publisher()
{
  if (ring_buffer != nullptr)
  {
    delete ring_buffer;
  }
  if (shared_memory != nullptr)
  {
    delete shared_memory;
  }
}
 
template <typename T>
void
Publisher<std::vector<T>>::publish(const std::vector<T>& data)
{
  if (data.size() > vector_size)
  {
    vector_size = data.size();
    delete ring_buffer;
    shared_memory->disconnect();
    shared_memory->connect(RingBuffer::getSize(sizeof(T)*vector_size, shm_buf_num));
    ring_buffer = new RingBuffer(shared_memory->getPtr(), sizeof(T)*vector_size, shm_buf_num);
  }
  
  int oldest_buffer = ring_buffer->getOldestBufferNum();
 
  T * first_ptr = &( (reinterpret_cast<T *>(ring_buffer->getDataList()))[oldest_buffer*vector_size] );
  memcpy(first_ptr, data.data(), sizeof(T) * vector_size);
 
  struct timespec t;
  clock_gettime(CLOCK_MONOTONIC_RAW, &t);
  ring_buffer->setTimestamp_us(((uint64_t) t.tv_sec * 1000000L) + ((uint64_t) t.tv_nsec / 1000L), oldest_buffer);
 
  ring_buffer->signal();
}
 
 
template <typename T>
void
Publisher<std::vector<T>>::_publish(const std::vector<T> data)
{
  publish(data);
}
 
 
template <typename T>
Subscriber<std::vector<T>>::Subscriber(std::string name)
: shm_name(name)
, shared_memory(nullptr)
, ring_buffer(nullptr)
, current_reading_buffer(0)
, data_expiry_time_us(2000000)
{
  if (!std::is_standard_layout<T>::value)
  {
    throw std::runtime_error("shm::Subscriber: Be setted not POD class!");
  }
  shared_memory = new SharedMemoryPosix(shm_name, O_RDWR, static_cast<PERM>(0));
 
}
 
 
template <typename T>
Subscriber<std::vector<T>>::~Subscriber()
{
  if (ring_buffer != nullptr)
  {
    delete ring_buffer;
  }
  if (shared_memory != nullptr)
  {
    delete shared_memory;
  }
}
 
 
template <typename T>
const std::vector<T>
Subscriber<std::vector<T>>::subscribe(bool *is_success)
{
  if (shared_memory == nullptr || shared_memory->isDisconnected())
  {
    shared_memory->connect();
    if (shared_memory->isDisconnected())
    {
      *is_success = false;
      return std::vector<T>(0);
    }
    ring_buffer = new RingBuffer(shared_memory->getPtr());
    size_t element_size = ring_buffer->getElementSize();
    vector_size = element_size / sizeof(T);
  }
  int newest_buffer = ring_buffer->getNewestBufferNum();
  if (newest_buffer < 0)
  {
    *is_success = false;
    T * first_ptr = &((reinterpret_cast<T*>(ring_buffer->getDataList()))[current_reading_buffer*vector_size]);
    T * last_ptr  = &((reinterpret_cast<T*>(ring_buffer->getDataList()))[current_reading_buffer*vector_size + vector_size]);
    return std::vector<T>(first_ptr, last_ptr);
  }
 
  *is_success = true;
  current_reading_buffer = newest_buffer;
  T * first_ptr = &((reinterpret_cast<T*>(ring_buffer->getDataList()))[newest_buffer*vector_size]);
  T * last_ptr  = &((reinterpret_cast<T*>(ring_buffer->getDataList()))[newest_buffer*vector_size + vector_size - 1]);
  return std::vector<T>(first_ptr, last_ptr);
}
 
 
template <typename T>
bool
Subscriber<std::vector<T>>::waitFor(uint64_t timeout_usec)
{
  if (shared_memory->isDisconnected())
  {
    shared_memory->connect();
    if (shared_memory->isDisconnected())
    {
      return false;
    }
    ring_buffer = new RingBuffer(shared_memory->getPtr());
  }
 
  return ring_buffer->waitFor(timeout_usec);
}
 
 
template <typename T>
void
Subscriber<std::vector<T>>::setDataExpiryTime_us(uint64_t time_us)
{
  data_expiry_time_us = time_us;
  if (ring_buffer != nullptr)
  {
    ring_buffer->setDataExpiryTime_us(data_expiry_time_us);
  }
}
 
 
}
 
}
 
#endif /* __SHM_PS_VECTOR_LIB_H__ */