eive/fsfw
eive/fsfw
15
1
Fork 0
forked from fsfw/fsfw
Code Issues Pull Requests Releases 1 Wiki Activity

Update FSFW from upstream #71

Closed
muellerr wants to merge 1112 commits from development into eive/develop
pull from: development
merge into: eive:eive/develop
Conversation 0 Commits 1112 Files Changed 1041 +464 -462
27 changed files with 464 additions and 462 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit b20e8a9679 - Show all commits

3
src/fsfw/cfdp/CMakeLists.txt
View File

@ -1,4 +1,5 @@
target_sources(${LIB_FSFW_NAME} PRIVATE CfdpHandler.cpp CfdpMessage.cpp)
target_sources(${LIB_FSFW_NAME} PRIVATE CfdpMessage.cpp CfdpRouter.cpp)
# CfdpDistributor.cpp CfdpHandler.cpp
add_subdirectory(pdu)
add_subdirectory(tlv)

144
src/fsfw/cfdp/CfdpDistributor.cpp Normal file
View File

@ -0,0 +1,144 @@
//#include "CfdpRouter.h"
//#include "fsfw/tcdistribution/CcsdsDistributorIF.h"
//#include "fsfw/tmtcpacket/cfdp/CfdpPacketStored.h"
//
//#ifndef FSFW_CFDP_DISTRIBUTOR_DEBUGGING
//#define FSFW_CFDP_DISTRIBUTOR_DEBUGGING 1
//#endif
//
// CfdpHandler::CfdpHandler(uint16_t setApid, object_id_t setObjectId,
// object_id_t setPacketSource)
// : TcDistributorBase(setObjectId),
// apid(setApid),
// checker(setApid),
// tcStatus(RETURN_FAILED),
// packetSource(setPacketSource) {}
//
// CfdpHandler::~CfdpHandler() = default;
//
// ReturnValue_t CfdpHandler::selectDestination(MessageQueueId_t& destId) {
// //#if FSFW_CFDP_DISTRIBUTOR_DEBUGGING == 1
// // store_address_t storeId = this->currentMessage.getStorageId();
// //#if FSFW_CPP_OSTREAM_ENABLED == 1
// // sif::debug << "CFDPDistributor::handlePacket received: " << storeId.poolIndex << ", "
// // << storeId.packetIndex << std::endl;
// //#else
// // sif::printDebug("CFDPDistributor::handlePacket received: %d, %d\n", storeId.poolIndex,
// // storeId.packetIndex);
// //#endif
// //#endif
// // auto queueMapIt = this->queueMap.end();
// // if (this->currentPacket == nullptr) {
// // return queueMapIt;
// // }
// // this->currentPacket->setStoreAddress(this->currentMessage.getStorageId());
// // if (currentPacket->getFullData() != nullptr) {
// // tcStatus = checker.checkPacket(*currentPacket, currentPacket->getFullPacketLen());
// // if (tcStatus != HasReturnvaluesIF::RETURN_OK) {
// //#if FSFW_VERBOSE_LEVEL >= 1
// //#if FSFW_CPP_OSTREAM_ENABLED == 1
// // sif::debug << "CFDPDistributor::handlePacket: Packet format invalid, code "
// // << static_cast<int>(tcStatus) << std::endl;
// //#else
// // sif::printDebug("CFDPDistributor::handlePacket: Packet format invalid, code %d\n",
// // static_cast<int>(tcStatus));
// //#endif
// //#endif
// // }
// // queueMapIt = this->queueMap.find(0);
// // } else {
// // tcStatus = PACKET_LOST;
// // }
// //
// // if (queueMapIt == this->queueMap.end()) {
// // tcStatus = DESTINATION_NOT_FOUND;
// //#if FSFW_VERBOSE_LEVEL >= 1
// //#if FSFW_CPP_OSTREAM_ENABLED == 1
// // sif::debug << "CFDPDistributor::handlePacket: Destination not found" << std::endl;
// //#else
// // sif::printDebug("CFDPDistributor::handlePacket: Destination not found\n");
// //#endif /* !FSFW_CPP_OSTREAM_ENABLED == 1 */
// //#endif
// // }
// //
// // if (tcStatus != RETURN_OK) {
// // return this->queueMap.end();
// // } else {
// // return queueMapIt;
// // }
// return HasReturnvaluesIF::RETURN_OK;
//}
//
// ReturnValue_t CfdpHandler::registerHandler(AcceptsTelecommandsIF* handler) {
// // uint16_t handlerId =
// // handler->getIdentifier(); // should be 0, because CfdpHandler does not set a set a
// // service-ID
// //#if FSFW_CFDP_DISTRIBUTOR_DEBUGGING == 1
// //#if FSFW_CPP_OSTREAM_ENABLED == 1
// // sif::info << "CFDPDistributor::registerHandler: Handler ID: " << static_cast<int>(handlerId)
// // << std::endl;
// //#else
// // sif::printInfo("CFDPDistributor::registerHandler: Handler ID: %d\n",
// // static_cast<int>(handlerId));
// //#endif
// //#endif
// // MessageQueueId_t queue = handler->getRequestQueue();
// // auto returnPair = queueMap.emplace(handlerId, queue);
// // if (not returnPair.second) {
// //#if FSFW_VERBOSE_LEVEL >= 1
// //#if FSFW_CPP_OSTREAM_ENABLED == 1
// // sif::error << "CFDPDistributor::registerHandler: Service ID already"
// // " exists in map"
// // << std::endl;
// //#else
// // sif::printError("CFDPDistributor::registerHandler: Service ID already exists in map\n");
// //#endif
// //#endif
// // return SERVICE_ID_ALREADY_EXISTS;
// // }
// return HasReturnvaluesIF::RETURN_OK;
//}
//
// MessageQueueId_t CfdpHandler::getRequestQueue() const { return tcQueue->getId(); }
//
//// ReturnValue_t CFDPDistributor::callbackAfterSending(ReturnValue_t queueStatus) {
//// if (queueStatus != RETURN_OK) {
//// tcStatus = queueStatus;
//// }
//// if (tcStatus != RETURN_OK) {
//// this->verifyChannel.sendFailureReport(tc_verification::ACCEPTANCE_FAILURE,
//// currentPacket, tcStatus);
//// // A failed packet is deleted immediately after reporting,
//// // otherwise it will block memory.
//// currentPacket->deletePacket();
//// return RETURN_FAILED;
//// } else {
//// this->verifyChannel.sendSuccessReport(tc_verification::ACCEPTANCE_SUCCESS,
//// currentPacket);
//// return RETURN_OK;
//// }
//// }
//
// uint32_t CfdpHandler::getIdentifier() const { return this->apid; }
//
// ReturnValue_t CfdpHandler::initialize() {
// // currentPacket = new CfdpPacketStored();
// // if (currentPacket == nullptr) {
// // // Should not happen, memory allocation failed!
// // return ObjectManagerIF::CHILD_INIT_FAILED;
// // }
// //
// // auto* ccsdsDistributor = ObjectManager::instance()->get<CcsdsDistributorIF>(packetSource);
// // if (ccsdsDistributor == nullptr) {
// //#if FSFW_CPP_OSTREAM_ENABLED == 1
// // sif::error << "CFDPDistributor::initialize: Packet source invalid" << std::endl;
// // sif::error << " Make sure it exists and implements CCSDSDistributorIF!" << std::endl;
// //#else
// // sif::printError("CFDPDistributor::initialize: Packet source invalid\n");
// // sif::printError("Make sure it exists and implements CcsdsDistributorIF\n");
// //#endif
// // return RETURN_FAILED;
// // }
// // return ccsdsDistributor->registerApplication(this);
// return HasReturnvaluesIF::RETURN_OK;
//}

