/* * Copyright (C) 2014 Pavel Kirienko */ #pragma once #include #include #include #include #include #include #include #include #include #include namespace uavcan { template &), unsigned int NumStaticBufs = 1, unsigned int NumStaticReceivers = NumStaticBufs + 1> class Subscriber : Noncopyable { typedef Subscriber SelfType; public: typedef DataType_ DataType; private: typedef TransferListener::Result, NumStaticBufs ? NumStaticBufs : 1, // TODO: add support for zero buffers NumStaticReceivers ? NumStaticReceivers : 1> TransferListenerType; // We need to break the inheritance chain here to implement lazy initialization class TransferForwarder : public TransferListenerType { SelfType& obj_; void handleIncomingTransfer(IncomingTransfer& transfer) { obj_.handleIncomingTransfer(transfer); } public: TransferForwarder(SelfType& obj, const DataTypeDescriptor& data_type, IAllocator& allocator) : TransferListenerType(data_type, allocator) , obj_(obj) { } }; struct ReceivedDataStructureSpec : public ReceivedDataStructure { using ReceivedDataStructure::setTransfer; }; Scheduler& scheduler_; IAllocator& allocator_; Callback callback_; LazyConstructor forwarder_; ReceivedDataStructureSpec message_; uint32_t failure_count_; bool checkInit() { if (forwarder_) return true; GlobalDataTypeRegistry::instance().freeze(); const DataTypeDescriptor* const descr = GlobalDataTypeRegistry::instance().find(DataTypeKindMessage, DataType::getDataTypeFullName()); if (!descr) { UAVCAN_TRACE("Subscriber", "Type [%s] is not registered", DataType::getDataTypeFullName()); return false; } forwarder_.template construct(*this, *descr, allocator_); return true; } void handleIncomingTransfer(IncomingTransfer& transfer) { assert(transfer.getTransferType() == TransferTypeMessageBroadcast || transfer.getTransferType() == TransferTypeMessageUnicast); { BitStream bitstream(transfer); ScalarCodec codec(bitstream); const int decode_res = DataType::decode(message_, codec); // We don't need the data anymore, the memory can be reused from the callback: transfer.release(); if (decode_res <= 0) { UAVCAN_TRACE("Subscriber", "Unable to decode the message [%i] [%s]", decode_res, DataType::getDataTypeFullName()); failure_count_++; return; } } message_.setTransfer(&transfer); if (try_implicit_cast(callback_, true)) callback_(message_); // Callback can accept non-const message reference and mutate it, that's OK else assert(0); } public: Subscriber(Scheduler& scheduler, IAllocator& allocator) : scheduler_(scheduler) , allocator_(allocator) , callback_() , failure_count_(0) { StaticAssert::check(); } virtual ~Subscriber() { stop(); } int start(Callback callback) { stop(); if (!try_implicit_cast(callback, true)) { UAVCAN_TRACE("Subscriber", "Invalid callback"); return -1; } callback_ = callback; if (!checkInit()) { UAVCAN_TRACE("Subscriber", "Initialization failure [%s]", DataType::getDataTypeFullName()); return -1; } if (!scheduler_.getDispatcher().registerMessageListener(forwarder_)) { UAVCAN_TRACE("Subscriber", "Failed to register message listener [%s]", DataType::getDataTypeFullName()); return -1; } return 1; } void stop() { if (forwarder_) scheduler_.getDispatcher().unregisterMessageListener(forwarder_); } Scheduler& getScheduler() const { return scheduler_; } uint32_t getFailureCount() const { return failure_count_; } }; }