multicastreceiver.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_NETWORK
#include <functional>
#include <promeki/function.h>
#include <promeki/atomic.h>
#include <promeki/buffer.h>
#include <promeki/error.h>
#include <promeki/list.h>
#include <promeki/multicastmanager.h>
#include <promeki/socketaddress.h>
#include <promeki/string.h>
#include <promeki/thread.h>
#include <promeki/udpsocket.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class MulticastReceiver : public Thread {
                PROMEKI_OBJECT(MulticastReceiver, Thread)
        public:
                using DatagramCallback = Function<void(Buffer data, const SocketAddress &sender)>;
 
                struct GroupEntry {
                                SocketAddress group;
                                SocketAddress source;
                                bool isSSM = false;
                };
 
                using GroupList = ::promeki::List<GroupEntry>;
 
                static constexpr size_t DefaultMaxPacketSize = 2048;
 
                static constexpr unsigned int DefaultReceiveTimeoutMs = 200;
 
                MulticastReceiver(ObjectBase *parent = nullptr);
 
                ~MulticastReceiver() override;
 
                void setLocalAddress(const SocketAddress &address);
 
                const SocketAddress &localAddress() const { return _localAddress; }
 
                void setInterface(const String &iface);
 
                const String &interfaceName() const { return _interfaceName; }
 
                void setMaxPacketSize(size_t bytes);
 
                size_t maxPacketSize() const { return _maxPacketSize; }
 
                void setReceiveTimeout(unsigned int timeoutMs);
 
                unsigned int receiveTimeout() const { return _receiveTimeoutMs; }
 
                void setThreadName(const String &name);
 
                const String &threadName() const { return _threadName; }
 
                void setDatagramCallback(DatagramCallback callback);
 
                Error addGroup(const SocketAddress &group);
 
                Error addSourceGroup(const SocketAddress &group, const SocketAddress &source);
 
                const GroupList &groups() const { return _groups; }
 
                Error start();
 
                void stop();
 
                bool isActive() const { return _active.value(); }
 
                UdpSocket *socket() const { return _socket.get(); }
 
                uint64_t datagramCount() const { return _datagramCount.value(); }
 
                uint64_t byteCount() const { return _byteCount.value(); }
 
                PROMEKI_SIGNAL(datagramReceived, Buffer, SocketAddress);
 
                PROMEKI_SIGNAL(receiveError, Error);
 
        protected:
                void run() override;
 
        private:
                Error openAndJoin();
                void  closeAndLeave();
 
                UdpSocket::UPtr  _socket;
                MulticastManager _multicastManager;
                SocketAddress    _localAddress;
                String           _interfaceName;
                String           _threadName;
                size_t           _maxPacketSize = DefaultMaxPacketSize;
                unsigned int     _receiveTimeoutMs = DefaultReceiveTimeoutMs;
                DatagramCallback _callback;
                GroupList        _groups;
 
                Atomic<bool>     _active;
                Atomic<bool>     _stopRequested;
                Atomic<uint64_t> _datagramCount;
                Atomic<uint64_t> _byteCount;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_NETWORK