111
src/fsfw/cfdp/CfdpHandler.cpp
View File

@ -1,57 +1,54 @@
#include "fsfw/cfdp/CfdpHandler.h"
#include "fsfw/cfdp/CfdpMessage.h"
#include "fsfw/ipc/CommandMessage.h"
#include "fsfw/ipc/QueueFactory.h"
#include "fsfw/objectmanager/ObjectManager.h"
#include "fsfw/storagemanager/storeAddress.h"
#include "fsfw/tmtcservices/AcceptsTelemetryIF.h"
object_id_t CfdpHandler::packetSource = 0;
object_id_t CfdpHandler::packetDestination = 0;
CfdpHandler::CfdpHandler(object_id_t setObjectId, CfdpDistributor* dist)
: SystemObject(setObjectId) {
requestQueue = QueueFactory::instance()->createMessageQueue(CFDP_HANDLER_MAX_RECEPTION);
distributor = dist;
}
CfdpHandler::~CfdpHandler() = default;
ReturnValue_t CfdpHandler::initialize() {
ReturnValue_t result = SystemObject::initialize();
if (result != RETURN_OK) {
return result;
}
this->distributor->registerHandler(this);
return HasReturnvaluesIF::RETURN_OK;
}
ReturnValue_t CfdpHandler::handleRequest(store_address_t storeId) {
#if FSFW_VERBOSE_LEVEL >= 1
#if FSFW_CPP_OSTREAM_ENABLED == 1
sif::debug << "CFDPHandler::handleRequest" << std::endl;
#else
sif::printDebug("CFDPHandler::handleRequest\n");
#endif /* !FSFW_CPP_OSTREAM_ENABLED == 1 */
#endif
// TODO read out packet from store using storeId
return RETURN_OK;
}
ReturnValue_t CfdpHandler::performOperation(uint8_t opCode) {
ReturnValue_t status = RETURN_OK;
CommandMessage currentMessage;
for (status = this->requestQueue->receiveMessage(&currentMessage); status == RETURN_OK;
status = this->requestQueue->receiveMessage(&currentMessage)) {
store_address_t storeId = CfdpMessage::getStoreId(&currentMessage);
this->handleRequest(storeId);
}
return RETURN_OK;
}
uint32_t CfdpHandler::getIdentifier() const { return 0; }
MessageQueueId_t CfdpHandler::getRequestQueue() const { return this->requestQueue->getId(); }
//#include "fsfw/cfdp/CfdpHandler.h"
//
//#include "fsfw/cfdp/CfdpMessage.h"
//#include "fsfw/ipc/CommandMessage.h"
//#include "fsfw/ipc/QueueFactory.h"
//#include "fsfw/objectmanager/ObjectManager.h"
//#include "fsfw/storagemanager/storeAddress.h"
//#include "fsfw/tmtcservices/AcceptsTelemetryIF.h"
//
// CfdpHandler::CfdpHandler(object_id_t setObjectId, CfdpHandler* dist)
// : SystemObject(setObjectId) {
// requestQueue = QueueFactory::instance()->createMessageQueue(CFDP_HANDLER_MAX_RECEPTION);
// distributor = dist;
//}
//
// CfdpHandler::~CfdpHandler() = default;
//
// ReturnValue_t CfdpHandler::initialize() {
// ReturnValue_t result = SystemObject::initialize();
// if (result != RETURN_OK) {
// return result;
// }
// this->distributor->registerHandler(this);
// return HasReturnvaluesIF::RETURN_OK;
//}
//
// ReturnValue_t CfdpHandler::handleRequest(store_address_t storeId) {
//#if FSFW_VERBOSE_LEVEL >= 1
//#if FSFW_CPP_OSTREAM_ENABLED == 1
// sif::debug << "CFDPHandler::handleRequest" << std::endl;
//#else
// sif::printDebug("CFDPHandler::handleRequest\n");
//#endif /* !FSFW_CPP_OSTREAM_ENABLED == 1 */
//#endif
//
// // TODO read out packet from store using storeId
//
// return RETURN_OK;
//}
//
// ReturnValue_t CfdpHandler::performOperation(uint8_t opCode) {
// ReturnValue_t status = RETURN_OK;
// CommandMessage currentMessage;
// for (status = this->requestQueue->receiveMessage(&currentMessage); status == RETURN_OK;
// status = this->requestQueue->receiveMessage(&currentMessage)) {
// store_address_t storeId = CfdpMessage::getStoreId(&currentMessage);
// this->handleRequest(storeId);
// }
// return RETURN_OK;
//}
//
// uint32_t CfdpHandler::getIdentifier() const { return 0; }
//
// MessageQueueId_t CfdpHandler::getRequestQueue() const { return this->requestQueue->getId(); }

126
src/fsfw/cfdp/CfdpHandler.h
View File

