Feature/source plugins #1
No files matched your search
@@ -9,11 +9,14 @@
|
||||
#include <chrono>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
#include <ifaddrs.h>
|
||||
#include <net/if.h>
|
||||
#include <netdb.h>
|
||||
#include <netinet/in.h>
|
||||
#include <netinet/tcp.h>
|
||||
#include <poll.h>
|
||||
#include <sys/socket.h>
|
||||
#include <thread>
|
||||
#include <unistd.h>
|
||||
|
||||
namespace udpscope::e1608 {
|
||||
@@ -120,6 +123,25 @@ Fd tcp_connect(const std::string& host, int port, int timeout_ms, int rcvbuf) {
|
||||
return fd;
|
||||
}
|
||||
|
||||
/// A `D` reply (at least 29 bytes, first byte `D`, our product id). True when it added a board
|
||||
/// that `found` did not already hold (one entry per MAC).
|
||||
bool take_reply(const std::uint8_t* buf, ssize_t n, const sockaddr_in& from,
|
||||
std::vector<DeviceInfo>& found) {
|
||||
if (n < 29 || buf[0] != 'D') return false;
|
||||
DeviceInfo info;
|
||||
info.product_id = static_cast<std::uint16_t>(buf[7] | (buf[8] << 8));
|
||||
if (info.product_id != kProductId) return false;
|
||||
std::copy(buf + 1, buf + 7, info.mac.begin());
|
||||
for (const DeviceInfo& d : found) {
|
||||
if (d.mac == info.mac) return false;
|
||||
}
|
||||
char ip[INET_ADDRSTRLEN] = {};
|
||||
::inet_ntop(AF_INET, &from.sin_addr, ip, sizeof ip);
|
||||
info.ip = ip;
|
||||
found.push_back(info);
|
||||
return true;
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
// --- addresses ------------------------------------------------------------------------------
|
||||
@@ -222,22 +244,70 @@ std::vector<DeviceInfo> discover(const std::vector<std::string>& targets, int po
|
||||
if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) continue;
|
||||
break;
|
||||
}
|
||||
if (n < 29 || buf[0] != 'D') continue;
|
||||
DeviceInfo info;
|
||||
info.product_id = static_cast<std::uint16_t>(buf[7] | (buf[8] << 8));
|
||||
if (info.product_id != kProductId) continue;
|
||||
std::copy(buf + 1, buf + 7, info.mac.begin());
|
||||
bool dup = false;
|
||||
for (const DeviceInfo& d : found) dup = dup || d.mac == info.mac;
|
||||
if (dup) continue;
|
||||
char ip[INET_ADDRSTRLEN] = {};
|
||||
::inet_ntop(AF_INET, &from.sin_addr, ip, sizeof ip);
|
||||
info.ip = ip;
|
||||
found.push_back(info);
|
||||
take_reply(buf, n, from, found);
|
||||
}
|
||||
return found;
|
||||
}
|
||||
|
||||
Scan::Scan(int port) {
|
||||
ifaddrs* list = nullptr;
|
||||
if (::getifaddrs(&list) != 0) return;
|
||||
sockaddr_in to{};
|
||||
to.sin_family = AF_INET;
|
||||
to.sin_port = htons(static_cast<std::uint16_t>(port));
|
||||
to.sin_addr.s_addr = htonl(INADDR_BROADCAST);
|
||||
const std::uint8_t probe = 'D';
|
||||
std::vector<std::string> done; // an interface with several addresses needs one probe, not several
|
||||
for (const ifaddrs* a = list; a != nullptr; a = a->ifa_next) {
|
||||
if (a->ifa_addr == nullptr || a->ifa_addr->sa_family != AF_INET) continue;
|
||||
const unsigned fl = a->ifa_flags;
|
||||
if ((fl & IFF_UP) == 0 || (fl & IFF_LOOPBACK) != 0 || (fl & IFF_BROADCAST) == 0) continue;
|
||||
const std::string name = a->ifa_name;
|
||||
if (std::find(done.begin(), done.end(), name) != done.end()) continue;
|
||||
Fd s(::socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, 0));
|
||||
if (!s.valid()) continue;
|
||||
const int one = 1;
|
||||
::setsockopt(s.get(), SOL_SOCKET, SO_BROADCAST, &one, sizeof one);
|
||||
if (::setsockopt(s.get(), SOL_SOCKET, SO_BINDTODEVICE, name.c_str(),
|
||||
static_cast<socklen_t>(name.size())) != 0) {
|
||||
sockaddr_in me = *reinterpret_cast<const sockaddr_in*>(a->ifa_addr);
|
||||
me.sin_port = 0;
|
||||
if (::bind(s.get(), reinterpret_cast<const sockaddr*>(&me), sizeof me) != 0) continue;
|
||||
}
|
||||
if (::sendto(s.get(), &probe, 1, MSG_NOSIGNAL, reinterpret_cast<const sockaddr*>(&to),
|
||||
sizeof to) != 1)
|
||||
continue;
|
||||
done.push_back(name);
|
||||
socks_.push_back(std::move(s));
|
||||
}
|
||||
::freeifaddrs(list);
|
||||
}
|
||||
|
||||
const std::vector<DeviceInfo>& Scan::poll() {
|
||||
for (const Fd& s : socks_) {
|
||||
for (;;) {
|
||||
std::uint8_t buf[128];
|
||||
sockaddr_in from{};
|
||||
socklen_t flen = sizeof from;
|
||||
const ssize_t n = ::recvfrom(s.get(), buf, sizeof buf, MSG_DONTWAIT,
|
||||
reinterpret_cast<sockaddr*>(&from), &flen);
|
||||
if (n < 0) break; // nothing more on this socket for now
|
||||
take_reply(buf, n, from, found_);
|
||||
}
|
||||
}
|
||||
return found_;
|
||||
}
|
||||
|
||||
std::vector<DeviceInfo> discover_broadcast(int port, int wait_ms) {
|
||||
Scan scan(port);
|
||||
const auto deadline = Clock::now() + std::chrono::milliseconds(wait_ms);
|
||||
while (ms_left(deadline) > 0) {
|
||||
scan.poll();
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(10));
|
||||
}
|
||||
return scan.poll();
|
||||
}
|
||||
|
||||
// --- Link -----------------------------------------------------------------------------------
|
||||
|
||||
void Link::connect(const std::string& host, int port, std::uint32_t code, int timeout_ms) {
|
||||
|
||||
@@ -76,6 +76,30 @@ std::vector<std::string> sweep_targets(const std::string& subnet);
|
||||
/// `kProductId`) and one entry per MAC.
|
||||
std::vector<DeviceInfo> discover(const std::vector<std::string>& targets, int port, int wait_ms);
|
||||
|
||||
/// A broadcast discovery that never blocks. The limited broadcast alone leaves through the default
|
||||
/// route only, so on a host with several interfaces it can miss a board on another one; this opens
|
||||
/// one socket per up, non-loopback IPv4 interface with a broadcast address (bound to the device, or
|
||||
/// to the interface address when binding to the device is refused), sends `D` to
|
||||
/// 255.255.255.255 on each, and collects the replies as `poll()` is called. The caller owns the
|
||||
/// clock. An interface that cannot be opened or sent on is skipped; no interface is an empty scan.
|
||||
class Scan {
|
||||
public:
|
||||
explicit Scan(int port);
|
||||
Scan(const Scan&) = delete;
|
||||
Scan& operator=(const Scan&) = delete;
|
||||
/// Drains every socket without blocking. The devices found so far: E-1608s only, one per MAC.
|
||||
const std::vector<DeviceInfo>& poll();
|
||||
/// How many interfaces a probe was sent on.
|
||||
std::size_t interfaces() const noexcept { return socks_.size(); }
|
||||
|
||||
private:
|
||||
std::vector<Fd> socks_;
|
||||
std::vector<DeviceInfo> found_;
|
||||
};
|
||||
|
||||
/// `Scan` plus a blocking wait of `wait_ms`: what a lookup by MAC with no subnet uses.
|
||||
std::vector<DeviceInfo> discover_broadcast(int port, int wait_ms);
|
||||
|
||||
/// One board: the UDP handshake, the TCP command channel, and the scan socket (`port + 1`).
|
||||
/// Not thread safe; the source's producer thread owns it.
|
||||
class Link {
|
||||
|
||||
@@ -274,8 +274,13 @@ void E1608Source::connect_and_start() {
|
||||
if (host.empty() || have_mac) {
|
||||
const std::vector<std::string> targets =
|
||||
host.empty() ? sweep_targets(cfg_.subnet) : std::vector<std::string>{host};
|
||||
const int wait_ms = std::min(cfg_.timeout_ms, 1500);
|
||||
// No address and no subnet: the limited broadcast, sent on every interface rather than only
|
||||
// the default route's.
|
||||
const bool broadcast =
|
||||
host.empty() && targets.size() == 1 && targets[0] == "255.255.255.255";
|
||||
const std::vector<DeviceInfo> found =
|
||||
discover(targets, cfg_.port, std::min(cfg_.timeout_ms, 1500));
|
||||
broadcast ? discover_broadcast(cfg_.port, wait_ms) : discover(targets, cfg_.port, wait_ms);
|
||||
if (host.empty()) {
|
||||
for (const DeviceInfo& d : found) {
|
||||
if (d.mac == want) host = d.ip;
|
||||
|
||||
Reference in new issue
Block a user