cyclic_raw_ethernet_socket_stub.h

Go to the documentation of this file.

 
#pragma once
 
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <deque>
#include <functional>
#include <memory>
#include <mutex>
#include <stdexcept>
#include <string>
#include <utility>
#include <vector>
 
#include "profinet/IRawEthernetSocket.h"
#include "profinet/protocol.h"
#include "profinet/util.h"
 
namespace profinet::cyclic::test
{
 
class CyclicRawEthernetSocketStub final : public IRawEthernetSocket
{
 public:
  struct ReceiveAction
  {
    bool throwException{false};
 
    Bytes data{};
  };
 
  void SetSendThrows(bool enabled)
  {
    sendThrows = enabled;
  }
 
  void SetSendExceptionMessage(std::string message)
  {
    sendExceptionMessage = std::move(message);
  }
 
  void SetRecvExceptionMessage(std::string message)
  {
    recvExceptionMessage = std::move(message);
  }
 
  void QueueTimeout()
  {
    receiveActions.push_back({false, {}});
  }
 
  void QueueReceive(Bytes data)
  {
    receiveActions.push_back({false, std::move(data)});
  }
 
  void QueueReceiveException()
  {
    receiveActions.push_back({true, {}});
  }
 
  [[nodiscard]] std::size_t SentFrameCount() const
  {
    std::lock_guard<std::mutex> lock(mutex);
    return sentFrames.size();
  }
 
  [[nodiscard]] const Bytes& SentFrame(std::size_t index) const
  {
    return sentFrames.at(index);
  }
 
  [[nodiscard]] std::chrono::milliseconds Timeout() const
  {
    return timeout;
  }
 
  void OnSend(std::function<void()> callback)
  {
    onSend = std::move(callback);
  }
 
  void Send(const Bytes& data) override
  {
    if (sendThrows)
    {
      throw std::runtime_error(sendExceptionMessage);
    }
    {
      std::lock_guard<std::mutex> lock(mutex);
      sentFrames.push_back(data);
    }
    if (onSend)
    {
      onSend();
    }
  }
 
  [[nodiscard]] Bytes Recv() override
  {
    if (receiveActions.empty())
    {
      return {};
    }
 
    ReceiveAction action = std::move(receiveActions.front());
    receiveActions.pop_front();
 
    if (action.throwException)
    {
      throw std::runtime_error(recvExceptionMessage);
    }
 
    return std::move(action.data);
  }
 
  void SetTimeout(std::chrono::milliseconds value) override
  {
    timeout = value;
  }
 
 private:
  std::deque<ReceiveAction> receiveActions;
 
  std::vector<Bytes> sentFrames;
 
  bool sendThrows{false};
 
  std::chrono::milliseconds timeout{0};
 
  std::string sendExceptionMessage{"send failure"};
 
  std::function<void()> onSend;
 
  std::string recvExceptionMessage{"receive failure"};
 
  mutable std::mutex mutex;
};
 
class BlockingRawEthernetSocket final : public IRawEthernetSocket
{
 public:
  void Send(const Bytes& frame) override
  {
  }
 
  Bytes Recv() override
  {
    BlockReceive();
 
    return {};
  }
 
  void SetTimeout(std::chrono::milliseconds timeout) override
  {
  }
 
  void BlockReceive()
  {
    std::unique_lock lock(mutex);
    receiveEntered = true;
    receiveEnteredCv.notify_all();
 
    receiveCv.wait(lock, [this]
    {
      return receiveReleased;
    });
  }
 
  void ReleaseReceive()
  {
    {
      std::lock_guard lock(mutex);
      receiveReleased = true;
    }
 
    receiveCv.notify_all();
  }
 
  void WaitUntilReceiveEntered()
  {
    std::unique_lock lock(mutex);
 
    receiveEnteredCv.wait(lock, [this]
    {
      return receiveEntered;
    });
  }
 
  // Receive() should call BlockReceive().
  // Other IRrawEthernetSocket methods implement the minimum
  // behavior required by CyclicController::Start()/TxCycle().
 
 private:
  std::mutex mutex;
  std::condition_variable receiveCv;
  std::condition_variable receiveEnteredCv;
 
  bool receiveEntered{false};
  bool receiveReleased{false};
};
 
struct LifetimeTrackingSocketState
{
  std::mutex mutex;
  std::condition_variable receiveCv;
  std::condition_variable receiveEnteredCv;
  std::condition_variable receiveExitedCv;
 
  bool receiveEntered{false};
  bool receiveExited{false};
  bool releaseReceive{false};
 
  std::atomic<bool> destroyed{false};
  std::atomic<bool> destroyedWhileReceiving{false};
  std::atomic<bool> receiveActive{false};
};
 
class LifetimeTrackingRawEthernetSocket final : public IRawEthernetSocket
{
 public:
  explicit LifetimeTrackingRawEthernetSocket(
      std::shared_ptr<LifetimeTrackingSocketState> state)
      : state(std::move(state))
  {
  }
 
  ~LifetimeTrackingRawEthernetSocket() override
  {
    if (state->receiveActive.load(std::memory_order_acquire))
    {
      state->destroyedWhileReceiving.store(
          true, std::memory_order_release);
    }
 
    state->destroyed.store(true, std::memory_order_release);
  }
 
  void Send(const Bytes&) override
  {
  }
 
  Bytes Recv() override
  {
    state->receiveActive.store(true, std::memory_order_release);
 
    {
      std::lock_guard lock(state->mutex);
      state->receiveEntered = true;
    }
 
    state->receiveEnteredCv.notify_one();
 
    {
      std::unique_lock lock(state->mutex);
      state->receiveCv.wait(lock, [this]
      {
        return state->releaseReceive;
      });
    }
 
    state->receiveActive.store(false, std::memory_order_release);
 
    {
      std::lock_guard lock(state->mutex);
      state->receiveExited = true;
    }
 
    state->receiveExitedCv.notify_one();
 
    return {};
  }
 
  void SetTimeout(std::chrono::milliseconds) override
  {
  }
 
  void WaitUntilReceiveEntered()
  {
    std::unique_lock lock(state->mutex);
 
    state->receiveEnteredCv.wait(lock, [this]
    {
      return state->receiveEntered;
    });
  }
 
  void WaitUntilReceiveExited()
  {
    std::unique_lock lock(state->mutex);
 
    state->receiveExitedCv.wait(lock, [this]
    {
      return state->receiveExited;
    });
  }
 
  void ReleaseReceive()
  {
    {
      std::lock_guard lock(state->mutex);
      state->releaseReceive = true;
    }
 
    state->receiveCv.notify_one();
  }
 
 private:
  std::shared_ptr<LifetimeTrackingSocketState> state;
};
struct BlockingReceiveState
{
  std::mutex mutex;
  std::condition_variable receiveEnteredCv;
  std::condition_variable releaseReceiveCv;
 
  bool receiveEntered{false};
  bool releaseReceive{false};
};
} // namespace profinet::cyclic::test