@ -1,63 +1,63 @@
#ifndef FSFW_CFDP_CFDPHANDLER_H_
#define FSFW_CFDP_CFDPHANDLER_H_
#include "fsfw/ipc/MessageQueueIF.h"
#include "fsfw/objectmanager/SystemObject.h"
#include "fsfw/returnvalues/HasReturnvaluesIF.h"
#include "fsfw/tasks/ExecutableObjectIF.h"
#include "fsfw/tcdistribution/CfdpDistributor.h"
#include "fsfw/tmtcservices/AcceptsTelecommandsIF.h"
namespace Factory {
void setStaticFrameworkObjectIds();
}
class CfdpHandler : public ExecutableObjectIF,
public AcceptsTelecommandsIF,
public SystemObject,
public HasReturnvaluesIF {
friend void(Factory::setStaticFrameworkObjectIds)();
public:
CfdpHandler(object_id_t setObjectId, CfdpDistributor* distributor);
/**
* The destructor is empty.
*/
virtual ~CfdpHandler();
virtual ReturnValue_t handleRequest(store_address_t storeId);
virtual ReturnValue_t initialize() override;
virtual uint32_t getIdentifier() const override;
MessageQueueId_t getRequestQueue() const override;
ReturnValue_t performOperation(uint8_t opCode) override;
protected:
/**
* This is a complete instance of the telecommand reception queue
* of the class. It is initialized on construction of the class.
*/
MessageQueueIF* requestQueue = nullptr;
CfdpDistributor* distributor = nullptr;
/**
* The current CFDP packet to be processed.
* It is deleted after handleRequest was executed.
*/
CfdpPacketStored currentPacket;
static object_id_t packetSource;
static object_id_t packetDestination;
private:
/**
* This constant sets the maximum number of packets accepted per call.
* Remember that one packet must be completely handled in one
* #handleRequest call.
*/
static const uint8_t CFDP_HANDLER_MAX_RECEPTION = 100;
};
#endif /* FSFW_CFDP_CFDPHANDLER_H_ */
//#ifndef FSFW_CFDP_CFDPHANDLER_H_
//#define FSFW_CFDP_CFDPHANDLER_H_
//
//#include "fsfw/ipc/MessageQueueIF.h"
//#include "fsfw/objectmanager/SystemObject.h"
//#include "fsfw/returnvalues/HasReturnvaluesIF.h"
//#include "fsfw/storagemanager/storeAddress.h"
//#include "fsfw/tasks/ExecutableObjectIF.h"
//#include "fsfw/tmtcservices/AcceptsTelecommandsIF.h"
//
// namespace Factory {
// void setStaticFrameworkObjectIds();
//}
//
// class CfdpHandler : public ExecutableObjectIF,
// public AcceptsTelecommandsIF,
// public SystemObject,
// public HasReturnvaluesIF {
// friend void(Factory::setStaticFrameworkObjectIds)();
//
// public:
// CfdpHandler(object_id_t setObjectId, CfdpHandler* distributor);
// /**
// * The destructor is empty.
// */
// virtual ~CfdpHandler();
//
// virtual ReturnValue_t handleRequest(store_address_t storeId);
//
// virtual ReturnValue_t initialize() override;
// virtual uint32_t getIdentifier() const override;
// MessageQueueId_t getRequestQueue() const override;
// ReturnValue_t performOperation(uint8_t opCode) override;
//
// protected:
// /**
// * This is a complete instance of the telecommand reception queue
// * of the class. It is initialized on construction of the class.
// */
// MessageQueueIF* requestQueue = nullptr;
//
// CfdpHandler* distributor = nullptr;
//
// /**
// * The current CFDP packet to be processed.
// * It is deleted after handleRequest was executed.
// */
// CfdpPacketStored currentPacket;
//
// static object_id_t packetSource;
//
// static object_id_t packetDestination;
//
// private:
// /**
// * This constant sets the maximum number of packets accepted per call.
// * Remember that one packet must be completely handled in one
// * #handleRequest call.
// */
// static const uint8_t CFDP_HANDLER_MAX_RECEPTION = 100;
//};
//
//#endif /* FSFW_CFDP_CFDPHANDLER_H_ */

65
src/fsfw/cfdp/CfdpRouter.cpp Normal file
View File

@ -0,0 +1,65 @@
#include "CfdpRouter.h"
#include <algorithm>
#include "fsfw/tcdistribution/definitions.h"
CfdpRouter::CfdpRouter(CfdpRouterCfg cfg) : TcDistributorBase(cfg.objectId), tmQueue(cfg.tmQueue) {}
ReturnValue_t CfdpRouter::registerTmSink(cfdp::EntityId address, AcceptsTelemetryIF& tmDest) {
for (const auto& dest : tmDestinations) {
if (dest.id == address) {
return HasReturnvaluesIF::RETURN_FAILED;
}
}
tmDestinations.emplace_back(address, tmDest.getName(), tmDest.getReportReceptionQueue());
return HasReturnvaluesIF::RETURN_OK;
}
ReturnValue_t CfdpRouter::registerTcDestination(cfdp::EntityId address,
AcceptsTelecommandsIF& tcDest) {
for (const auto& dest : tcDestinations) {
if (dest.id == address) {
return HasReturnvaluesIF::RETURN_FAILED;
}
}
tcDestinations.emplace_back(address, tcDest.getName(), tcDest.getRequestQueue());
return HasReturnvaluesIF::RETURN_OK;
}
ReturnValue_t CfdpRouter::selectDestination(MessageQueueId_t& destId) { return 0; }
const char* CfdpRouter::getName() const { return "CFDP Router"; }
uint32_t CfdpRouter::getIdentifier() const { return 0; }
MessageQueueId_t CfdpRouter::getRequestQueue() const { return tcQueue->getId(); }
MessageQueueId_t CfdpRouter::getReportReceptionQueue(uint8_t virtualChannel) {
return tmQueue.getId();
}
ReturnValue_t CfdpRouter::performOperation(uint8_t opCode) {
lastTcError = TcDistributorBase::performOperation(opCode);
lastTmError = performTmHandling();
if (lastTcError != HasReturnvaluesIF::RETURN_OK or lastTmError != HasReturnvaluesIF::RETURN_OK) {
return HasReturnvaluesIF::RETURN_FAILED;
}
return HasReturnvaluesIF::RETURN_OK;
}
ReturnValue_t CfdpRouter::performTmHandling() {
ReturnValue_t status;
ReturnValue_t result = HasReturnvaluesIF::RETURN_OK;
for (status = tmQueue.receiveMessage(&currentMessage); status == RETURN_OK;
status = tmQueue.receiveMessage(&currentMessage)) {
ReturnValue_t packetResult = handlePacket();
if (packetResult != HasReturnvaluesIF::RETURN_OK) {
result = packetResult;
triggerEvent(tmtcdistrib::HANDLE_PACKET_FAILED, packetResult, ccsds::PacketType::TM);
}
}
if (status == MessageQueueIF::EMPTY) {
return result;
}
return result;
}

