-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Started implementing npcap based receiver
- Loading branch information
1 parent
655bbf1
commit afc5ad4
Showing
14 changed files
with
1,006 additions
and
5 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,97 @@ | ||
| #include <array> | ||
| #include <cstddef> | ||
| #include <functional> | ||
| #include <memory> | ||
| #include <chrono> | ||
| #include <thread> | ||
|
|
||
| #include <asio.hpp> | ||
|
|
||
| #include <ecaludp/error.h> | ||
| #include <ecaludp/owning_buffer.h> | ||
| #include <ecaludp/raw_memory.h> | ||
| #include <vector> | ||
|
|
||
| #include <ecaludp/ecaludp_export.h> | ||
|
|
||
| namespace ecaludp | ||
| { | ||
| namespace v5 | ||
| { | ||
| class Reassembly; | ||
| } | ||
|
|
||
| class recycle_shared_pool; | ||
| class AsyncUdpcapSocket; | ||
|
|
||
| class SocketUdpcap | ||
| { | ||
| ///////////////////////////////////////////////////////////////// | ||
| // Constructor | ||
| ///////////////////////////////////////////////////////////////// | ||
| public: | ||
| ECALUDP_EXPORT SocketUdpcap(std::array<char, 4> magic_header_bytes); | ||
|
|
||
| // Destructor | ||
| ECALUDP_EXPORT ~SocketUdpcap(); | ||
|
|
||
| // Disable copy constructor and assignment operator | ||
| SocketUdpcap(const SocketUdpcap&) = delete; | ||
| SocketUdpcap& operator=(const SocketUdpcap&) = delete; | ||
|
|
||
| // Disable move constructor and assignment operator | ||
| SocketUdpcap(SocketUdpcap&&) = delete; | ||
| SocketUdpcap& operator=(SocketUdpcap&&) = delete; | ||
|
|
||
| ///////////////////////////////////////////////////////////////// | ||
| // Settings | ||
| ///////////////////////////////////////////////////////////////// | ||
| public: | ||
| ECALUDP_EXPORT void set_max_reassembly_age(std::chrono::steady_clock::duration max_reassembly_age); | ||
| ECALUDP_EXPORT std::chrono::steady_clock::duration get_max_reassembly_age() const; | ||
|
|
||
| ///////////////////////////////////////////////////////////////// | ||
| // API "Passthrough" (and a bit conversion to asio types) | ||
| ///////////////////////////////////////////////////////////////// | ||
| public: | ||
| bool is_valid() const; | ||
| bool bind(const asio::ip::udp::endpoint& sender_endpoint); | ||
| bool is_bound() const; | ||
| asio::ip::udp::endpoint local_endpoint(); | ||
| bool set_receive_buffer_size(int size); | ||
| bool has_pending_datagrams() const; | ||
| bool join_multicast_group(const asio::ip::address_v4& group_address); | ||
| bool leave_multicast_group(const asio::ip::address_v4& group_address); | ||
| void set_multicast_loopback_enabled(bool enabled); | ||
| bool is_multicast_loopback_enabled() const; | ||
| void close(); | ||
|
|
||
| ///////////////////////////////////////////////////////////////// | ||
| // Receiving | ||
| ///////////////////////////////////////////////////////////////// | ||
| public: | ||
| ECALUDP_EXPORT void async_receive_from(asio::ip::udp::endpoint& sender_endpoint | ||
| , const std::function<void(const std::shared_ptr<ecaludp::OwningBuffer>&, ecaludp::Error)>& completion_handler); | ||
|
|
||
|
|
||
| private: | ||
| void receive_next_datagram_from(asio::ip::udp::endpoint& sender_endpoint | ||
| , const std::function<void(const std::shared_ptr<ecaludp::OwningBuffer>&, ecaludp::Error)>& completion_handler); | ||
|
|
||
| std::shared_ptr<ecaludp::OwningBuffer> handle_datagram(const std::shared_ptr<ecaludp::RawMemory>& buffer | ||
| , const std::shared_ptr<asio::ip::udp::endpoint>& sender_endpoint | ||
| , ecaludp::Error& error); | ||
|
|
||
| ///////////////////////////////////////////////////////////////// | ||
| // Member Variables | ||
| ///////////////////////////////////////////////////////////////// | ||
| private: | ||
| std::unique_ptr<ecaludp::AsyncUdpcapSocket> socket_; ///< The "socket" implementation | ||
|
|
||
| std::unique_ptr<recycle_shared_pool> datagram_buffer_pool_; ///< A reusable buffer pool for single datagrams (i.e. tyically 1500 byte fragments) | ||
| std::unique_ptr<ecaludp::v5::Reassembly> reassembly_v5_; ///< The reassembly for the eCAL v5 protocol | ||
|
|
||
| std::array<char, 4> magic_header_bytes_; ///< The magic bytes that are expected to start each fragment. If the received datagram doesn't have those, it will be dropped immediatelly | ||
| std::chrono::steady_clock::duration max_reassembly_age_; ///< Fragments that are stored in the reassembly for longer than that period will be dropped. | ||
| }; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| #include "async_udpcap_socket.h" | ||
|
|
||
| #include <iostream> // TODO: Remove | ||
|
|
||
| namespace ecaludp | ||
| { | ||
| AsyncUdpcapSocket::AsyncUdpcapSocket() | ||
| : udpcap_socket_() | ||
| , is_closed(false) | ||
| {} | ||
|
|
||
| AsyncUdpcapSocket::~AsyncUdpcapSocket() | ||
| { | ||
| close(); | ||
| // TODO: Stop the socket and the wait thread | ||
| if (wait_thread_) | ||
| { | ||
| wait_thread_->join(); | ||
| } | ||
|
|
||
| // TODO: implement | ||
| } | ||
|
|
||
| void AsyncUdpcapSocket::asyncReceiveFrom( char* buffer | ||
| , size_t max_buffer_size | ||
| , Udpcap::HostAddress& sender_address | ||
| , uint16_t& sender_port | ||
| , const std::function<void(ecaludp::Error, size_t)>& read_handler) | ||
| { | ||
| std::unique_lock<std::mutex> lock(wait_thread_trigger_mutex_); | ||
| std::cerr << "===Pushing async receive from parameters" << std::endl; // TODO REMOVE | ||
| async_receive_from_parameters_queue_.push_back({ buffer, max_buffer_size, &sender_address, &sender_port, read_handler }); | ||
| wait_thread_trigger_cv_.notify_one(); | ||
| } | ||
| ///////////////////////////////////////////////////// | ||
| // udpcap forwarded methods | ||
| ///////////////////////////////////////////////////// | ||
| bool AsyncUdpcapSocket::bind(const Udpcap::HostAddress& local_address, uint16_t local_port) | ||
| { | ||
| bool success = udpcap_socket_.bind(local_address, local_port); | ||
|
|
||
| if (success) | ||
| { | ||
| if (wait_thread_ && wait_thread_->joinable()) | ||
| { | ||
| wait_thread_->join(); | ||
| } | ||
|
|
||
| is_closed = false; | ||
|
|
||
| wait_thread_ = std::make_unique<std::thread>(&AsyncUdpcapSocket::waitForData, this); | ||
| } | ||
|
|
||
| return success; | ||
| } | ||
|
|
||
| void AsyncUdpcapSocket::close() | ||
| { | ||
| udpcap_socket_.close(); | ||
| { | ||
| std::lock_guard<std::mutex> lock(wait_thread_trigger_mutex_); | ||
| is_closed = true; | ||
| wait_thread_trigger_cv_.notify_one(); | ||
| } | ||
| } | ||
|
|
||
| void AsyncUdpcapSocket::waitForData() | ||
| { | ||
| while (true) | ||
| { | ||
| //TODO: Revise this function, it is very important, but currently not safe. | ||
|
|
||
| AsyncReceiveFromParameters next_async_receive_from_parameters{}; | ||
|
|
||
| // Wait until there is somebody requesting some data. This is done by waiting for the callback queue to be non-empty. | ||
| { | ||
| std::unique_lock<std::mutex> lock(wait_thread_trigger_mutex_); | ||
| wait_thread_trigger_cv_.wait(lock, [this] { return is_closed || !async_receive_from_parameters_queue_.empty(); }); | ||
|
|
||
| if (!async_receive_from_parameters_queue_.empty()) | ||
| { | ||
| next_async_receive_from_parameters = async_receive_from_parameters_queue_.front(); | ||
| async_receive_from_parameters_queue_.pop_front(); | ||
| } | ||
|
|
||
| if (async_receive_from_parameters_queue_.empty() && is_closed) | ||
| { | ||
| return; | ||
| } | ||
| } | ||
|
|
||
| if (udpcap_socket_.isBound()) | ||
| { | ||
| std::cerr << "===Start receiving data" << std::endl; // TODO REMOVE | ||
| size_t rec_bytes = udpcap_socket_.receiveDatagram(next_async_receive_from_parameters.buffer_ | ||
| , next_async_receive_from_parameters.max_buffer_size_ | ||
| , next_async_receive_from_parameters.sender_address_ | ||
| , next_async_receive_from_parameters.sender_port_); | ||
|
|
||
| std::cerr << "===Received " << rec_bytes << " bytes from " << next_async_receive_from_parameters.sender_address_->toString() << ":" << std::to_string(*next_async_receive_from_parameters.sender_port_) << std::endl; // TODO REMOVE | ||
|
|
||
| if (next_async_receive_from_parameters.sender_address_->isValid()) | ||
| { | ||
| next_async_receive_from_parameters.read_handler_(ecaludp::Error::OK, rec_bytes); | ||
| } | ||
| else | ||
| { | ||
| next_async_receive_from_parameters.read_handler_(ecaludp::Error::GENERIC_ERROR, rec_bytes); | ||
| } | ||
| } | ||
| else | ||
| { | ||
| next_async_receive_from_parameters.read_handler_(ecaludp::Error::GENERIC_ERROR, 0); | ||
| } | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.