Test.h

#pragma once
 
#include <QThread>
 
#include "AbstractCommunicationInterfaces.h"
#include "Messages.h"
 
namespace knowDataTransfert::Test
{
  class SendingCommunicationInterface;
  class ReceivingCommunicationInterface;
 
  class CommunicationCenter : public QThread
  {
  public:
    CommunicationCenter() { start(); }
    ~CommunicationCenter()
    {
      m_running = false;
      m_cv.wakeOne();
      wait();
    }
    void add(const QString& _communicationNamespace, ReceivingCommunicationInterface* _receiver)
    {
      m_receivers[_communicationNamespace].append(_receiver);
    }
    void add(const QString& _communicationNamespace, SendingCommunicationInterface* _sender)
    {
      m_senders[_communicationNamespace] = _sender;
    }
    void add(const Data& _data, void* _sender, const QString& _communicationNamespace)
    {
      QMutexLocker l(&m_mutex);
      m_dmis.append({_communicationNamespace, _data, _sender});
      m_cv.wakeOne();
    }
    void add(const DataControl& _data, void* _sender, const QString& _communicationNamespace)
    {
      QMutexLocker l(&m_mutex);
      m_dcmis.append({_communicationNamespace, _data, _sender});
      m_cv.wakeOne();
    }
  protected:
    void run() override;
  private:
    template<typename _MsgT_>
    struct MessageInfo
    {
      QString communicationNamespace;
      _MsgT_ msg;
      void* sender;
    };
 
    using DataMessageInfo = MessageInfo<Data>;
    using DataControlMessageInfo = MessageInfo<DataControl>;
 
    int m_fault_generator = 0;
    QList<DataMessageInfo> m_dmis;
    QList<DataControlMessageInfo> m_dcmis;
    QHash<QString, QList<ReceivingCommunicationInterface*>> m_receivers;
    QHash<QString, SendingCommunicationInterface*> m_senders;
    QMutex m_mutex;
    QWaitCondition m_cv;
    bool m_running = true;
  };
 
  class SendingCommunicationInterface : public AbstractSendingCommunicationInterface
  {
  public:
    SendingCommunicationInterface(const QString& _communicationNamespace,
                                  CommunicationCenter* _center)
        : AbstractSendingCommunicationInterface(_communicationNamespace), m_center(_center)
    {
      m_center->add(_communicationNamespace, this);
    }
    void initialise() override {}
    void send(const Data& _data) override;
    void send(const DataControl& _dataControl) override;
    using AbstractSendingCommunicationInterface::handle;
  private:
    int m_fault_generator = 1;
    CommunicationCenter* m_center;
  };
 
  class ReceivingCommunicationInterface : public AbstractReceivingCommunicationInterface
  {
  public:
    ReceivingCommunicationInterface(const QString& _communicationNamespace,
                                    CommunicationCenter* _center)
        : AbstractReceivingCommunicationInterface(_communicationNamespace), m_center(_center)
    {
      m_center->add(_communicationNamespace, this);
    }
    void initialise() override {}
    void send(const DataControl& _dataControl) override;
    using AbstractReceivingCommunicationInterface::handle;
  private:
    CommunicationCenter* m_center;
  };
 
  void CommunicationCenter::run()
  {
    while(m_running)
    {
      QList<DataMessageInfo> dmis;
      QList<DataControlMessageInfo> dcmis;
      {
        QMutexLocker l(&m_mutex);
        std::swap(dmis, m_dmis);
        std::swap(dcmis, m_dcmis);
      }
      for(const DataMessageInfo& dmi : dmis)
      {
        for(ReceivingCommunicationInterface* i : m_receivers[dmi.communicationNamespace])
        {
          if(m_fault_generator % 3 != 0)
          {
            i->handle(dmi.msg);
          }
          ++m_fault_generator;
        }
      }
 
      for(const DataControlMessageInfo& dcmi : dcmis)
      {
        for(ReceivingCommunicationInterface* i : m_receivers[dcmi.communicationNamespace])
        {
          if(i != dcmi.sender)
          {
            i->handle(dcmi.msg);
          }
        }
        if(m_senders[dcmi.communicationNamespace] != dcmi.sender)
        {
          m_senders[dcmi.communicationNamespace]->handle(dcmi.msg);
        }
      }
 
      QMutexLocker l(&m_mutex);
      m_cv.wait(&m_mutex);
    }
  }
 
  void SendingCommunicationInterface::send(const Data& _data)
  {
    m_center->add(_data, this, communicationNamespace());
  }
 
  void SendingCommunicationInterface::send(const DataControl& _dataControl)
  {
    m_center->add(_dataControl, this, communicationNamespace());
  }
 
  void ReceivingCommunicationInterface::send(const DataControl& _dataControl)
  {
    m_center->add(_dataControl, this, communicationNamespace());
  }
 
  class CommunicationInterfaceFactory : public AbstractCommunicationInterfaceFactory
  {
  public:
    CommunicationInterfaceFactory(CommunicationCenter* _center) : m_center(_center) {}
    AbstractSendingCommunicationInterface*
      createSendingInterface(const QString& _communicationNamespace) override
    {
      return new SendingCommunicationInterface(_communicationNamespace, m_center);
    }
    AbstractReceivingCommunicationInterface*
      createReceivingInterface(const QString& _communicationNamespace) override
    {
      return new ReceivingCommunicationInterface(_communicationNamespace, m_center);
    }
  private:
    CommunicationCenter* m_center;
  };
} // namespace knowDataTransfert::Test