74
src/fsfw/cfdp/CfdpRouter.h Normal file
View File

@ -0,0 +1,74 @@
#ifndef FSFW_TCDISTRIBUTION_CFDPDISTRIBUTOR_H_
#define FSFW_TCDISTRIBUTION_CFDPDISTRIBUTOR_H_
#include <utility>
#include <vector>
#include "CfdpRouterIF.h"
#include "fsfw/returnvalues/HasReturnvaluesIF.h"
#include "fsfw/tcdistribution/CfdpPacketChecker.h"
#include "fsfw/tcdistribution/TcDistributorBase.h"
#include "fsfw/tmtcpacket/cfdp/CfdpPacketStored.h"
#include "fsfw/tmtcservices/AcceptsTelecommandsIF.h"
#include "fsfw/tmtcservices/VerificationReporter.h"
struct CfdpRouterCfg {
CfdpRouterCfg(object_id_t objectId, MessageQueueIF& tmQueue)
: objectId(objectId), tmQueue(tmQueue) {}
object_id_t objectId;
MessageQueueIF& tmQueue;
};
/**
* This will be the primary component to perform PDU forwading procedures. This includes forwarding
* CFDP TC packets to registered source or destination handlers, and forwarding all telemetry
* generated by them to registered TM sinks.
* @ingroup tc_distribution
*/
class CfdpRouter : public TcDistributorBase,
public CfdpRouterIF,
public AcceptsTelecommandsIF,
public AcceptsTelemetryIF {
public:
/**
* The ctor passes @c set_apid to the checker class and calls the
* TcDistribution ctor with a certain object id.
* @param setApid The APID of this receiving Application.
* @param setObjectId Object ID of the distributor itself
* @param setPacketSource Object ID of the source of TC packets.
* Must implement CcsdsDistributorIF.
*/
explicit CfdpRouter(CfdpRouterCfg cfg);
ReturnValue_t performOperation(uint8_t opCode) override;
ReturnValue_t performTmHandling();
[[nodiscard]] const char* getName() const override;
[[nodiscard]] uint32_t getIdentifier() const override;
[[nodiscard]] MessageQueueId_t getRequestQueue() const override;
MessageQueueId_t getReportReceptionQueue(uint8_t virtualChannel) override;
ReturnValue_t registerTmSink(cfdp::EntityId address, AcceptsTelemetryIF& tmDest) override;
ReturnValue_t registerTcDestination(cfdp::EntityId address,
AcceptsTelecommandsIF& tcDest) override;
protected:
struct EntityInfo {
EntityInfo(cfdp::EntityId id, const char* name, MessageQueueId_t queueId)
: id(std::move(id)), name(name), queueId(queueId) {}
cfdp::EntityId id;
const char* name;
MessageQueueId_t queueId;
};
ReturnValue_t lastTcError = HasReturnvaluesIF::RETURN_OK;
ReturnValue_t lastTmError = HasReturnvaluesIF::RETURN_OK;
// I don't think a regular OBSW will have more than 1 or 2 of these destinations, so I think
// it is okay to accept the overhead here
std::vector<EntityInfo> tmDestinations;
std::vector<EntityInfo> tcDestinations;
MessageQueueIF& tmQueue;
ReturnValue_t selectDestination(MessageQueueId_t& destId) override;
private:
};
#endif /* FSFW_TCDISTRIBUTION_CFDPDISTRIBUTOR_H_ */

16
src/fsfw/tcdistribution/CfdpDistributorIF.h → src/fsfw/cfdp/CfdpRouterIF.h
View File

@ -1,26 +1,32 @@
#ifndef FSFW_TCDISTRIBUTION_CFDPDISTRIBUTORIF_H_
#define FSFW_TCDISTRIBUTION_CFDPDISTRIBUTORIF_H_
#include "../ipc/MessageQueueSenderIF.h"
#include "../tmtcservices/AcceptsTelecommandsIF.h"
#include "fsfw/cfdp/definitions.h"
#include "fsfw/cfdp/pdu/PduHeaderIF.h"
#include "fsfw/ipc/MessageQueueSenderIF.h"
#include "fsfw/tmtcservices/AcceptsTelecommandsIF.h"
#include "fsfw/tmtcservices/AcceptsTelemetryIF.h"
/**
* This interface allows CFDP Services to register themselves at a CFDP Distributor.
* @ingroup tc_distribution
*/
class CfdpDistributorIF {
class CfdpRouterIF {
public:
/**
* The empty virtual destructor.
*/
virtual ~CfdpDistributorIF() = default;
virtual ~CfdpRouterIF() = default;
virtual ReturnValue_t registerTmSink(cfdp::EntityId address, AcceptsTelemetryIF& tmDest) = 0;
/**
* With this method, Handlers can register themselves at the CFDP Distributor.
* @param handler A pointer to the registering Handler.
* @return - @c RETURN_OK on success,
* - @c RETURN_FAILED on failure.
*/
virtual ReturnValue_t registerHandler(AcceptsTelecommandsIF* handler) = 0;
virtual ReturnValue_t registerTcDestination(cfdp::EntityId address,
AcceptsTelecommandsIF& tcDest) = 0;
};
#endif /* FSFW_TCDISTRIBUTION_CFDPDISTRIBUTORIF_H_ */

8
src/fsfw/cfdp/pdu/VarLenField.cpp
View File

@ -118,3 +118,11 @@ ReturnValue_t cfdp::VarLenField::deSerialize(const uint8_t **buffer, size_t *siz
}
}
}
bool cfdp::VarLenField::operator<(const cfdp::VarLenField &other) const {
return getValue() < other.getValue();
}
bool cfdp::VarLenField::operator==(const cfdp::VarLenField &other) const {
return getValue() == other.getValue();
}

3
src/fsfw/cfdp/pdu/VarLenField.h
View File

@ -23,6 +23,9 @@ class VarLenField : public SerializeIF {
explicit VarLenField(UnsignedByteField<T> byteField);
VarLenField(cfdp::WidthInBytes width, size_t value);
bool operator==(const VarLenField &other) const;
bool operator<(const VarLenField &other) const;
ReturnValue_t setValue(cfdp::WidthInBytes, size_t value);
ReturnValue_t serialize(uint8_t **buffer, size_t *size, size_t maxSize,

2
src/fsfw/events/fwSubsystemIdRanges.h
View File

@ -19,7 +19,7 @@ enum : uint8_t {
HK = 73,
SYSTEM_MANAGER = 74,
SYSTEM_MANAGER_1 = 75,
TC_DISTRIBUTION = 76,
TMTC_DISTRIBUTION = 76,
SYSTEM_1 = 79,
PUS_SERVICE_1 = 80,
PUS_SERVICE_2 = 82,

10
src/fsfw/tcdistribution/CMakeLists.txt
View File

@ -1,10 +1,4 @@
target_sources(
${LIB_FSFW_NAME}
PRIVATE CcsdsDistributor.cpp
PusDistributor.cpp
TcDistributorBase.cpp
PusPacketChecker.cpp
TcPacketCheckCFDP.cpp
CfdpDistributor.cpp
CcsdsPacketChecker.cpp
CcsdsUnpacker.cpp)
PRIVATE CcsdsDistributor.cpp PusDistributor.cpp TcDistributorBase.cpp
PusPacketChecker.cpp TcPacketCheckCFDP.cpp CcsdsPacketChecker.cpp)

6
src/fsfw/tcdistribution/CcsdsDistributor.cpp
View File

@ -96,13 +96,13 @@ ReturnValue_t CcsdsDistributor::selectDestination(MessageQueueId_t& destId) {
void CcsdsDistributor::handlePacketCheckFailure(ReturnValue_t result) {
#if FSFW_VERBOSE_LEVEL >= 1
const char* reason = "Unknown reason";
if (result == tcdistrib::INVALID_CCSDS_VERSION) {
if (result == tmtcdistrib::INVALID_CCSDS_VERSION) {
reason = "Invalid CCSDS version";
} else if (result == tcdistrib::INCOMPLETE_PACKET) {
reason = "Size missmatch between CCSDS data length and packet length";
} else if (result == tcdistrib::INVALID_APID) {
} else if (result == tmtcdistrib::INVALID_APID) {
reason = "No valid handler APID found";
} else if (result == tcdistrib::INVALID_PACKET_TYPE) {
} else if (result == tmtcdistrib::INVALID_PACKET_TYPE) {
reason = "Invalid Packet Type TM detected";
}
#if FSFW_CPP_OSTREAM_ENABLED == 1

12
src/fsfw/tcdistribution/CcsdsDistributorIF.h
View File

@ -35,12 +35,12 @@ class CcsdsDistributorIF {
* @param info Contains all necessary info to register an application.
* @return
* - @c RETURN_OK on success,
* - @c RETURN_FAILED or tcdistrib error code on failure.
* - @c tcdistrib::INVALID_CCSDS_VERSION
* - @c tcdistrib::INVALID_APID No APID available to handle this packet
* - @c tcdistrib::INVALID_PACKET_TYPE Packet type TM detected
* - @c tcdistrib::INCORRECT_PRIMARY_HEADER Something other wrong with primary header
* - @c tcdistrib::INCOMPLETE_PACKET Size missmatch between data length field and actual
* - @c RETURN_FAILED or tmtcdistrib error code on failure.
* - @c tmtcdistrib::INVALID_CCSDS_VERSION
* - @c tmtcdistrib::INVALID_APID No APID available to handle this packet
* - @c tmtcdistrib::INVALID_PACKET_TYPE Packet type TM detected
* - @c tmtcdistrib::INCORRECT_PRIMARY_HEADER Something other wrong with primary header
* - @c tmtcdistrib::INCOMPLETE_PACKET Size missmatch between data length field and actual
* length
*/
virtual ReturnValue_t registerApplication(DestInfo info) = 0;

8
src/fsfw/tcdistribution/CcsdsPacketChecker.cpp
View File

@ -10,19 +10,19 @@ ReturnValue_t CcsdsPacketChecker::checkPacket(const SpacePacketReader& currentPa
size_t packetLen) {
if (checkApid) {
if (currentPacket.getApid() != apid) {
return tcdistrib::INVALID_APID;
return tmtcdistrib::INVALID_APID;
}
}
if (currentPacket.getVersion() != ccsdsVersion) {
return tcdistrib::INVALID_CCSDS_VERSION;
return tmtcdistrib::INVALID_CCSDS_VERSION;
}
if (currentPacket.getPacketType() != packetType) {
return tcdistrib::INVALID_PACKET_TYPE;
return tmtcdistrib::INVALID_PACKET_TYPE;
}
// This assumes that the getFullPacketLen version uses the space packet data length field
if (currentPacket.getFullPacketLen() != packetLen) {
return tcdistrib::INCOMPLETE_PACKET;
return tmtcdistrib::INCOMPLETE_PACKET;
}
return HasReturnvaluesIF::RETURN_OK;
}

57
src/fsfw/tcdistribution/CcsdsUnpacker.cpp
View File

@ -1,57 +0,0 @@
#include "CcsdsUnpacker.h"
#include "fsfw/tmtcservices/TmTcMessage.h"
CcsdsUnpacker::CcsdsUnpacker(MessageQueueIF& msgQueue, AcceptsTelecommandsIF& receiver,
StorageManagerIF& sourceStore)
: sourceStore(sourceStore), msgQueue(msgQueue), receiver(receiver) {
msgQueue.setDefaultDestination(receiver.getRequestQueue());
}
ReturnValue_t CcsdsUnpacker::performOperation(uint8_t operationCode) {
TmTcMessage msg;
ReturnValue_t result;
for (result = msgQueue.receiveMessage(&msg); result == HasReturnvaluesIF::RETURN_OK;
result = msgQueue.receiveMessage(&msg)) {
auto resultPair = sourceStore.getData(msg.getStorageId());
if (resultPair.first != HasReturnvaluesIF::RETURN_OK) {
continue;
}
if (resultPair.second.size() < 6) {
// TODO: This is a config error. Does it make sense to forward the message?
result = msgQueue.sendToDefault(&msg);
if (result != HasReturnvaluesIF::RETURN_OK) {
}
continue;
}
StorageManagerIF* tgtStore;
if (targetStore != nullptr) {
tgtStore = targetStore;
} else {
tgtStore = &sourceStore;
}
store_address_t newId;
uint8_t* ptr;
result = tgtStore->getFreeElement(&newId, resultPair.second.size(), &ptr);
if (result != HasReturnvaluesIF::RETURN_OK) {
// TODO: Implement error handling
}
std::memcpy(ptr, resultPair.second.data() + 6, resultPair.second.size() - 6);
result = sourceStore.deleteData(msg.getStorageId());
if (result != HasReturnvaluesIF::RETURN_OK) {
// TODO: Implement error handling (though this really should not happen)
}
TmTcMessage newMsg(newId);
result = msgQueue.sendToDefault(&newMsg);
if (result != HasReturnvaluesIF::RETURN_OK) {
}
}
return result;
}
void CcsdsUnpacker::setDifferentTargetStore(StorageManagerIF& otherTargetStore) {
targetStore = &otherTargetStore;
}
uint32_t CcsdsUnpacker::getIdentifier() const { return 0; }
MessageQueueId_t CcsdsUnpacker::getRequestQueue() const { return 0; }

24
src/fsfw/tcdistribution/CcsdsUnpacker.h
View File

@ -1,24 +0,0 @@
#ifndef FSFW_TCDISTRIBUTION_CCSDSUNPACKER_H
#define FSFW_TCDISTRIBUTION_CCSDSUNPACKER_H
#include "fsfw/storagemanager/StorageManagerIF.h"
#include "fsfw/tasks/ExecutableObjectIF.h"
#include "fsfw/tmtcservices/AcceptsTelecommandsIF.h"
class CcsdsUnpacker : public ExecutableObjectIF, public AcceptsTelecommandsIF {
public:
CcsdsUnpacker(MessageQueueIF& msgQueue, AcceptsTelecommandsIF& receiver,
StorageManagerIF& sourceStore);
void setDifferentTargetStore(StorageManagerIF& otherTargetStore);
ReturnValue_t performOperation(uint8_t operationCode) override;
uint32_t getIdentifier() const override;
MessageQueueId_t getRequestQueue() const override;
private:
StorageManagerIF& sourceStore;
StorageManagerIF* targetStore = nullptr;
MessageQueueIF& msgQueue;
AcceptsTelecommandsIF& receiver;
};
#endif // FSFW_TCDISTRIBUTION_CCSDSUNPACKER_H

146
src/fsfw/tcdistribution/CfdpDistributor.cpp
View File

@ -1,146 +0,0 @@
#include "fsfw/tcdistribution/CfdpDistributor.h"
#include "fsfw/objectmanager/ObjectManager.h"
#include "fsfw/tcdistribution/CcsdsDistributorIF.h"
#include "fsfw/tmtcpacket/cfdp/CfdpPacketStored.h"
#ifndef FSFW_CFDP_DISTRIBUTOR_DEBUGGING
#define FSFW_CFDP_DISTRIBUTOR_DEBUGGING 1
#endif
CfdpDistributor::CfdpDistributor(uint16_t setApid, object_id_t setObjectId,
object_id_t setPacketSource)
: TcDistributorBase(setObjectId),
apid(setApid),
checker(setApid),
tcStatus(RETURN_FAILED),
packetSource(setPacketSource) {}
CfdpDistributor::~CfdpDistributor() = default;
ReturnValue_t CfdpDistributor::selectDestination(MessageQueueId_t& destId) {
//#if FSFW_CFDP_DISTRIBUTOR_DEBUGGING == 1
// store_address_t storeId = this->currentMessage.getStorageId();
//#if FSFW_CPP_OSTREAM_ENABLED == 1
// sif::debug << "CFDPDistributor::handlePacket received: " << storeId.poolIndex << ", "
// << storeId.packetIndex << std::endl;
//#else
// sif::printDebug("CFDPDistributor::handlePacket received: %d, %d\n", storeId.poolIndex,
// storeId.packetIndex);
//#endif
//#endif
// auto queueMapIt = this->queueMap.end();
// if (this->currentPacket == nullptr) {
// return queueMapIt;
// }
// this->currentPacket->setStoreAddress(this->currentMessage.getStorageId());
// if (currentPacket->getFullData() != nullptr) {
// tcStatus = checker.checkPacket(*currentPacket, currentPacket->getFullPacketLen());
// if (tcStatus != HasReturnvaluesIF::RETURN_OK) {
//#if FSFW_VERBOSE_LEVEL >= 1
//#if FSFW_CPP_OSTREAM_ENABLED == 1
// sif::debug << "CFDPDistributor::handlePacket: Packet format invalid, code "
// << static_cast<int>(tcStatus) << std::endl;
//#else
// sif::printDebug("CFDPDistributor::handlePacket: Packet format invalid, code %d\n",
// static_cast<int>(tcStatus));
//#endif
//#endif
// }
// queueMapIt = this->queueMap.find(0);
// } else {
// tcStatus = PACKET_LOST;
// }
//
// if (queueMapIt == this->queueMap.end()) {
// tcStatus = DESTINATION_NOT_FOUND;
//#if FSFW_VERBOSE_LEVEL >= 1
//#if FSFW_CPP_OSTREAM_ENABLED == 1
// sif::debug << "CFDPDistributor::handlePacket: Destination not found" << std::endl;
//#else
// sif::printDebug("CFDPDistributor::handlePacket: Destination not found\n");
//#endif /* !FSFW_CPP_OSTREAM_ENABLED == 1 */
//#endif
// }
//
// if (tcStatus != RETURN_OK) {
// return this->queueMap.end();
// } else {
// return queueMapIt;
// }
return HasReturnvaluesIF::RETURN_OK;
}
ReturnValue_t CfdpDistributor::registerHandler(AcceptsTelecommandsIF* handler) {
// uint16_t handlerId =
// handler->getIdentifier(); // should be 0, because CfdpHandler does not set a set a
// service-ID
//#if FSFW_CFDP_DISTRIBUTOR_DEBUGGING == 1
//#if FSFW_CPP_OSTREAM_ENABLED == 1
// sif::info << "CFDPDistributor::registerHandler: Handler ID: " << static_cast<int>(handlerId)
// << std::endl;
//#else
// sif::printInfo("CFDPDistributor::registerHandler: Handler ID: %d\n",
// static_cast<int>(handlerId));
//#endif
//#endif
// MessageQueueId_t queue = handler->getRequestQueue();
// auto returnPair = queueMap.emplace(handlerId, queue);
// if (not returnPair.second) {
//#if FSFW_VERBOSE_LEVEL >= 1
//#if FSFW_CPP_OSTREAM_ENABLED == 1
// sif::error << "CFDPDistributor::registerHandler: Service ID already"
// " exists in map"
// << std::endl;
//#else
// sif::printError("CFDPDistributor::registerHandler: Service ID already exists in map\n");
//#endif
//#endif
// return SERVICE_ID_ALREADY_EXISTS;
// }
return HasReturnvaluesIF::RETURN_OK;
}
MessageQueueId_t CfdpDistributor::getRequestQueue() const { return tcQueue->getId(); }
// ReturnValue_t CFDPDistributor::callbackAfterSending(ReturnValue_t queueStatus) {
// if (queueStatus != RETURN_OK) {
// tcStatus = queueStatus;
// }
// if (tcStatus != RETURN_OK) {
// this->verifyChannel.sendFailureReport(tc_verification::ACCEPTANCE_FAILURE,
// currentPacket, tcStatus);
// // A failed packet is deleted immediately after reporting,
// // otherwise it will block memory.
// currentPacket->deletePacket();
// return RETURN_FAILED;
// } else {
// this->verifyChannel.sendSuccessReport(tc_verification::ACCEPTANCE_SUCCESS,
// currentPacket);
// return RETURN_OK;
// }
// }
uint32_t CfdpDistributor::getIdentifier() const { return this->apid; }
ReturnValue_t CfdpDistributor::initialize() {
// currentPacket = new CfdpPacketStored();
// if (currentPacket == nullptr) {
// // Should not happen, memory allocation failed!
// return ObjectManagerIF::CHILD_INIT_FAILED;
// }
//
// auto* ccsdsDistributor = ObjectManager::instance()->get<CcsdsDistributorIF>(packetSource);
// if (ccsdsDistributor == nullptr) {
//#if FSFW_CPP_OSTREAM_ENABLED == 1
// sif::error << "CFDPDistributor::initialize: Packet source invalid" << std::endl;
// sif::error << " Make sure it exists and implements CCSDSDistributorIF!" << std::endl;
//#else
// sif::printError("CFDPDistributor::initialize: Packet source invalid\n");
// sif::printError("Make sure it exists and implements CcsdsDistributorIF\n");
//#endif
// return RETURN_FAILED;
// }
// return ccsdsDistributor->registerApplication(this);
return HasReturnvaluesIF::RETURN_OK;
}

71
src/fsfw/tcdistribution/CfdpDistributor.h
View File

@ -1,71 +0,0 @@
#ifndef FSFW_TCDISTRIBUTION_CFDPDISTRIBUTOR_H_
#define FSFW_TCDISTRIBUTION_CFDPDISTRIBUTOR_H_
#include <fsfw/tcdistribution/CfdpPacketChecker.h>
#include "../returnvalues/HasReturnvaluesIF.h"
#include "../tmtcpacket/cfdp/CfdpPacketStored.h"
#include "../tmtcservices/AcceptsTelecommandsIF.h"
#include "../tmtcservices/VerificationReporter.h"
#include "CfdpDistributorIF.h"
#include "TcDistributorBase.h"
/**
* This class accepts CFDP Telecommands and forwards them to Application
* services.
* @ingroup tc_distribution
*/
class CfdpDistributor : public TcDistributorBase,
public CfdpDistributorIF,
public AcceptsTelecommandsIF {
public:
/**
* The ctor passes @c set_apid to the checker class and calls the
* TcDistribution ctor with a certain object id.
* @param setApid The APID of this receiving Application.
* @param setObjectId Object ID of the distributor itself
* @param setPacketSource Object ID of the source of TC packets.
* Must implement CcsdsDistributorIF.
*/
CfdpDistributor(uint16_t setApid, object_id_t setObjectId, object_id_t setPacketSource);
/**
* The destructor is empty.
*/
~CfdpDistributor() override;
ReturnValue_t registerHandler(AcceptsTelecommandsIF* handler) override;
MessageQueueId_t getRequestQueue() const override;
ReturnValue_t initialize() override;
uint32_t getIdentifier() const override;
protected:
uint16_t apid;
/**
* The currently handled packet is stored here.
*/
CfdpPacketStored* currentPacket = nullptr;
CfdpPacketChecker checker;
/**
* With this variable, the current check status is stored to generate
* acceptance messages later.
*/
ReturnValue_t tcStatus;
const object_id_t packetSource;
/**
* This method reads the packet service, checks if such a service is
* registered and forwards the packet to the destination.
* It also initiates the formal packet check and sending of verification
* messages.
* @return Iterator to map entry of found service id
* or iterator to @c map.end().
*/
ReturnValue_t selectDestination(MessageQueueId_t& destId) override;
/**
* The callback here handles the generation of acceptance
* success/failure messages.
*/
// ReturnValue_t callbackAfterSending(ReturnValue_t queueStatus) override;
};
#endif /* FSFW_TCDISTRIBUTION_CFDPDISTRIBUTOR_H_ */

8
src/fsfw/tcdistribution/PusDistributor.cpp
View File

@ -147,13 +147,13 @@ void PusDistributor::checkerFailurePrinter() const {
const char* reason = "Unknown reason";
if (tcStatus == tcdistrib::INCORRECT_CHECKSUM) {
reason = "Checksum Error";
} else if (tcStatus == tcdistrib::INCORRECT_PRIMARY_HEADER) {
} else if (tcStatus == tmtcdistrib::INCORRECT_PRIMARY_HEADER) {
reason = "Incorrect Primary Header";
} else if (tcStatus == tcdistrib::INVALID_APID) {
} else if (tcStatus == tmtcdistrib::INVALID_APID) {
reason = "Illegal APID";
} else if (tcStatus == tcdistrib::INCORRECT_SECONDARY_HEADER) {
} else if (tcStatus == tmtcdistrib::INCORRECT_SECONDARY_HEADER) {
reason = "Incorrect Secondary Header";
} else if (tcStatus == tcdistrib::INCOMPLETE_PACKET) {
} else if (tcStatus == tmtcdistrib::INCOMPLETE_PACKET) {
reason = "Incomplete packet";
}
#if FSFW_CPP_OSTREAM_ENABLED == 1

8
src/fsfw/tcdistribution/PusPacketChecker.cpp
View File

@ -12,17 +12,17 @@ PusPacketChecker::PusPacketChecker(uint16_t apid, ccsds::PacketType packetType_,
ReturnValue_t PusPacketChecker::checkPacket(const PusTcReader& pusPacket, size_t packetLen) {
// Other primary header fields are checked by base class
if (not pusPacket.hasSecHeader()) {
return tcdistrib::INVALID_SEC_HEADER_FIELD;
return tmtcdistrib::INVALID_SEC_HEADER_FIELD;
}
uint16_t calculated_crc = CRC::crc16ccitt(pusPacket.getFullData(), pusPacket.getFullPacketLen());
if (calculated_crc != 0) {
return tcdistrib::INCORRECT_CHECKSUM;
return tmtcdistrib::INCORRECT_CHECKSUM;
}
if (pusPacket.getApid() != apid) {
return tcdistrib::INVALID_APID;
return tmtcdistrib::INVALID_APID;
}
if (pusPacket.getPusVersion() != pusVersion) {
return tcdistrib::INVALID_PUS_VERSION;
return tmtcdistrib::INVALID_PUS_VERSION;
}
return HasReturnvaluesIF::RETURN_OK;
}

3
src/fsfw/tcdistribution/TcDistributorBase.cpp
View File

@ -1,5 +1,6 @@
#include "fsfw/tcdistribution/TcDistributorBase.h"
#include "definitions.h"
#include "fsfw/ipc/QueueFactory.h"
#include "fsfw/tmtcservices/TmTcMessage.h"
@ -25,7 +26,7 @@ ReturnValue_t TcDistributorBase::performOperation(uint8_t opCode) {
ReturnValue_t packetResult = handlePacket();
if (packetResult != HasReturnvaluesIF::RETURN_OK) {
result = packetResult;
triggerEvent(HANDLE_PACKET_FAILED, packetResult, __LINE__);
triggerEvent(tmtcdistrib::HANDLE_PACKET_FAILED, packetResult, 1);
}
}
if (status == MessageQueueIF::EMPTY) {

4
src/fsfw/tcdistribution/TcDistributorBase.h
View File

@ -34,10 +34,6 @@ class TcDistributorBase : public SystemObject, public ExecutableObjectIF, public
static constexpr ReturnValue_t PACKET_LOST = MAKE_RETURN_CODE(1);
static constexpr ReturnValue_t DESTINATION_NOT_FOUND = MAKE_RETURN_CODE(2);
static constexpr ReturnValue_t SERVICE_ID_ALREADY_EXISTS = MAKE_RETURN_CODE(3);
static constexpr uint8_t SUBSYSTEM_ID = SUBSYSTEM_ID::TC_DISTRIBUTION;
//! P1: Returnvalue, P2: Line number
static constexpr Event HANDLE_PACKET_FAILED = event::makeEvent(SUBSYSTEM_ID, 0, severity::LOW);
/**
* Within the default constructor, the SystemObject id is set and the
* message queue is initialized.

10
src/fsfw/tcdistribution/definitions.h
View File

@ -3,10 +3,12 @@
#include <cstdint>
#include "fsfw/events/Event.h"
#include "fsfw/events/fwSubsystemIdRanges.h"
#include "fsfw/returnvalues/FwClassIds.h"
#include "fsfw/returnvalues/HasReturnvaluesIF.h"
namespace tcdistrib {
namespace tmtcdistrib {
static const uint8_t INTERFACE_ID = CLASS_ID::PACKET_CHECK;
static constexpr ReturnValue_t INVALID_CCSDS_VERSION = MAKE_RETURN_CODE(0);
static constexpr ReturnValue_t INVALID_APID = MAKE_RETURN_CODE(1);
@ -20,5 +22,9 @@ static constexpr ReturnValue_t INCORRECT_CHECKSUM = MAKE_RETURN_CODE(7);
static constexpr ReturnValue_t ILLEGAL_PACKET_SUBTYPE = MAKE_RETURN_CODE(8);
static constexpr ReturnValue_t INCORRECT_SECONDARY_HEADER = MAKE_RETURN_CODE(9);
}; // namespace tcdistrib
static constexpr uint8_t SUBSYSTEM_ID = SUBSYSTEM_ID::TMTC_DISTRIBUTION;
//! P1: Returnvalue, P2: 0 for TM issues, 1 for TC issues
static constexpr Event HANDLE_PACKET_FAILED = event::makeEvent(SUBSYSTEM_ID, 0, severity::LOW);
}; // namespace tmtcdistrib
#endif // FSFW_TMTCPACKET_DEFINITIONS_H

2
src/fsfw/tmtcservices/AcceptsTelemetryIF.h
View File

@ -14,6 +14,8 @@ class AcceptsTelemetryIF {
* @brief The virtual destructor as it is mandatory for C++ interfaces.
*/
virtual ~AcceptsTelemetryIF() = default;
[[nodiscard]] virtual const char* getName() const = 0;
/**
* @brief This method returns the message queue id of the telemetry
* receiving message queue.

2
unittests/mocks/AcceptsTmMock.cpp
View File

@ -9,3 +9,5 @@ AcceptsTmMock::AcceptsTmMock(MessageQueueId_t queueToReturn)
MessageQueueId_t AcceptsTmMock::getReportReceptionQueue(uint8_t virtualChannel) {
return returnedQueue;
}
const char* AcceptsTmMock::getName() const { return "TM Acceptor Mock"; }

1
unittests/mocks/AcceptsTmMock.h
View File

@ -10,6 +10,7 @@ class AcceptsTmMock : public SystemObject, public AcceptsTelemetryIF {
explicit AcceptsTmMock(MessageQueueId_t queueToReturn);
MessageQueueId_t getReportReceptionQueue(uint8_t virtualChannel) override;
const char* getName() const override;
MessageQueueId_t returnedQueue;
};

2
unittests/tcdistributor/testCcsdsDistributor.cpp
View File

@ -8,7 +8,7 @@
#include "mocks/CcsdsCheckerMock.h"
#include "mocks/MessageQueueMock.h"
TEST_CASE("CCSDS Distributor", "[ccsds][tcdistrib]") {
TEST_CASE("CCSDS Distributor", "[ccsds][tmtcdistrib]") {
LocalPool::LocalPoolConfig cfg = {{5, 32}, {2, 64}};
LocalPool pool(objects::NO_OBJECT, cfg);
auto queue = MessageQueueMock(1);