janis/aes67-daemon
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'master' into codec_am824

Browse Source
This commit is contained in:
Andrea Bondavalli 2020-04-23 12:11:26 -07:00
commit b35b36c91e
32 changed files with 1290 additions and 228 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
README.md
View File

@ -5,7 +5,7 @@ See [https://en.wikipedia.org/wiki/AES67](https://en.wikipedia.org/wiki/AES67) f
# Introduction
The daemon is a Linux process that uses the [Merging Technologies ALSA RAVENNA/AES67 Driver](https://bitbucket.org/MergingTechnologies/ravenna-alsa-lkm/src/master) to handles PTP synchronization and RTP streams and exposes a REST interface for configuration and status monitoring.
The daemon is a Linux process that uses the [Merging Technologies ALSA RAVENNA/AES67 Driver](https://bitbucket.org/MergingTechnologies/ravenna-alsa-lkm/src/master) to handle PTP synchronization and RTP streams and exposes a REST interface for configuration and status monitoring.
The **ALSA AES67 Driver** implements a virtual ALSA audio device that can be configured using _Sources_ and _Sinks_ and it's clocked using the PTP clock.
A _Source_ reads audio samples from the ALSA playback device and sends RTP packets to a configured multicast address.
@ -40,15 +40,16 @@ The daemon uses the following open source:
This directory contains the AES67 daemon source code.
The daemon can be cross-compiled for multiple platforms and implements the following functionalities:
* communication and configuration of the ALSA RAVENNA/AES67 device driver
* control and configuration of up to 64 sources and sinks using the ALSA RAVENNA/AES67 driver via netlink
* session handling and SDP parsing and creation
* HTTP REST API for control and configuration
* SAP discovery protocol and SAP browser
* mDNS sources discovery (using Avahi) and SDP transfer via RTSP
* IGMP handling for SAP, RTP and PTP multicast traffic
The directory also contains the daemon regression tests in the [tests](daemon/tests) subdirectory. To run daemon tests install the ALSA RAVENNA/AES67 kernel module enter the [tests](daemon/tests) subdirectory and run *./daemon-test -l all*
* HTTP REST API for the daemon control and configuration
* SAP sources discovery and advertisement compatible with AES67 standard
* mDNS sources discovery and advertisement (using Linux Avahi) compatible with Ravenna standard
* RTSP client and server to retrieve or return SDP files via DESCRIBE method
* IGMP handling for SAP and RTP sessions
The directory also contains the daemon regression tests in the [tests](daemon/tests) subdirectory.
See the [README](daemon/README.md) file in this directory for additional information about the AES67 daemon configuration and the HTTP REST API.
### [webui](webui) directory ###
@ -85,7 +86,7 @@ This directory contains a the daemon configuration and status files used to run
The daemon and the demo have been tested with **Ubuntu 18.04** distro on **ARMv7** and with **Ubuntu 18.04, 19.10 and 20.04** distros on **x86** using:
* Linux kernel version >= 4.14.x
* GCC version >= 7.4 / clang >= 6.0.0 (C++17 support required, clang is required to compile on ARMv7)
* GCC version >= 7.4 / clang >= 6.0.0 (C++17 support required)
* cmake version >= 3.10.2
* node version >= 8.10.0
* npm version >= 3.5.2
@ -108,7 +109,20 @@ The script performs the following operations:
* build and deploy the WebUI
* build the AES67 daemon
## Run the Demo ##
## Run the regression tests ##
To run daemon regression tests install the ALSA RAVENNA/AES67 kernel module with:
sudo insmod 3rdparty/ravenna-alsa-lkm/driver/MergingRavennaALSA.ko
setup the kernel parameters with:
sudo sysctl -w net/ipv4/igmp_max_memberships=66
make sure that no instances of the aes67-daemon are running, enter the [tests](daemon/tests) subdirectory and run:
./daemon-test
## Run the demo ##
<a name="demo"></a>
To run a simple demo use the [run\_demo.sh](run_demo.sh) script. See [script notes](#notes).
@ -147,6 +161,7 @@ To run interoperability tests using the [Hasseb audio over Ethernet receiver](ht
* open the Hasseb WebUI and do the following:
* deselect the "PTP slave only" checkbox to enable PTP master on Hasseb device
* select the "Add SDP file manually" checkbox and copy the previous Source SDP into the SDP field
* alternatively to the step above, select in the "Stream name" drop down the daemon Source just added
* press the Submit button
* return to the daemon WebUI, click on the PTP tab and wait for the "PTP Status" to report "locked"
* open a shell on the Linux host and start the playback on the ravenna ALSA device. For example to playback a test sound use:

2
daemon/CMakeLists.txt
View File

@ -13,7 +13,7 @@ set(AVAHI_LIBRARIES ${AVAHI_LIBRARY-COMMON} ${AVAHI_LIBRARY-CLIENT})
set(AVAHI_INCLUDE_DIRS ${AVAHI_INCLUDE_DIR})
find_package(Boost COMPONENTS system thread log program_options REQUIRED)
include_directories(aes67-daemon ${RAVENNA_ALSA_LKM}/common ${RAVENNA_ALSA_LKM}/driver ${CPP_HTTPLIB} ${Boost_INCLUDE_DIR})
add_executable(aes67-daemon error_code.cpp json.cpp main.cpp driver_handler.cpp driver_manager.cpp session_manager.cpp http_server.cpp config.cpp interface.cpp log.cpp sap.cpp browser.cpp rtsp_client.cpp mdns_client.cpp utils.cpp)
add_executable(aes67-daemon error_code.cpp json.cpp main.cpp driver_handler.cpp driver_manager.cpp session_manager.cpp http_server.cpp config.cpp interface.cpp log.cpp sap.cpp browser.cpp rtsp_client.cpp mdns_client.cpp mdns_server.cpp rtsp_server.cpp utils.cpp)
add_subdirectory(tests)
target_link_libraries(aes67-daemon ${Boost_LIBRARIES})
if(WITH_AVAHI)

30
daemon/README.md
View File

@ -4,13 +4,16 @@ AES67 daemon uses the Merging Technologies device driver (MergingRavennaALSA.ko)
The daemon is responsible for:
* communication and configuration of the device driver
* provide an HTTP REST API for the daemon control and configuration
* communication and configuration of the ALSA RAVENNA/AES67 device driver
* control and configuration of up to 64 sources and sinks using the ALSA RAVENNA/AES67 driver via netlink
* session handling and SDP parsing and creation
* SAP discovery protocol and SAP browser
* mDNS sources discovery (using Avahi) and SDP transfer via RTSP
* HTTP REST API for the daemon control and configuration
* SAP sources discovery and advertisement compatible with AES67 standard
* mDNS sources discovery and advertisement (using Linux Avahi) compatible with Ravenna standard
* RTSP client and server to retrieve or return SDP files via DESCRIBE method
* IGMP handling for SAP and RTP sessions
## Configuration file ##
The daemon uses a JSON file to store the configuration parameters.
@ -130,10 +133,10 @@ In case of failure the server returns a **text/plain** content type with the cat
* **Body** [RTP Streams params](#rtp-streams)
### Get all remote RTP Sources ###
* **Description** retrieve all the remote sources collected via SAP
* **URL** /api/browse/sources
* **Description** retrieve all the remote sources collected via SAP and mDNS
* **URL** /api/browse/sources/[all|mdns|sap]
* **Method** GET
* **URL Params** none
* **URL Params** all=[all sources], mdns=[mDNS sources only], sap=[sap sources only]
* **Body type** application/json
* **Body** [RTP Remote Sources params](#rtp-remote-sources)
@ -146,6 +149,7 @@ Example
{
"interface_name": "lo",
"http_port": 8080,
"rtsp_port": 8854,
"log_severity": 2,
"syslog_proto": "none",
"syslog_server": "255.255.255.254:1234",
@ -161,7 +165,8 @@ Example
"sap_mcast_addr": "239.255.255.255",
"sap_interval": 30,
"mac_addr": "01:00:5e:01:00:01",
"ip_addr": "127.0.0.1"
"ip_addr": "127.0.0.1",
"node_id": "AES67 daemon ubuntu-d9aca383"
}
where:
@ -170,7 +175,10 @@ where:
> JSON string specifying the network interface used by the daemon and the driver for both the RTP, PTP, SAP and HTTP traffic.
> **http\_port**
> JSON number specifying the HTTP port number used by the embedded web server in the daemon implementing the REST interface.
> JSON number specifying the HTTP port number used by the web server in the daemon implementing the REST interface.
> **rtsp\_port**
> JSON number specifying the RTSP port number used by the RTSP server in the daemon.
> **log\_severity**
> JSON integer specifying the process log severity level (0 to 5).
@ -236,6 +244,10 @@ where:
> **mdns\_enabled**
> JSON boolean specifying whether the mDNS discovery is enabled or disabled.
> **node\_id**
> JSON string specifying the unique node identifier used to identify mDNS, SAP and SDP services announced by the daemon.
> **_NOTE:_** This parameter is read-only and cannot be set. The server will determine the node id at startup time.
### JSON PTP Config<a name="ptp-config"></a> ###
Example

37
daemon/browser.cpp
View File

@ -18,6 +18,8 @@
//
#include <boost/algorithm/string.hpp>
#include "utils.hpp"
#include "browser.hpp"
using namespace boost::algorithm;
@ -37,12 +39,16 @@ std::shared_ptr<Browser> Browser::create(
return ptr;
}
std::list<RemoteSource> Browser::get_remote_sources() const {
std::list<RemoteSource> Browser::get_remote_sources(
const std::string& _source) const {
std::list<RemoteSource> sources_list;
std::shared_lock sources_lock(sources_mutex_);
// return list of remote sources ordered by name
for (const auto& source: sources_.get<name_tag>()) {
sources_list.push_back(source);
if (boost::iequals(source.source, _source) ||
boost::iequals("all", _source)) {
sources_list.push_back(source);
}
}
return sources_list;
}
@ -52,7 +58,9 @@ static std::string sdp_get_subject(const std::string& sdp) {
std::string line;
while (getline(ssstrem, line, '\n')) {
if (line.substr(0, 2) == "s=") {
return line.substr(2);
auto subject = line.substr(2);
trim(subject);
return subject;
}
}
return "";
@ -75,7 +83,7 @@ bool Browser::worker() {
if (sap_.receive(is_announce, msg_id_hash, addr, sdp)) {
std::stringstream ss;
ss << "sap:" << std::hex << addr << msg_id_hash;
ss << "sap:" << msg_id_hash;
std::string id(ss.str());
BOOST_LOG_TRIVIAL(debug) << "browser:: received SAP message for " << id;
@ -86,19 +94,12 @@ bool Browser::worker() {
// Source is not in the map
if (is_announce) {
// annoucement, add new source
RemoteSource source;
source.id = id;
source.sdp = sdp;
source.source = "SAP";
source.address = ip::address_v4(ntohl(addr)).to_string();
source.name = sdp_get_subject(sdp);
trim(source.name);
source.last_seen =
duration_cast<second_t>(steady_clock::now() - startup_).count();
source.announce_period = 360; //default period
BOOST_LOG_TRIVIAL(info) << "browser:: adding SAP source " << source.id
<< " name " << source.name;
sources_.insert(source);
sources_.insert({
id, "SAP", ip::address_v4(ntohl(addr)).to_string(),
sdp_get_subject(sdp), {}, sdp,
static_cast<uint32_t>(duration_cast<second_t>(steady_clock::now()
- startup_).count()),
360 });
}
} else {
// Source is already in the map
@ -156,7 +157,7 @@ bool Browser::worker() {
void Browser::on_new_rtsp_source(const std::string& name,
const std::string& domain,
const RTSPSSource& s) {
const RtspSource& s) {
uint32_t last_seen = duration_cast<second_t>(steady_clock::now() -
startup_).count();
std::unique_lock sources_lock(sources_mutex_);

5
daemon/browser.hpp
View File

@ -62,7 +62,8 @@ class Browser : public MDNSClient {
bool init() override;
bool terminate() override;
std::list<RemoteSource> get_remote_sources() const;
std::list<RemoteSource> get_remote_sources(
const std::string& source = "all") const;
protected:
// singleton, use create() to build
@ -75,7 +76,7 @@ class Browser : public MDNSClient {
virtual void on_new_rtsp_source(
const std::string& name,
const std::string& domain,
const RTSPSSource& source) override;
const RtspSource& source) override;
virtual void on_remove_rtsp_source(
const std::string& name,
const std::string& domain) override;

3
daemon/config.hpp
View File

@ -33,6 +33,7 @@ class Config {
/* attributes retrieved from config json */
uint16_t get_http_port() const { return http_port_; };
uint16_t get_rtsp_port() const { return rtsp_port_; };
const std::string get_http_base_dir() const { return http_base_dir_; };
int get_log_severity() const { return log_severity_; };
uint32_t get_playout_delay() const { return playout_delay_; };
@ -61,6 +62,7 @@ class Config {
int get_interface_idx() { return interface_idx_; };
void set_http_port(uint16_t http_port) { http_port_ = http_port; };
void set_rtsp_port(uint16_t rtsp_port) { rtsp_port_ = rtsp_port; };
void set_http_base_dir(const std::string& http_base_dir) { http_base_dir_ = http_base_dir; };
void set_log_severity(int log_severity) { log_severity_ = log_severity; };
void set_playout_delay(uint32_t playout_delay) {
@ -111,6 +113,7 @@ class Config {
private:
/* from json */
uint16_t http_port_{8080};
uint16_t rtsp_port_{8854};
std::string http_base_dir_{"../webui/build"};
int log_severity_{2};
uint32_t playout_delay_{0};

1
daemon/daemon.conf
View File

@ -1,5 +1,6 @@
{
"http_port": 8080,
"rtsp_port": 8854,
"http_base_dir": "../webui/build",
"log_severity": 2,
"playout_delay": 0,

2
daemon/error_code.cpp
View File

@ -93,6 +93,8 @@ std::string DaemonErrCategory::message(int ev) const {
return "invalid stream id";
case DaemonErrc::stream_id_in_use:
return "stream id is in use";
case DaemonErrc::stream_name_in_use:
return "stream name is in use";
case DaemonErrc::stream_id_not_in_use:
return "stream not in use";
case DaemonErrc::invalid_url:

1
daemon/error_code.hpp
View File

@ -51,6 +51,7 @@ enum class DaemonErrc {
invalid_url = 43, // daemon invalid URL
cannot_retrieve_sdp = 44, // daemon cannot retrieve SDP
cannot_parse_sdp = 45, // daemon cannot parse SDP
stream_name_in_use = 46, // daemon source or sink name in use
send_invalid_size = 50, // daemon data size too big for buffer
send_u2k_failed = 51, // daemon failed to send command to driver
send_k2u_failed = 52, // daemon failed to send event response to driver

5
daemon/http_server.cpp
View File

@ -281,8 +281,9 @@ bool HttpServer::init() {
});
/* get remote sources */
svr_.Get("/api/browse/sources", [this](const Request& req, Response& res) {
auto const sources = browser_->get_remote_sources();
svr_.Get("/api/browse/sources/(all|mdns|sap)",
[this](const Request& req, Response& res) {
auto const sources = browser_->get_remote_sources(req.matches[1]);
set_headers(res, "application/json");
res.body = remote_sources_to_json(sources);
});

7
daemon/json.cpp
View File

@ -26,6 +26,7 @@
#include "json.hpp"
#include "log.hpp"
#include "utils.hpp"
static inline std::string remove_undesired_chars(const std::string& s) {
@ -75,6 +76,7 @@ std::string config_to_json(const Config& config) {
std::stringstream ss;
ss << "{"
<< "\n \"http_port\": " << config.get_http_port()
<< ",\n \"rtsp_port\": " << config.get_rtsp_port()
<< ",\n \"http_base_dir\": \"" << config.get_http_base_dir() << "\""
<< ",\n \"log_severity\": " << config.get_log_severity()
<< ",\n \"playout_delay\": " << config.get_playout_delay()
@ -94,6 +96,7 @@ std::string config_to_json(const Config& config) {
<< ",\n \"mdns_enabled\": " << std::boolalpha << config.get_mdns_enabled()
<< ",\n \"mac_addr\": \"" << escape_json(config.get_mac_addr_str()) << "\""
<< ",\n \"ip_addr\": \"" << escape_json(config.get_ip_addr_str()) << "\""
<< ",\n \"node_id\": \"" << escape_json(get_node_id()) << "\""
<< "\n}\n";
return ss.str();
}
@ -265,6 +268,8 @@ Config json_to_config_(std::istream& js, Config& config) {
for (auto const& [key, val] : pt) {
if (key == "http_port") {
config.set_http_port(val.get_value<int>());
} else if (key == "rtsp_port") {
config.set_rtsp_port(val.get_value<int>());
} else if (key == "http_base_dir") {
config.set_http_base_dir(remove_undesired_chars(val.get_value<std::string>()));
} else if (key == "log_severity") {
@ -299,7 +304,7 @@ Config json_to_config_(std::istream& js, Config& config) {
config.set_syslog_proto(remove_undesired_chars(val.get_value<std::string>()));
} else if (key == "syslog_server") {
config.set_syslog_server(remove_undesired_chars(val.get_value<std::string>()));
} else if (key == "mac_addr" || key == "ip_addr") {
} else if (key == "mac_addr" || key == "ip_addr" || key == "node_id" ) {
/* ignored */
} else {
std::cerr << "Warning: unkown configuration option " << key

15
daemon/main.cpp
View File

@ -24,6 +24,7 @@
#include "config.hpp"
#include "driver_manager.hpp"
#include "http_server.hpp"
#include "rtsp_server.hpp"
#include "log.hpp"
#include "session_manager.hpp"
#include "interface.hpp"
@ -116,6 +117,12 @@ int main(int argc, char* argv[]) {
std::string("SessionManager:: init failed"));
}
/* start rtsp server */
RtspServer rtsp_server(session_manager, config);
if (!rtsp_server.init()) {
throw std::runtime_error(std::string("RtspServer:: init failed"));
}
/* start browser */
auto browser = Browser::create(config);
if (browser == nullptr || !browser->init()) {
@ -147,6 +154,8 @@ int main(int argc, char* argv[]) {
break;
}
rtsp_server.process();
std::this_thread::sleep_for(std::chrono::seconds(1));
}
@ -165,6 +174,12 @@ int main(int argc, char* argv[]) {
std::string("Browser:: terminate failed"));
}
/* stop rtsp server */
if (!rtsp_server.terminate()) {
throw std::runtime_error(
std::string("RtspServer:: terminate failed"));
}
/* stop session manager */
if (!session_manager->terminate()) {
throw std::runtime_error(

63
daemon/mdns_client.cpp
View File

@ -44,7 +44,7 @@ void MDNSClient::resolve_callback(AvahiServiceResolver* r,
/* Called whenever a service has been resolved successfully or timed out */
switch (event) {
case AVAHI_RESOLVER_FAILURE:
BOOST_LOG_TRIVIAL(error) << "avahi_client:: (Resolver) failed to resolve "
BOOST_LOG_TRIVIAL(error) << "mdns_client:: (Resolver) failed to resolve "
<< "service " << name << " of type " << type
<< " in domain " << domain << " : "
<< avahi_strerror(avahi_client_errno(
@ -52,12 +52,18 @@ void MDNSClient::resolve_callback(AvahiServiceResolver* r,
break;
case AVAHI_RESOLVER_FOUND:
BOOST_LOG_TRIVIAL(debug) << "avahi_client:: (Resolver) "
BOOST_LOG_TRIVIAL(debug) << "mdns_client:: (Resolver) "
<< "service " << name << " of type " << type
<< " in domain " << domain;
char addr[AVAHI_ADDRESS_STR_MAX];
avahi_address_snprint(addr, sizeof(addr), address);
if ((flags & AVAHI_LOOKUP_RESULT_LOCAL) &&
(mdns.config_->get_interface_name() == "lo")) {
::strncpy(addr, mdns.config_->get_ip_addr_str().c_str(), sizeof addr - 1);
addr[sizeof addr - 1] = 0;
} else {
avahi_address_snprint(addr, sizeof(addr), address);
}
char info[256];
snprintf(info, sizeof(info),
@ -67,27 +73,30 @@ void MDNSClient::resolve_callback(AvahiServiceResolver* r,
"wide_area: %i, "
"multicast: %i, "
"cached: %i",
host_name, port, addr, !!(flags & AVAHI_LOOKUP_RESULT_LOCAL),
host_name, port, addr,
!!(flags & AVAHI_LOOKUP_RESULT_LOCAL),
!!(flags & AVAHI_LOOKUP_RESULT_OUR_OWN),
!!(flags & AVAHI_LOOKUP_RESULT_WIDE_AREA),
!!(flags & AVAHI_LOOKUP_RESULT_MULTICAST),
!!(flags & AVAHI_LOOKUP_RESULT_CACHED));
BOOST_LOG_TRIVIAL(debug) << "avahi_client:: (Resolver) " << info;
BOOST_LOG_TRIVIAL(debug) << "mdns_client:: (Resolver) " << info;
boost::system::error_code ec;
boost::asio::ip::address_v4::from_string(addr, ec);
if (!ec) {
/* if valid IPv4 address retrieve source data via RTSP */
/* if not on loopback interface we don't want to receive self announced sessions
* or if on loopback interface we want only local announced sessions */
if ((!(flags & AVAHI_LOOKUP_RESULT_LOCAL) &&
(mdns.config_->get_interface_name() != "lo")) ||
((flags & AVAHI_LOOKUP_RESULT_LOCAL) &&
(mdns.config_->get_interface_name() == "lo"))) {
std::lock_guard<std::mutex> lock(mdns.sources_res_mutex_);
/* have fun ;-) */
mdns.sources_res_.emplace_back(std::async(
std::launch::async,
[&mdns, name_ = std::forward<std::string>(name),
[&mdns,
name_ = std::forward<std::string>(name),
domain_ = std::forward<std::string>(domain),
addr_ = std::forward<std::string>(addr),
port_ = std::forward<std::string>(std::to_string(port))] {
auto res = RTSPClient::describe(std::string("/by-name/") + name_,
auto res = RtspClient::describe(std::string("/by-name/") + name_,
addr_, port_);
if (res.first) {
mdns.on_new_rtsp_source(name_, domain_, res.second);
@ -115,14 +124,14 @@ void MDNSClient::browse_callback(AvahiServiceBrowser* b,
* from the LAN */
switch (event) {
case AVAHI_BROWSER_FAILURE:
BOOST_LOG_TRIVIAL(fatal) << "avahi_client:: (Browser) "
BOOST_LOG_TRIVIAL(fatal) << "mdns_client:: (Browser) "
<< avahi_strerror(avahi_client_errno(
avahi_service_browser_get_client(b)));
avahi_threaded_poll_quit(mdns.poll_.get());
return;
case AVAHI_BROWSER_NEW:
BOOST_LOG_TRIVIAL(info) << "avahi_client:: (Browser) NEW: "
BOOST_LOG_TRIVIAL(info) << "mdns_client:: (Browser) NEW: "
<< "service " << name << " of type " << type
<< " in domain " << domain;
/* We ignore the returned resolver object. In the callback
@ -134,25 +143,25 @@ void MDNSClient::browse_callback(AvahiServiceBrowser* b,
AVAHI_LOOKUP_NO_TXT, resolve_callback,
&mdns))) {
BOOST_LOG_TRIVIAL(error)
<< "avahi_client:: "
<< "mdns_client:: "
<< "Failed to resolve service " << name << " : "
<< avahi_strerror(avahi_client_errno(mdns.client_.get()));
}
break;
case AVAHI_BROWSER_REMOVE:
BOOST_LOG_TRIVIAL(info) << "avahi_client:: (Browser) REMOVE: "
BOOST_LOG_TRIVIAL(info) << "mdns_client:: (Browser) REMOVE: "
<< "service " << name << " of type " << type
<< " in domain " << domain;
mdns.on_remove_rtsp_source(name, domain);
break;
case AVAHI_BROWSER_ALL_FOR_NOW:
BOOST_LOG_TRIVIAL(debug) << "avahi_client:: (Browser) ALL_FOR_NOW";
BOOST_LOG_TRIVIAL(debug) << "mdns_client:: (Browser) ALL_FOR_NOW";
break;
case AVAHI_BROWSER_CACHE_EXHAUSTED:
BOOST_LOG_TRIVIAL(debug) << "avahi_client:: (Browser) CACHE_EXHAUSTED";
BOOST_LOG_TRIVIAL(debug) << "mdns_client:: (Browser) CACHE_EXHAUSTED";
break;
}
}
@ -165,7 +174,7 @@ void MDNSClient::client_callback(AvahiClient* client,
switch (state) {
case AVAHI_CLIENT_FAILURE:
BOOST_LOG_TRIVIAL(fatal) << "avahi_client:: server connection failure: "
BOOST_LOG_TRIVIAL(fatal) << "mdns_client:: server connection failure: "
<< avahi_strerror(avahi_client_errno(client));
/* TODO reconnect if disconnected */
avahi_threaded_poll_quit(mdns.poll_.get());
@ -177,11 +186,12 @@ void MDNSClient::client_callback(AvahiClient* client,
/* Create the service browser */
mdns.sb_.reset(avahi_service_browser_new(client,
mdns.config_->get_interface_idx(),
AVAHI_PROTO_INET, "_rtsp._tcp", nullptr,
{}, browse_callback, &mdns));
AVAHI_PROTO_INET,
"_ravenna_session._sub._rtsp._tcp",
nullptr, {}, browse_callback, &mdns));
if (mdns.sb_ == nullptr) {
BOOST_LOG_TRIVIAL(fatal)
<< "avahi_client:: failed to create service browser: "
<< "mdns_client:: failed to create service browser: "
<< avahi_strerror(avahi_client_errno(mdns.client_.get()));
avahi_threaded_poll_quit(mdns.poll_.get());
}
@ -204,7 +214,7 @@ bool MDNSClient::init() {
poll_.reset(avahi_threaded_poll_new());
if (poll_ == nullptr) {
BOOST_LOG_TRIVIAL(fatal)
<< "avahi_client:: failed to create threaded poll object";
<< "mdns_client:: failed to create threaded poll object";
return false;
}
@ -215,7 +225,7 @@ bool MDNSClient::init() {
&error));
if (client_ == nullptr) {
BOOST_LOG_TRIVIAL(fatal)
<< "avahi_client:: failed to create client: " << avahi_strerror(error);
<< "mdns_client:: failed to create client: " << avahi_strerror(error);
return false;
}
@ -250,11 +260,10 @@ bool MDNSClient::terminate() {
if (running_) {
running_ = false;
#ifdef _USE_AVAHI_
/* remove all completed results and populate remote sources list */
/* wait for all pending results and remove from list */
std::lock_guard<std::mutex> lock(sources_res_mutex_);
BOOST_LOG_TRIVIAL(fatal) << "avahi_client:: waiting for "
<< sources_res_.size() << " RTSP clients";
BOOST_LOG_TRIVIAL(info) << "mdns_client:: waiting for "
<< sources_res_.size() << " RTSP clients";
sources_res_.remove_if([](auto& result) {
if (result.valid()) {
result.wait();

16
daemon/mdns_client.hpp
View File

@ -17,8 +17,8 @@
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//
#ifndef _AVAHI_CLIENT_HPP_
#define _AVAHI_CLIENT_HPP_
#ifndef _MDNS_CLIENT_HPP_
#define _MDNS_CLIENT_HPP_
#ifdef _USE_AVAHI_
#include <avahi-client/client.h>
@ -50,7 +50,7 @@ class MDNSClient {
protected:
virtual void on_new_rtsp_source(const std::string& name,
const std::string& domain,
const RTSPSSource& source) = 0;
const RtspSource& source) = 0;
virtual void on_remove_rtsp_source(const std::string& name,
const std::string& domain) = 0;
@ -59,13 +59,14 @@ class MDNSClient {
std::mutex sources_res_mutex_;
std::atomic_bool running_{false};
std::shared_ptr<Config> config_;
#ifdef _USE_AVAHI_
/* order is important here */
std::unique_ptr<AvahiThreadedPoll, decltype(&avahi_threaded_poll_free)> poll_{
nullptr, &avahi_threaded_poll_free};
std::unique_ptr< ::AvahiClient, decltype(&avahi_client_free)> client_{
nullptr, &avahi_client_free};
std::unique_ptr<AvahiThreadedPoll, decltype(&avahi_threaded_poll_free)>
poll_{nullptr, &avahi_threaded_poll_free};
std::unique_ptr<AvahiClient, decltype(&avahi_client_free)>
client_{nullptr, &avahi_client_free};
std::unique_ptr<AvahiServiceBrowser, decltype(&avahi_service_browser_free)>
sb_{nullptr, &avahi_service_browser_free};
@ -95,7 +96,6 @@ class MDNSClient {
AvahiClientState state,
void* userdata);
std::shared_ptr<Config> config_;
#endif
};

338
daemon/mdns_server.cpp Normal file
View File

@ -0,0 +1,338 @@
//
// mdns_server.cpp
//
// Copyright (c) 2019 2020 Andrea Bondavalli. All rights reserved.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//
#include "mdns_server.hpp"
#include <boost/asio.hpp>
#include "config.hpp"
#include "interface.hpp"
#include "log.hpp"
#include "utils.hpp"
#ifdef _USE_AVAHI_
struct AvahiLockGuard {
AvahiLockGuard() = delete;
AvahiLockGuard(AvahiThreadedPoll* poll) : poll_(poll) {
if (poll_ != nullptr) {
avahi_threaded_poll_lock(poll_);
}
}
~AvahiLockGuard() {
if (poll_ != nullptr) {
avahi_threaded_poll_unlock(poll_);
}
}
private:
AvahiThreadedPoll* poll_{nullptr};
};
void MDNSServer::entry_group_callback(AvahiEntryGroup* g,
AvahiEntryGroupState state,
void* userdata) {
MDNSServer& mdns = *(reinterpret_cast<MDNSServer*>(userdata));
/* Called whenever the entry group state changes */
auto it = mdns.groups_.right.find(g);
if (it == mdns.groups_.right.end()) {
BOOST_LOG_TRIVIAL(debug)
<< "mdns_server:: cannot find name associated to group, skipping ...";
return;
}
switch (state) {
case AVAHI_ENTRY_GROUP_ESTABLISHED:
/* The entry group has been established successfully */
BOOST_LOG_TRIVIAL(debug) << "mdns_server:: entry group name ("
<< it->second << ") established";
break;
case AVAHI_ENTRY_GROUP_COLLISION: {
BOOST_LOG_TRIVIAL(warning)
<< "mdns_server:: entry group name (" << it->second << ") collision";
break;
}
case AVAHI_ENTRY_GROUP_FAILURE:
BOOST_LOG_TRIVIAL(error) << "mdns_server:: entry group name "
<< "(" << it->second << ") "
<< "failure"
<< avahi_strerror(avahi_client_errno(
avahi_entry_group_get_client(g)));
/* Some kind of failure happened while we were registering our services */
avahi_threaded_poll_quit(mdns.poll_.get());
break;
case AVAHI_ENTRY_GROUP_UNCOMMITED:
case AVAHI_ENTRY_GROUP_REGISTERING:
break;
}
}
bool MDNSServer::create_services(AvahiClient* client) {
if (!config_->get_mdns_enabled()) {
return false;
}
std::unique_ptr<AvahiEntryGroup, decltype(&avahi_entry_group_free)> group{
avahi_entry_group_new(client, entry_group_callback, this),
&avahi_entry_group_free};
if (group == nullptr) {
BOOST_LOG_TRIVIAL(error)
<< "mdns_server:: create_services() failed avahi_entry_group_new(): "
<< avahi_strerror(avahi_client_errno(client));
return false;
}
/* register ravenna services, without user defined name */
int ret = avahi_entry_group_add_service(
group.get(), this->config_->get_interface_idx(), AVAHI_PROTO_INET, {},
node_id_.c_str(), "_http._tcp", nullptr, nullptr,
this->config_->get_http_port(), nullptr);
if (ret < 0) {
BOOST_LOG_TRIVIAL(error)
<< "mdns_server:: add_service() " << node_id_
<< "failed to add entry _http._tcp" << avahi_strerror(ret);
return false;
}
BOOST_LOG_TRIVIAL(info) << "mdns_server:: adding service _http._tcp for "
<< node_id_;
ret = avahi_entry_group_add_service_subtype(
group.get(), this->config_->get_interface_idx(), AVAHI_PROTO_INET, {},
node_id_.c_str(), "_http._tcp", nullptr, "_ravenna._sub._http._tcp");
if (ret < 0) {
BOOST_LOG_TRIVIAL(error) << "mdns_server:: add_service() " << node_id_
<< "failed to add subtype _ravenna._sub._http._tcp"
<< avahi_strerror(ret);
return false;
}
ret = avahi_entry_group_add_service(
group.get(), this->config_->get_interface_idx(), AVAHI_PROTO_INET, {},
node_id_.c_str(), "_rtsp._tcp", nullptr, nullptr,
this->config_->get_rtsp_port(), nullptr);
if (ret < 0) {
BOOST_LOG_TRIVIAL(error)
<< "mdns_server:: add_service() " << node_id_
<< "failed to add entry _rtsp._tcp" << avahi_strerror(ret);
return false;
}
BOOST_LOG_TRIVIAL(info) << "mdns_server:: adding service _rtsp._tcp for "
<< node_id_;
ret = avahi_entry_group_add_service_subtype(
group.get(), this->config_->get_interface_idx(), AVAHI_PROTO_INET, {},
node_id_.c_str(), "_rtsp._tcp", nullptr, "_ravenna._sub._rtsp._tcp");
if (ret < 0) {
BOOST_LOG_TRIVIAL(error) << "mdns_server:: add_service() " << node_id_
<< "failed to add subtype _ravenna._sub._rtsp._tcp"
<< avahi_strerror(ret);
return false;
}
/* Tell the server to register the service */
ret = avahi_entry_group_commit(group.get());
if (ret < 0) {
BOOST_LOG_TRIVIAL(error)
<< "mdns_server:: create_services() failed to commit entry group "
<< avahi_strerror(ret);
return false;
}
groups_.insert(entry_group_bimap_t::value_type(node_id_, group.release()));
return true;
}
#endif
bool MDNSServer::add_service(const std::string& name, const std::string& sdp) {
if (!running_ || !config_->get_mdns_enabled()) {
return false;
}
#ifdef _USE_AVAHI_
AvahiLockGuard avahi_lock(this->poll_.get());
if (avahi_client_get_state(client_.get()) != AVAHI_CLIENT_S_RUNNING) {
BOOST_LOG_TRIVIAL(error)
<< "mdns_server:: add_service() failed client is not running";
return false;
}
std::unique_ptr<AvahiEntryGroup, decltype(&avahi_entry_group_free)> group{
avahi_entry_group_new(client_.get(), entry_group_callback, this),
&avahi_entry_group_free};
if (group == nullptr) {
BOOST_LOG_TRIVIAL(error)
<< "mdns_server:: add_service() failed avahi_entry_group_new(): "
<< avahi_strerror(avahi_client_errno(client_.get()));
return false;
}
std::stringstream ss;
ss << node_id_ << " " << name;
int ret = avahi_entry_group_add_service(
group.get(), this->config_->get_interface_idx(), AVAHI_PROTO_INET, {},
ss.str().c_str(), "_rtsp._tcp", nullptr, nullptr,
this->config_->get_rtsp_port(), nullptr);
if (ret < 0) {
BOOST_LOG_TRIVIAL(error)
<< "mdns_server:: add_service() " << ss.str()
<< "failed to add service _rtsp._tcp" << avahi_strerror(ret);
return false;
}
ret = avahi_entry_group_add_service_subtype(
group.get(), this->config_->get_interface_idx(), AVAHI_PROTO_INET, {},
ss.str().c_str(), "_rtsp._tcp", nullptr,
"_ravenna_session._sub._rtsp._tcp");
if (ret < 0) {
BOOST_LOG_TRIVIAL(error)
<< "mdns_server:: add_service() " << ss.str()
<< "failed to add subtype _ravenna_session._sub._rtsp._tcp"
<< avahi_strerror(ret);
return false;
}
/* Tell the server to register the service */
ret = avahi_entry_group_commit(group.get());
if (ret < 0) {
BOOST_LOG_TRIVIAL(error)
<< "mdns_server:: add_service() " << ss.str()
<< "failed to commit entry group " << avahi_strerror(ret);
return false;
}
BOOST_LOG_TRIVIAL(info) << "mdns_server:: adding service for " << ss.str();
groups_.insert(entry_group_bimap_t::value_type(name, group.release()));
#endif
return true;
}
bool MDNSServer::remove_service(const std::string& name) {
if (!running_ || !config_->get_mdns_enabled()) {
return false;
}
#ifdef _USE_AVAHI_
AvahiLockGuard avahi_lock(poll_.get());
if (avahi_client_get_state(client_.get()) != AVAHI_CLIENT_S_RUNNING) {
BOOST_LOG_TRIVIAL(error)
<< "mdns_server:: remove_sub_service() failed client is not running";
return false;
}
auto it = groups_.left.find(name);
if (it == groups_.left.end()) {
return false;
}
BOOST_LOG_TRIVIAL(info) << "mdns_server:: removing service _rtsp._tcp for "
<< name;
avahi_entry_group_free(it->second);
groups_.left.erase(name);
#endif
return true;
}
#ifdef _USE_AVAHI_
void MDNSServer::client_callback(AvahiClient* client,
AvahiClientState state,
void* userdata) {
MDNSServer& mdns = *(reinterpret_cast<MDNSServer*>(userdata));
/* Called whenever the client or server state changes */
switch (state) {
case AVAHI_CLIENT_FAILURE:
BOOST_LOG_TRIVIAL(fatal) << "mdns_server:: server connection failure: "
<< avahi_strerror(avahi_client_errno(client));
/* TODO reconnect if disconnected */
avahi_threaded_poll_quit(mdns.poll_.get());
break;
case AVAHI_CLIENT_S_RUNNING:
/* The server has startup successfully and registered its host
* name on the network, so it's time to create our services */
if (!mdns.create_services(client)) {
avahi_threaded_poll_quit(mdns.poll_.get());
}
break;
case AVAHI_CLIENT_S_REGISTERING:
/* The server records are now being established. This
* might be caused by a host name change. We need to wait
* for our own records to register until the host name is
* properly esatblished. */
case AVAHI_CLIENT_S_COLLISION:
/* Let's drop our registered services. When the server is back
* in AVAHI_SERVER_RUNNING state we will register them
* again with the new host name. */
/* TODO */
break;
case AVAHI_CLIENT_CONNECTING:
break;
}
}
#endif
bool MDNSServer::init() {
if (running_) {
return true;
}
#ifdef _USE_AVAHI_
/* allocate poll loop object */
poll_.reset(avahi_threaded_poll_new());
if (poll_ == nullptr) {
BOOST_LOG_TRIVIAL(fatal)
<< "mdns_server:: failed to create threaded poll object";
return false;
}
/* allocate a new client */
int error;
client_.reset(avahi_client_new(avahi_threaded_poll_get(poll_.get()),
AVAHI_CLIENT_NO_FAIL, client_callback, this,
&error));
if (client_ == nullptr) {
BOOST_LOG_TRIVIAL(fatal)
<< "mdns_server:: failed to create client: " << avahi_strerror(error);
return false;
}
(void)avahi_threaded_poll_start(poll_.get());
#endif
running_ = true;
return true;
}
bool MDNSServer::terminate() {
if (running_) {
running_ = false;
#ifdef _USE_AVAHI_
/* remove base services */
groups_.left.erase(node_id_);
avahi_threaded_poll_stop(poll_.get());
#endif
}
return true;
}

81
daemon/mdns_server.hpp Normal file
View File

@ -0,0 +1,81 @@
//
// mdns_server.hpp
//
// Copyright (c) 2019 2020 Andrea Bondavalli. All rights reserved.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//
#ifndef _MDNS_SERVER_HPP_
#define _MDNS_SERVER_HPP_
#ifdef _USE_AVAHI_
#include <avahi-client/client.h>
#include <avahi-client/publish.h>
#include <avahi-common/error.h>
#include <avahi-common/malloc.h>
#include <avahi-common/thread-watch.h>
#endif
#include <atomic>
#include <boost/bimap.hpp>
#include <mutex>
#include <thread>
#include "config.hpp"
#include "utils.hpp"
class MDNSServer {
public:
MDNSServer(std::shared_ptr<Config> config) : config_(config){};
MDNSServer() = delete;
MDNSServer(const MDNSServer&) = delete;
MDNSServer& operator=(const MDNSServer&) = delete;
virtual ~MDNSServer() { terminate(); };
virtual bool init();
virtual bool terminate();
bool add_service(const std::string& name, const std::string& sdp);
bool remove_service(const std::string& name);
protected:
std::atomic_bool running_{false};
std::shared_ptr<Config> config_;
std::string node_id_{get_node_id()};
#ifdef _USE_AVAHI_
using entry_group_bimap_t =
boost::bimap<std::string /* name */, AvahiEntryGroup*>;
entry_group_bimap_t groups_;
/* order is important here */
std::unique_ptr<AvahiThreadedPoll, decltype(&avahi_threaded_poll_free)> poll_{
nullptr, &avahi_threaded_poll_free};
std::unique_ptr<AvahiClient, decltype(&avahi_client_free)> client_{
nullptr, &avahi_client_free};
static void entry_group_callback(AvahiEntryGroup* g,
AvahiEntryGroupState state,
void* userdata);
static void client_callback(AvahiClient* c,
AvahiClientState state,
void* userdata);
bool create_services(AvahiClient* client);
#endif
};
#endif

8
daemon/rtsp_client.cpp
View File

@ -83,10 +83,10 @@ RtspResponse read_response(tcp::iostream& s, uint16_t max_length) {
return res;
}
std::pair<bool, RTSPSSource> RTSPClient::describe(const std::string& path,
std::pair<bool, RtspSource> RtspClient::describe(const std::string& path,
const std::string& address,
const std::string& port) {
RTSPSSource rs;
RtspSource rs;
bool success{false};
try {
tcp::iostream s;
@ -155,8 +155,8 @@ std::pair<bool, RTSPSSource> RTSPClient::describe(const std::string& path,
std::stringstream ss;
ss << "rtsp:" << std::hex
<< crc16(reinterpret_cast<const uint8_t*>(res.body.c_str()),
res.body.length())
<< std::hex << ip::address_v4::from_string(address.c_str()).to_ulong();
res.body.length());
/*<< std::hex << ip::address_v4::from_string(address.c_str()).to_ulong();*/
rs.id = ss.str();
rs.source = "mDNS";

10
daemon/rtsp_client.hpp
View File

@ -17,23 +17,23 @@
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//
#ifndef _RTSP_HPP_
#define _RTSP_HPP_
#ifndef _RTSP_CLIENT_HPP_
#define _RTSP_CLIENT_HPP_
struct RTSPSSource {
struct RtspSource {
std::string id;
std::string source;
std::string address;
std::string sdp;
};
class RTSPClient {
class RtspClient {
public:
constexpr static uint16_t max_body_length = 4096; // byte
constexpr static uint16_t client_timeout = 10; // sec
constexpr static const char dft_port[] = "554";
static std::pair<bool, RTSPSSource> describe(
static std::pair<bool, RtspSource> describe(
const std::string& path,
const std::string& address,
const std::string& port = dft_port);

239
daemon/rtsp_server.cpp Normal file
View File

@ -0,0 +1,239 @@
// rtsp_server.cpp
//
// Copyright (c) 2019 2020 Andrea Bondavalli. All rights reserved.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
#include "rtsp_server.hpp"
#include "utils.hpp"
using boost::asio::ip::tcp;
void RtspServer::worker() {
io_service_.run();
}
void RtspServer::process() {
/* cleanup of expired sessions */
std::lock_guard<std::mutex> lock(mutex_);
for (unsigned int i = 0; i < sessions_.size(); i++) {
if (duration_cast<second_t>(steady_clock::now() - sessions_start_point_[i])
.count() > RtspSession::session_tout_secs) {
auto session = sessions_[i].lock();
if (session != nullptr) {
session->stop();
}
}
}
}
void RtspServer::accept() {
acceptor_.async_accept(socket_, [this](boost::system::error_code ec) {
if (!ec) {
std::lock_guard<std::mutex> lock(mutex_);
/* check for free sessions */
unsigned int i = 0;
for (; i < sessions_.size(); i++) {
if (sessions_[i].use_count() == 0) {
auto session = std::make_shared<RtspSession>(session_manager_,
std::move(socket_));
sessions_[i] = session;
sessions_start_point_[i] = steady_clock::now();
session->start();
break;
}
}
if (i == sessions_.size()) {
BOOST_LOG_TRIVIAL(warning)
<< "rtsp_server:: too many clients connected, "
<< socket_.remote_endpoint() << " closing...";
socket_.close();
}
}
accept();
});
}
bool RtspSession::process_request() {
/*
DESCRIBE rtsp://127.0.0.1:8080/by-name/test RTSP/1.0
CSeq: 312
User-Agent: pippo
Accept: application/sdp
*/
data_[length_] = 0;
std::stringstream sstream(data_);
/* read the request */
if (!getline(sstream, request_, '\n')) {
return false;
}
consumed_ = request_.length() + 1;
boost::trim(request_);
std::vector<std::string> fields;
split(fields, request_, boost::is_any_of(" "));
if (fields.size() < 3) {
return false;
}
/* read the header */
bool is_end{false};
std::string header;
while (getline(sstream, header, '\n')) {
consumed_ += header.length() + 1;
if (header == "" || header == "\r") {
is_end = true;
break;
}
boost::to_lower(header);
boost::trim(header);
if (header.rfind("cseq:", 0) != std::string::npos) {
try {
cseq_ = stoi(header.substr(5));
} catch (...) {
break;
}
}
}
if (!is_end) {
return false;
}
if (cseq_ < 0) {
BOOST_LOG_TRIVIAL(error) << "rtsp_server:: CSeq not specified from "
<< socket_.remote_endpoint();
send_error(400, "Bad Request");
} else if (fields[2].substr(0, 5) != "RTSP/") {
BOOST_LOG_TRIVIAL(error)
<< "rtsp_server:: no RTSP specified from " << socket_.remote_endpoint();
send_error(400, "Bad Request");
} else if (fields[0] != "DESCRIBE") {
send_error(405, "Method Not Allowed");
} /*else if (fields[1].substr(0, 7) != "rtsp://") {
BOOST_LOG_TRIVIAL(error)
<< "rtsp_server:: no rtsp protocol from " << socket_.remote_endpoint();
send_error(404, "Not supported");
} */
else {
boost::trim(fields[1]);
build_response(fields[1]);
}
return true;
}
void RtspSession::build_response(const std::string& url) {
auto const [ok, protocol, host, port, path] = parse_url(url);
if (!ok) {
BOOST_LOG_TRIVIAL(error) << "rtsp_server:: cannot parse URL " << url
<< " from " << socket_.remote_endpoint();
send_error(400, "Bad Request");
return;
}
auto base_path = std::string("/by-name/") + get_node_id() + " ";
uint8_t id = SessionManager::stream_id_max + 1;
if (path.rfind(base_path) != std::string::npos) {
/* extract the source name from path and retrive the id */
id = session_manager_->get_source_id(path.substr(base_path.length()));
} else if (path.rfind("/by-id/") != std::string::npos) {
try {
id = stoi(path.substr(7));
} catch (...) {
}
}
if (id != (SessionManager::stream_id_max + 1)) {
std::string sdp;
if (!session_manager_->get_source_sdp(id, sdp)) {
std::stringstream ss;
ss << "RTSP/1.0 200 OK\r\n"
<< "CSeq: " << cseq_ << "\r\n"
<< "Content-Length: " << sdp.length() << "\r\n"
<< "Content-Type: application/sdp\r\n"
<< "\r\n"
<< sdp;
BOOST_LOG_TRIVIAL(info)
<< "rtsp_server:: " << request_ << " response 200 to "
<< socket_.remote_endpoint();
send_response(ss.str());
return;
}
}
send_error(404, "Not found");
}
void RtspSession::read_request() {
auto self(shared_from_this());
if (length_ == max_length) {
/* request cannot be consumed and we execeed max length */
stop();
} else {
socket_.async_read_some(
boost::asio::buffer(data_ + length_, max_length - length_),
[this, self](boost::system::error_code ec, std::size_t length) {
if (!ec) {
BOOST_LOG_TRIVIAL(debug) << "rtsp_server:: received " << length
<< " from " << socket_.remote_endpoint();
length_ += length;
while (length_ && process_request()) {
/* step to the next request */
std::memmove(data_, data_ + consumed_, length_ - consumed_);
length_ -= consumed_;
cseq_ = -1;
}
/* read more data */
read_request();
}
});
}
}
void RtspSession::send_error(int status_code, const std::string& description) {
BOOST_LOG_TRIVIAL(error) << "rtsp_server:: " << request_ << " response "
<< status_code << " to "
<< socket_.remote_endpoint();
std::stringstream ss;
ss << "RTSP/1.0 " << status_code << " " << description << "\r\n";
if (cseq_ >= 0) {
ss << "CSeq: " << cseq_ << "\r\n";
}
ss << "\n\r";
send_response(ss.str());
}
void RtspSession::send_response(const std::string& response) {
auto self(shared_from_this());
boost::asio::async_write(
socket_, boost::asio::buffer(response.c_str(), response.length()),
[self](boost::system::error_code ec, std::size_t /*length*/) {
if (!ec) {
// we accept multiple requests within timeout
// stop();
}
});
}
void RtspSession::start() {
BOOST_LOG_TRIVIAL(debug) << "rtsp_server:: starting session with "
<< socket_.remote_endpoint();
read_request();
}
void RtspSession::stop() {
BOOST_LOG_TRIVIAL(debug) << "rtsp_server:: stopping session with "
<< socket_.remote_endpoint();
socket_.close();
}

117
daemon/rtsp_server.hpp Normal file
View File

@ -0,0 +1,117 @@
// rtsp_server.hpp
//
// Copyright (c) 2019 2020 Andrea Bondavalli. All rights reserved.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
#include <boost/algorithm/string.hpp>
#include <boost/asio.hpp>
#include <chrono>
#include <cstdlib>
#include <iostream>
#include <memory>
#include "session_manager.hpp"
#ifndef _RTSP_SERVER_HPP_
#define _RTSP_SERVER_HPP_
using namespace std::chrono;
using boost::asio::deadline_timer;
using boost::asio::ip::tcp;
using second_t = duration<double, std::ratio<1> >;
class RtspSession : public std::enable_shared_from_this<RtspSession> {
public:
constexpr static uint16_t max_length = 4096; // byte
constexpr static uint16_t session_tout_secs = 10; // sec
RtspSession(std::shared_ptr<SessionManager> session_manager,
tcp::socket socket)
: session_manager_(session_manager),
socket_(std::move(socket)),
length_{0},
cseq_{-1},
consumed_{0} {}
virtual ~RtspSession() {
BOOST_LOG_TRIVIAL(debug) << "rtsp_server:: session end";
}
void start();
void stop();
private:
bool process_request();
void build_response(const std::string& url);
void read_request();
void send_error(int status_code, const std::string& description);
void send_response(const std::string& response);
std::shared_ptr<SessionManager> session_manager_;
tcp::socket socket_;
char data_[max_length + 1];
std::string request_;
size_t length_{0};
int32_t cseq_{-1};
size_t consumed_{0};
};
class RtspServer {
public:
RtspServer() = delete;
RtspServer(std::shared_ptr<SessionManager> session_manager,
std::shared_ptr<Config> config)
: session_manager_(session_manager),
config_(config),
sessions_(SessionManager::stream_id_max),
sessions_start_point_(SessionManager::stream_id_max),
acceptor_(io_service_,
tcp::endpoint(boost::asio::ip::address::from_string(
config_->get_ip_addr_str()),
config_->get_rtsp_port())),
socket_(io_service_) {}
bool init() {
accept();
/* start rtsp server on a separate thread */
res_ = std::async(std::launch::async, &RtspServer::worker, this);
return true;
}
bool terminate() {
BOOST_LOG_TRIVIAL(info) << "rtsp_server: stopping ... ";
io_service_.stop();
res_.get();
return true;
}
void process();
private:
void worker();
void accept();
std::mutex mutex_;
boost::asio::io_service io_service_;
std::shared_ptr<SessionManager> session_manager_;
std::shared_ptr<Config> config_;
std::vector<std::weak_ptr<RtspSession> > sessions_;
std::vector<time_point<steady_clock> > sessions_start_point_;
tcp::acceptor acceptor_;
tcp::socket socket_{io_service_};
std::future<void> res_;
};
#endif

3
daemon/sap.cpp
View File

@ -45,7 +45,8 @@ bool SAP::set_multicast_interface(const std::string& interface_ip) {
<< "sap::outbound_interface option " << ec.message();
return false;
}
ip::multicast::enable_loopback el_option(true);
/* we don't want receive self announced sessions */
ip::multicast::enable_loopback el_option(false);
socket_.set_option(el_option, ec);
if (ec) {
BOOST_LOG_TRIVIAL(error) << "sap::enable_loopback option " << ec.message();

172
daemon/session_manager.cpp
View File

@ -60,46 +60,6 @@ static uint8_t get_codec_word_lenght(const std::string& codec) {
return 0;
}
static std::tuple<bool /* res */,
std::string /* protocol */,
std::string /* host */,
std::string /* port */,
std::string /* path */>
parse_url(const std::string& _url) {
std::string url = httplib::detail::decode_url(_url);
size_t protocol_sep_pos = url.find_first_of("://");
if (protocol_sep_pos == std::string::npos) {
/* no protocol, invalid URL */
return std::make_tuple(false, "", "", "", "");
}
std::string port, host, path("/");
std::string protocol = url.substr(0, protocol_sep_pos);
std::string url_new = url.substr(protocol_sep_pos + 3);
size_t path_sep_pos = url_new.find_first_of("/");
size_t port_sep_pos = url_new.find_first_of(":");
if (port_sep_pos != std::string::npos) {
/* port specified */
if (path_sep_pos != std::string::npos) {
/* path specified */
port = url_new.substr(port_sep_pos + 1, path_sep_pos - port_sep_pos - 1);
path = url_new.substr(path_sep_pos);
} else {
/* path not specified */
port = url_new.substr(port_sep_pos + 1);
}
host = url_new.substr(0, port_sep_pos);
} else if (path_sep_pos != std::string::npos) {
/* port not specified, path specified */
host = url_new.substr(0, path_sep_pos);
path = url_new.substr(path_sep_pos);
} else {
/* port and path not specified */
host = url_new;
}
return std::make_tuple(host.length() > 0, protocol, host, port, path);
}
bool SessionManager::parse_sdp(const std::string sdp, StreamInfo& info) const {
/*
v=0
@ -362,37 +322,35 @@ std::list<StreamSource> SessionManager::get_sources() const {
StreamSource SessionManager::get_source_(uint8_t id,
const StreamInfo& info) const {
StreamSource source;
source.id = id;
source.enabled = info.enabled;
source.name = info.stream.m_cName;
source.io = info.io;
source.max_samples_per_packet = info.stream.m_ui32MaxSamplesPerPacket;
source.codec = info.stream.m_cCodec;
source.ttl = info.stream.m_byTTL;
source.payload_type = info.stream.m_byPayloadType;
source.dscp = info.stream.m_ucDSCP;
source.refclk_ptp_traceable = info.refclk_ptp_traceable;
for (auto i = 0; i < info.stream.m_byNbOfChannels; i++) {
source.map.push_back(info.stream.m_aui32Routing[i]);
}
return source;
return {
id,
info.enabled,
info.stream.m_cName,
info.io,
info.stream.m_ui32MaxSamplesPerPacket,
info.stream.m_cCodec,
info.stream.m_byTTL,
info.stream.m_byPayloadType,
info.stream.m_ucDSCP,
info.refclk_ptp_traceable,
{ info.stream.m_aui32Routing, info.stream.m_aui32Routing +
info.stream.m_byNbOfChannels }
};
}
StreamSink SessionManager::get_sink_(uint8_t id, const StreamInfo& info) const {
StreamSink sink;
sink.id = id;
sink.name = info.stream.m_cName;
sink.io = info.io;
sink.use_sdp = info.sink_use_sdp;
sink.sdp = info.sink_sdp;
sink.source = info.sink_source;
sink.delay = info.stream.m_ui32PlayOutDelay;
sink.ignore_refclk_gmid = info.ignore_refclk_gmid;
for (auto i = 0; i < info.stream.m_byNbOfChannels; i++) {
sink.map.push_back(info.stream.m_aui32Routing[i]);
}
return sink;
return {
id,
info.stream.m_cName,
info.io,
info.sink_use_sdp,
info.sink_source,
info.sink_sdp,
info.stream.m_ui32PlayOutDelay,
info.ignore_refclk_gmid,
{ info.stream.m_aui32Routing, info.stream.m_aui32Routing +
info.stream.m_byNbOfChannels }
};
}
bool SessionManager::load_status() {
@ -455,6 +413,26 @@ static std::array<uint8_t, 6> get_mcast_mac_addr(uint32_t mcast_ip) {
static_cast<uint8_t>(mcast_ip) };
}
uint8_t SessionManager::get_source_id(const std::string& name) const {
const auto it = source_names_.find(name);
return it != source_names_.end() ? it->second : stream_id_max;
}
void SessionManager::on_add_source(const StreamSource& source,
const StreamInfo& info) {
igmp_.join(config_->get_ip_addr_str(),
ip::address_v4(info.stream.m_ui32DestIP).to_string());
mdns_.add_service(source.name, get_source_sdp_(source.id, info));
source_names_[source.name] = source.id;
}
void SessionManager::on_remove_source(const StreamInfo& info) {
igmp_.leave(config_->get_ip_addr_str(),
ip::address_v4(info.stream.m_ui32DestIP).to_string());
mdns_.remove_service(info.stream.m_cName);
source_names_.erase(info.stream.m_cName);
}
std::error_code SessionManager::add_source(const StreamSource& source) {
if (source.id > stream_id_max) {
BOOST_LOG_TRIVIAL(error) << "session_manager:: source id "
@ -504,9 +482,12 @@ std::error_code SessionManager::add_source(const StreamSource& source) {
// remove previous stream if enabled
if ((*it).second.enabled) {
(void)driver_->remove_rtp_stream((*it).second.handle);
igmp_.leave(config_->get_ip_addr_str(),
ip::address_v4((*it).second.stream.m_ui32DestIP).to_string());
on_remove_source((*it).second);
}
} else if (source_names_.find(source.name) != source_names_.end()) {
BOOST_LOG_TRIVIAL(error) << "session_manager:: source name "
<< source.name << " is in use";
return DaemonErrc::stream_name_in_use;
}
std::error_code ret;
@ -519,8 +500,7 @@ std::error_code SessionManager::add_source(const StreamSource& source) {
}
return ret;
}
igmp_.join(config_->get_ip_addr_str(),
ip::address_v4(info.stream.m_ui32DestIP).to_string());
on_add_source(source, info);
}
// update source map
@ -557,7 +537,7 @@ std::string SessionManager::get_source_sdp_(uint32_t id,
ss << "v=0\n"
<< "o=- " << static_cast<unsigned>(id) << " 0 IN IP4 "
<< ip::address_v4(info.stream.m_ui32SrcIP).to_string() << "\n"
<< "s=" << info.stream.m_cName << "\n"
<< "s=" << get_node_id() << " " << info.stream.m_cName << "\n"
<< "c=IN IP4 " << ip::address_v4(info.stream.m_ui32DestIP).to_string()
<< "/" << static_cast<unsigned>(info.stream.m_byTTL) << "\n"
<< "t=0 0\n"
@ -619,8 +599,7 @@ std::error_code SessionManager::remove_source(uint32_t id) {
if (info.enabled) {
ret = driver_->remove_rtp_stream(info.handle);
if (!ret) {
igmp_.leave(config_->get_ip_addr_str(),
ip::address_v4(info.stream.m_ui32DestIP).to_string());
on_remove_source(info);
}
}
if (!ret) {
@ -630,6 +609,24 @@ std::error_code SessionManager::remove_source(uint32_t id) {
return ret;
}
uint8_t SessionManager::get_sink_id(const std::string& name) const {
const auto it = sink_names_.find(name);
return it != sink_names_.end() ? it->second : (stream_id_max + 1);
}
void SessionManager::on_add_sink(const StreamSink& sink,
const StreamInfo& info) {
igmp_.join(config_->get_ip_addr_str(),
ip::address_v4(info.stream.m_ui32DestIP).to_string());
sink_names_[sink.name] = sink.id;
}
void SessionManager::on_remove_sink(const StreamInfo& info) {
igmp_.leave(config_->get_ip_addr_str(),
ip::address_v4(info.stream.m_ui32DestIP).to_string());
sink_names_.erase(info.stream.m_cName);
}
std::error_code SessionManager::add_sink(const StreamSink& sink) {
if (sink.id > stream_id_max) {
BOOST_LOG_TRIVIAL(error) << "session_manager:: sink id "
@ -678,7 +675,7 @@ std::error_code SessionManager::add_sink(const StreamSink& sink) {
}
sdp = std::move(res->body);
} else if (boost::iequals(protocol, "rtsp")) {
auto res = RTSPClient::describe(path, host, port);
auto res = RtspClient::describe(path, host, port);
if (!res.first) {
BOOST_LOG_TRIVIAL(error)
<< "session_manager:: cannot retrieve SDP from URL " << sink.source;
@ -742,8 +739,11 @@ std::error_code SessionManager::add_sink(const StreamSink& sink) {
<< std::to_string(sink.id) << " is in use, updating";
// remove previous stream
(void)driver_->remove_rtp_stream((*it).second.handle);
igmp_.leave(config_->get_ip_addr_str(),
ip::address_v4((*it).second.stream.m_ui32DestIP).to_string());
on_remove_sink((*it).second);
} else if (sink_names_.find(sink.name) != sink_names_.end()) {
BOOST_LOG_TRIVIAL(error) << "session_manager:: sink name "
<< sink.name << " is in use";
return DaemonErrc::stream_name_in_use;
}
auto ret = driver_->add_rtp_stream(info.stream, info.handle);
@ -755,9 +755,7 @@ std::error_code SessionManager::add_sink(const StreamSink& sink) {
return ret;
}
igmp_.join(config_->get_ip_addr_str(),
ip::address_v4(info.stream.m_ui32DestIP).to_string());
on_add_sink(sink, info);
// update sinks map
sinks_[sink.id] = info;
BOOST_LOG_TRIVIAL(info) << "session_manager:: added sink "
@ -785,6 +783,7 @@ std::error_code SessionManager::remove_sink(uint32_t id) {
if (!ret) {
igmp_.leave(config_->get_ip_addr_str(),
ip::address_v4(info.stream.m_ui32DestIP).to_string());
on_remove_sink(info);
sinks_.erase(id);
}
@ -937,16 +936,11 @@ bool SessionManager::worker() {
// return false;
} else {
char ptp_clock_id[24];
uint8_t* pui64GMID = reinterpret_cast<uint8_t*>(&ptp_status.ui64GMID);
snprintf(ptp_clock_id, sizeof(ptp_clock_id),
"%02X-%02X-%02X-%02X-%02X-%02X-%02X-%02X",
(reinterpret_cast<uint8_t*>(&ptp_status.ui64GMID)[0]),
(reinterpret_cast<uint8_t*>(&ptp_status.ui64GMID)[1]),
(reinterpret_cast<uint8_t*>(&ptp_status.ui64GMID)[2]),
(reinterpret_cast<uint8_t*>(&ptp_status.ui64GMID)[3]),
(reinterpret_cast<uint8_t*>(&ptp_status.ui64GMID)[4]),
(reinterpret_cast<uint8_t*>(&ptp_status.ui64GMID)[5]),
(reinterpret_cast<uint8_t*>(&ptp_status.ui64GMID)[6]),
(reinterpret_cast<uint8_t*>(&ptp_status.ui64GMID)[7]));
pui64GMID[0], pui64GMID[1], pui64GMID[2], pui64GMID[3],
pui64GMID[4], pui64GMID[5], pui64GMID[6], pui64GMID[7]);
// update PTP clock status
std::unique_lock ptp_lock(ptp_mutex_);
// update status

21
daemon/session_manager.hpp
View File

@ -30,19 +30,20 @@
#include "driver_manager.hpp"
#include "igmp.hpp"
#include "sap.hpp"
#include "mdns_server.hpp"
struct StreamSource {
uint8_t id{0};
bool enabled{false};
std::string name;
std::string io;
std::vector<uint8_t> map;
uint32_t max_samples_per_packet{0};
std::string codec;
uint8_t ttl{0};
uint8_t payload_type{0};
uint8_t dscp{0};
bool refclk_ptp_traceable{false};
std::vector<uint8_t> map;
};
struct StreamSink {
@ -106,6 +107,10 @@ class SessionManager {
// session manager interface
bool init() {
if (!running_) {
/* init mDNS server */
if (config_->get_mdns_enabled() && !mdns_.init()) {
return false;
}
running_ = true;
res_ = std::async(std::launch::async, &SessionManager::worker, this);
}
@ -122,6 +127,9 @@ class SessionManager {
for (auto sink : get_sinks()) {
remove_sink(sink.id);
}
if (config_->get_mdns_enabled()) {
mdns_.terminate();
}
return ret;
}
return true;
@ -132,12 +140,14 @@ class SessionManager {
std::list<StreamSource> get_sources() const;
std::error_code get_source_sdp(uint32_t id, std::string& sdp) const;
std::error_code remove_source(uint32_t id);
uint8_t get_source_id(const std::string& name) const;
std::error_code add_sink(const StreamSink& sink);
std::error_code get_sink(uint8_t id, StreamSink& sink) const;
std::list<StreamSink> get_sinks() const;
std::error_code get_sink_status(uint32_t id, SinkStreamStatus& status) const;
std::error_code remove_sink(uint32_t id);
uint8_t get_sink_id(const std::string& name) const;
std::error_code set_ptp_config(const PTPConfig& config);
void get_ptp_config(PTPConfig& config) const;
@ -152,6 +162,12 @@ class SessionManager {
constexpr static const char ptp_primary_mcast_addr[] = "224.0.1.129";
constexpr static const char ptp_pdelay_mcast_addr[] = "224.0.1.107";
void on_add_source(const StreamSource& source, const StreamInfo& info);
void on_remove_source(const StreamInfo& info);
void on_add_sink(const StreamSink& sink, const StreamInfo& info);
void on_remove_sink(const StreamInfo& info);
std::string get_removed_source_sdp_(uint32_t id, uint32_t src_addr) const;
std::string get_source_sdp_(uint32_t id, const StreamInfo& info) const;
StreamSource get_source_(uint8_t id, const StreamInfo& info) const;
@ -174,10 +190,12 @@ class SessionManager {
/* current sources */
std::map<uint8_t /* id */, StreamInfo> sources_;
std::map<std::string, uint8_t /* id */> source_names_;
mutable std::shared_mutex sources_mutex_;
/* current sinks */
std::map<uint8_t /* id */, StreamInfo> sinks_;
std::map<std::string, uint8_t /* id */> sink_names_;
mutable std::shared_mutex sinks_mutex_;
/* current announced sources */
@ -192,6 +210,7 @@ class SessionManager {
PTPStatus ptp_status_;
mutable std::shared_mutex ptp_mutex_;
MDNSServer mdns_{config_};
SAP sap_{config_->get_sap_mcast_addr()};
IGMP igmp_;
};

4
daemon/tests/CMakeLists.txt
View File

@ -8,3 +8,7 @@ include_directories(aes67-daemon ${CPP_HTTPLIB} ${Boost_INCLUDE_DIR})
add_executable(daemon-test daemon_test.cpp)
target_link_libraries(daemon-test ${Boost_LIBRARIES})
add_test(daemon-test daemon-test)
if(WITH_AVAHI)
MESSAGE(STATUS "WITH_AVAHI")
add_definitions(-D_USE_AVAHI_)
endif()

6
daemon/tests/daemon.conf
View File

@ -1,5 +1,6 @@
{
"http_port": 9999,
"rtsp_port": 9997,
"http_base_dir": ".",
"log_severity": 5,
"playout_delay": 0,
@ -16,7 +17,8 @@
"syslog_server": "255.255.255.254:1234",
"status_file": "",
"interface_name": "lo",
"mdns_enabled": false,
"mdns_enabled": true,
"mac_addr": "00:00:00:00:00:00",
"ip_addr": "127.0.0.1"
"ip_addr": "127.0.0.1",
"node_id": "AES67 daemon d9aca383"
}

201
daemon/tests/daemon_test.cpp
View File

@ -25,6 +25,7 @@
#include <boost/process.hpp>
#include <boost/property_tree/json_parser.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/algorithm/string/replace.hpp>
#include <set>
@ -42,6 +43,7 @@ constexpr static const char g_sap_address[] = "224.2.127.254";
constexpr static uint16_t g_sap_port = 9875;
constexpr static uint16_t g_udp_size = 1024;
constexpr static uint16_t g_sap_header_len = 24;
constexpr static uint16_t g_stream_id_max = 64;
using namespace boost::process;
using namespace boost::asio::ip;
@ -152,6 +154,8 @@ struct Client {
"refclk_ptp_traceable": false
}
)";
boost::replace_first(json, "ALSA", "ALSA " + std::to_string(id));
std::string url = std::string("/api/source/") + std::to_string(id);
auto res = cli_.Put(url.c_str(), json, "application/json");
BOOST_REQUIRE_MESSAGE(res != nullptr, "server returned response");
@ -214,6 +218,7 @@ struct Client {
}
)";
boost::replace_first(json, "ALSA", "ALSA " + std::to_string(id));
std::string url = std::string("/api/sink/") + std::to_string(id);
auto res = cli_.Put(url.c_str(), json, "application/json");
BOOST_REQUIRE_MESSAGE(res != nullptr, "server returned response");
@ -223,7 +228,6 @@ struct Client {
bool add_sink_url(int id) {
std::string json1 = R"(
{
"name": "ALSA",
"io": "Audio Device",
"use_sdp": false,
"sdp": "",
@ -233,6 +237,7 @@ struct Client {
)";
std::string json = json1 +
std::string("\"name\": \"ALSA " + std::to_string(id) + "\",\n") +
std::string("\"source\": \"http://") +
g_daemon_address + ":" + std::to_string(g_daemon_port) +
std::string("/api/source/sdp/") + std::to_string(id) + "\"\n}";
@ -270,7 +275,7 @@ struct Client {
void sap_wait_all_deletions() {
char data[g_udp_size];
std::set<uint8_t> ids;
while (ids.size() < 64) {
while (ids.size() < g_stream_id_max) {
auto len = socket_.receive(boost::asio::buffer(data, g_udp_size));
if (len <= g_sap_header_len) {
continue;
@ -280,7 +285,7 @@ struct Client {
//o=- 56 0 IN IP4 127.0.0.1
ids.insert(std::atoi(sap_sdp_.c_str() + 3));
BOOST_TEST_MESSAGE("waiting deletion for " +
std::to_string(64 - ids.size()) + " sources");
std::to_string(g_stream_id_max - ids.size()) + " sources");
}
}
}
@ -302,8 +307,15 @@ struct Client {
return true;
}
std::pair<bool, std::string> get_remote_sources() {
std::string url = std::string("/api/browse/sources");
std::pair<bool, std::string> get_remote_sap_sources() {
std::string url = std::string("/api/browse/sources/sap");
auto res = cli_.Get(url.c_str());
BOOST_REQUIRE_MESSAGE(res != nullptr, "server returned response");
return std::make_pair(res->status == 200, res->body);
}
std::pair<bool, std::string> get_remote_mdns_sources() {
std::string url = std::string("/api/browse/sources/mdns");
auto res = cli_.Get(url.c_str());
BOOST_REQUIRE_MESSAGE(res != nullptr, "server returned response");
return std::make_pair(res->status == 200, res->body);
@ -413,13 +425,15 @@ BOOST_AUTO_TEST_CASE(set_ptp_config) {
BOOST_AUTO_TEST_CASE(add_invalid_source) {
Client cli;
BOOST_REQUIRE_MESSAGE(!cli.add_source(64), "not added source 64");
BOOST_REQUIRE_MESSAGE(!cli.add_source(g_stream_id_max),
"not added source " + std::to_string(g_stream_id_max));
BOOST_REQUIRE_MESSAGE(!cli.add_source(-1), "not added source -1");
}
BOOST_AUTO_TEST_CASE(remove_invalid_source) {
Client cli;
BOOST_REQUIRE_MESSAGE(!cli.remove_source(64), "not removed source 64");
BOOST_REQUIRE_MESSAGE(!cli.remove_source(g_stream_id_max),
"not removed source " + std::to_string(g_stream_id_max));
BOOST_REQUIRE_MESSAGE(!cli.remove_source(-1), "not removed source -1");
}
@ -459,30 +473,58 @@ BOOST_AUTO_TEST_CASE(source_check_sap) {
cli.sap_wait_deletion(0, sdp.second, 3);
}
BOOST_AUTO_TEST_CASE(source_check_browser) {
BOOST_AUTO_TEST_CASE(source_check_sap_browser) {
Client cli;
BOOST_REQUIRE_MESSAGE(cli.add_source(0), "added source 0");
auto sdp = cli.get_source_sdp(0);
BOOST_REQUIRE_MESSAGE(sdp.first, "got source sdp 0");
cli.sap_wait_announcement(0, sdp.second);
auto json = cli.get_remote_sources();
BOOST_REQUIRE_MESSAGE(json.first, "got remote sources");
auto json = cli.get_remote_sap_sources();
BOOST_REQUIRE_MESSAGE(json.first, "got remote sap sources");
boost::property_tree::ptree pt;
std::stringstream ss(json.second);
boost::property_tree::read_json(ss, pt);
BOOST_FOREACH (auto const& v, pt.get_child("remote_sources")) {
BOOST_REQUIRE_MESSAGE(v.second.get<std::string>("sdp") == sdp.second,
"returned source " + v.second.get<std::string>("id"));
"returned sap source " + v.second.get<std::string>("id"));
}
BOOST_REQUIRE_MESSAGE(cli.remove_source(0), "removed source 0");
cli.sap_wait_deletion(0, sdp.second, 3);
json = cli.get_remote_sources();
BOOST_REQUIRE_MESSAGE(json.first, "got remote sources");
json = cli.get_remote_sap_sources();
BOOST_REQUIRE_MESSAGE(json.first, "got remote sap sources");
std::stringstream ss1(json.second);
boost::property_tree::read_json(ss1, pt);
BOOST_REQUIRE_MESSAGE(pt.get_child("remote_sources").size() == 0, "no remote sources");
BOOST_REQUIRE_MESSAGE(pt.get_child("remote_sources").size() == 0,
"no remote sap sources");
}
#ifdef _USE_AVAHI_
BOOST_AUTO_TEST_CASE(source_check_mdns_browser) {
Client cli;
BOOST_REQUIRE_MESSAGE(cli.add_source(0), "added source 0");
auto sdp = cli.get_source_sdp(0);
BOOST_REQUIRE_MESSAGE(sdp.first, "got source sdp 0");
cli.sap_wait_announcement(0, sdp.second);
auto json = cli.get_remote_mdns_sources();
BOOST_REQUIRE_MESSAGE(json.first, "got remote mdns sources");
boost::property_tree::ptree pt;
std::stringstream ss(json.second);
boost::property_tree::read_json(ss, pt);
BOOST_FOREACH (auto const& v, pt.get_child("remote_sources")) {
BOOST_REQUIRE_MESSAGE(v.second.get<std::string>("sdp") == sdp.second,
"returned mdns source " + v.second.get<std::string>("id"));
}
BOOST_REQUIRE_MESSAGE(cli.remove_source(0), "removed source 0");
cli.sap_wait_deletion(0, sdp.second, 3);
json = cli.get_remote_mdns_sources();
BOOST_REQUIRE_MESSAGE(json.first, "got remote mdns sources");
std::stringstream ss1(json.second);
boost::property_tree::read_json(ss1, pt);
BOOST_REQUIRE_MESSAGE(pt.get_child("remote_sources").size() == 0,
"no remote mdns sources");
}
#endif
BOOST_AUTO_TEST_CASE(sink_check_status) {
Client cli;
BOOST_REQUIRE_MESSAGE(cli.add_sink_sdp(0), "added sink 0");
@ -500,7 +542,7 @@ BOOST_AUTO_TEST_CASE(sink_check_status) {
BOOST_AUTO_TEST_CASE(add_remove_all_sources) {
Client cli;
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.add_source(id),
std::string("added source ") + std::to_string(id));
}
@ -515,7 +557,7 @@ BOOST_AUTO_TEST_CASE(add_remove_all_sources) {
"returned source " + std::to_string(id));
++id;
}
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.remove_source(id),
std::string("removed source ") + std::to_string(id));
}
@ -523,7 +565,7 @@ BOOST_AUTO_TEST_CASE(add_remove_all_sources) {
BOOST_AUTO_TEST_CASE(add_remove_all_sinks) {
Client cli;
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.add_sink_sdp(id),
std::string("added sink ") + std::to_string(id));
}
@ -538,7 +580,7 @@ BOOST_AUTO_TEST_CASE(add_remove_all_sinks) {
"returned sink " + std::to_string(id));
++id;
}
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.remove_sink(id),
std::string("removed sink ") + std::to_string(id));
}
@ -546,11 +588,11 @@ BOOST_AUTO_TEST_CASE(add_remove_all_sinks) {
BOOST_AUTO_TEST_CASE(add_remove_check_all) {
Client cli;
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.add_source(id),
std::string("added source ") + std::to_string(id));
}
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.add_sink_sdp(id),
std::string("added sink ") + std::to_string(id));
}
@ -571,11 +613,11 @@ BOOST_AUTO_TEST_CASE(add_remove_check_all) {
"returned sink " + std::to_string(id));
++id;
}
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.remove_source(id),
std::string("removed source ") + std::to_string(id));
}
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.remove_sink(id),
std::string("removed sink ") + std::to_string(id));
}
@ -583,19 +625,19 @@ BOOST_AUTO_TEST_CASE(add_remove_check_all) {
BOOST_AUTO_TEST_CASE(add_remove_update_check_all) {
Client cli;
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.add_source(id),
std::string("added source ") + std::to_string(id));
}
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.add_source(id),
std::string("updated source ") + std::to_string(id));
}
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.add_sink_sdp(id),
std::string("added sink ") + std::to_string(id));
}
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.add_sink_url(id),
std::string("updated sink ") + std::to_string(id));
}
@ -616,11 +658,11 @@ BOOST_AUTO_TEST_CASE(add_remove_update_check_all) {
"returned sink " + std::to_string(id));
++id;
}
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.remove_source(id),
std::string("removed source ") + std::to_string(id));
}
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.remove_sink(id),
std::string("removed sink ") + std::to_string(id));
}
@ -628,26 +670,32 @@ BOOST_AUTO_TEST_CASE(add_remove_update_check_all) {
BOOST_AUTO_TEST_CASE(add_remove_check_sap_browser_all) {
Client cli;
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.add_source(id),
std::string("added source ") + std::to_string(id));
}
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
auto sdp = cli.get_source_sdp(id);
BOOST_REQUIRE_MESSAGE(sdp.first, std::string("got source sdp ") + std::to_string(id));
cli.sap_wait_announcement(id, sdp.second);
}
auto json = cli.get_remote_sources();
BOOST_REQUIRE_MESSAGE(json.first, "got remote sources");
boost::property_tree::ptree pt;
std::stringstream ss(json.second);
boost::property_tree::read_json(ss, pt);
BOOST_REQUIRE_MESSAGE(pt.get_child("remote_sources").size() == 64, "found 64 remote sources");
for (int id = 0; id < 64; id++) {
int retry = 10;
do {
std::this_thread::sleep_for(std::chrono::seconds(1));
auto json = cli.get_remote_sap_sources();
BOOST_REQUIRE_MESSAGE(json.first, "got remote sap sources");
std::stringstream ss(json.second);
boost::property_tree::read_json(ss, pt);
BOOST_TEST_MESSAGE(std::to_string(pt.get_child("remote_sources").size()));
} while (pt.get_child("remote_sources").size() != g_stream_id_max && retry--);
BOOST_REQUIRE_MESSAGE(pt.get_child("remote_sources").size() == g_stream_id_max,
"found " + std::to_string(pt.get_child("remote_sources").size()) + " remote sap sources");
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.add_sink_sdp(id),
std::string("added sink ") + std::to_string(id));
}
json = cli.get_streams();
auto json = cli.get_streams();
BOOST_REQUIRE_MESSAGE(json.first, "got streams");
std::stringstream ss1(json.second);
boost::property_tree::read_json(ss1, pt);
@ -663,18 +711,83 @@ BOOST_AUTO_TEST_CASE(add_remove_check_sap_browser_all) {
"returned sink " + std::to_string(id));
++id;
}
for (int id = 0; id < 64; id++) {
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.remove_source(id),
std::string("removed source ") + std::to_string(id));
}
cli.sap_wait_all_deletions();
json = cli.get_remote_sources();
BOOST_REQUIRE_MESSAGE(json.first, "got remote sources");
std::stringstream ss2(json.second);
boost::property_tree::read_json(ss2, pt);
BOOST_REQUIRE_MESSAGE(pt.get_child("remote_sources").size() == 0, "no remote sources");
for (int id = 0; id < 64; id++) {
retry = 10;
do {
std::this_thread::sleep_for(std::chrono::seconds(1));
auto json = cli.get_remote_sap_sources();
BOOST_REQUIRE_MESSAGE(json.first, "got remote sap sources");
std::stringstream ss2(json.second);
boost::property_tree::read_json(ss2, pt);
BOOST_TEST_MESSAGE(std::to_string(pt.get_child("remote_sources").size()));
} while (pt.get_child("remote_sources").size() > 0 && retry--);
BOOST_REQUIRE_MESSAGE(pt.get_child("remote_sources").size() == 0, "no remote sap sources");
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.remove_sink(id),
std::string("removed sink ") + std::to_string(id));
}
}
#ifdef _USE_AVAHI_
BOOST_AUTO_TEST_CASE(add_remove_check_mdns_browser_all) {
Client cli;
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.add_source(id),
std::string("added source ") + std::to_string(id));
}
boost::property_tree::ptree pt;
int retry = 10;
do {
std::this_thread::sleep_for(std::chrono::seconds(1));
auto json = cli.get_remote_mdns_sources();
BOOST_REQUIRE_MESSAGE(json.first, "got remote mdns sources");
std::stringstream ss(json.second);
boost::property_tree::read_json(ss, pt);
BOOST_TEST_MESSAGE(std::to_string(pt.get_child("remote_sources").size()));
} while (pt.get_child("remote_sources").size() != g_stream_id_max && retry--);
BOOST_REQUIRE_MESSAGE(pt.get_child("remote_sources").size() == g_stream_id_max,
"found " + std::to_string(pt.get_child("remote_sources").size()) + " remote mdns sources");
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.add_sink_sdp(id),
std::string("added sink ") + std::to_string(id));
}
auto json = cli.get_streams();
BOOST_REQUIRE_MESSAGE(json.first, "got streams");
std::stringstream ss1(json.second);
boost::property_tree::read_json(ss1, pt);
uint8_t id = 0;
BOOST_FOREACH (auto const& v, pt.get_child("sources")) {
BOOST_REQUIRE_MESSAGE(v.second.get<uint8_t>("id") == id,
"returned source " + std::to_string(id));
++id;
}
id = 0;
BOOST_FOREACH (auto const& v, pt.get_child("sinks")) {
BOOST_REQUIRE_MESSAGE(v.second.get<uint8_t>("id") == id,
"returned sink " + std::to_string(id));
++id;
}
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.remove_source(id),
std::string("removed source ") + std::to_string(id));
}
retry = 10;
do {
std::this_thread::sleep_for(std::chrono::seconds(1));
auto json = cli.get_remote_mdns_sources();
BOOST_REQUIRE_MESSAGE(json.first, "got remote mdns sources");
std::stringstream ss2(json.second);
boost::property_tree::read_json(ss2, pt);
BOOST_TEST_MESSAGE(std::to_string(pt.get_child("remote_sources").size()));
} while (pt.get_child("remote_sources").size() > 0 && retry--);
BOOST_REQUIRE_MESSAGE(pt.get_child("remote_sources").size() == 0, "no remote mdns sources");
for (int id = 0; id < g_stream_id_max; id++) {
BOOST_REQUIRE_MESSAGE(cli.remove_sink(id),
std::string("removed sink ") + std::to_string(id));
}
}
#endif

60
daemon/utils.cpp
View File

@ -32,3 +32,63 @@ uint16_t crc16(const uint8_t* p, size_t len) {
}
return crc;
}
std::tuple<bool /* res */,
std::string /* protocol */,
std::string /* host */,
std::string /* port */,
std::string /* path */>
parse_url(const std::string& _url) {
std::string url = httplib::detail::decode_url(_url);
size_t protocol_sep_pos = url.find_first_of("://");
if (protocol_sep_pos == std::string::npos) {
/* no protocol, invalid URL */
return std::make_tuple(false, "", "", "", "");
}
std::string port, host, path("/");
std::string protocol = url.substr(0, protocol_sep_pos);
std::string url_new = url.substr(protocol_sep_pos + 3);
size_t path_sep_pos = url_new.find_first_of("/");
size_t port_sep_pos = url_new.find_first_of(":");
if (port_sep_pos != std::string::npos) {
/* port specified */
if (path_sep_pos != std::string::npos) {
/* path specified */
port = url_new.substr(port_sep_pos + 1, path_sep_pos - port_sep_pos - 1);
path = url_new.substr(path_sep_pos);
} else {
/* path not specified */
port = url_new.substr(port_sep_pos + 1);
}
host = url_new.substr(0, port_sep_pos);
} else if (path_sep_pos != std::string::npos) {
/* port not specified, path specified */
host = url_new.substr(0, path_sep_pos);
path = url_new.substr(path_sep_pos);
} else {
/* port and path not specified */
host = url_new;
}
return std::make_tuple(host.length() > 0, protocol, host, port, path);
}
std::string get_host_id() {
char hostname[64];
gethostname(hostname, sizeof hostname);
std::stringstream ss;
ss << std::hex << (uint32_t)gethostid();
return ss.str();
}
std::string get_host_name() {
char hostname[64];
gethostname(hostname, sizeof hostname);
return hostname;
}
std::string get_node_id() {
std::stringstream ss;
ss << "AES67 daemon " << get_host_id();
return ss.str();
}

16
daemon/utils.hpp
View File

@ -20,10 +20,22 @@
#ifndef _UTILS_HPP_
#define _UTILS_HPP_
#include <stdint.h>
#include <iostream>
#include <cstddef>
#include <httplib.h>
uint16_t crc16(const uint8_t* p, size_t len);
std::tuple<bool /* res */,
std::string /* protocol */,
std::string /* host */,
std::string /* port */,
std::string /* path */>
parse_url(const std::string& _url);
std::string get_host_id();
std::string get_host_name();
std::string get_node_id();
#endif

1
demo/daemon.conf
View File

@ -1,5 +1,6 @@
{
"http_port": 8080,
"rtsp_port": 8854,
"http_base_dir": "./webui/build",
"log_severity": 3,
"playout_delay": 0,

2
run_demo.sh
View File

@ -40,7 +40,7 @@ fi
trap cleanup EXIT
#configure system parms
sudo sysctl -w net/ipv4/igmp_max_memberships=64
sudo sysctl -w net/ipv4/igmp_max_memberships=66
if [ -x /usr/bin/pulseaudio ]; then
#stop pulseaudio, this seems to open/close ALSA continuosly

14
webui/src/Config.js
View File

@ -31,6 +31,8 @@ class Config extends Component {
super(props);
this.state = {
httpPort: '',
rtspPort: '',
rtspPortErr: false,
logSeverity: '',
playoutDelay: '',
playoutDelayErr: false,
@ -70,6 +72,7 @@ class Config extends Component {
data => this.setState(
{
httpPort: data.http_port,
rtspPort: data.rtsp_port,
logSeverity: data.log_severity,
playoutDelay: data.playout_delay,
ticFrameSizeAt1fs: data.tic_frame_size_at_1fs,
@ -88,6 +91,7 @@ class Config extends Component {
interfaceName: data.interface_name,
macAddr: data.mac_addr,
ipAddr: data.ip_addr,
nodeId: data.node_id,
isConfigLoading: false
}))
.catch(err => this.setState({isConfigLoading: false}));
@ -100,6 +104,7 @@ class Config extends Component {
!this.state.rtpMcastBaseErr &&
!this.state.sapMcastAddrErr &&
!this.state.rtpPortErr &&
!this.state.rtspPortErr &&
!this.state.sapIntervalErr &&
!this.state.syslogServerErr &&
!this.state.isConfigLoading;
@ -113,6 +118,7 @@ class Config extends Component {
this.state.syslogServer,
this.state.rtpMcastBase,
this.state.rtpPort,
this.state.rtspPort,
this.state.playoutDelay,
this.state.ticFrameSizeAt1fs,
this.state.sampleRate,
@ -154,10 +160,18 @@ class Config extends Component {
<br/>
{this.state.isConfigLoading ? <Loader/> : <h3>Network Config</h3>}
<table><tbody>
<tr>
<th align="left"> <label>Node ID</label> </th>
<th align="left"> <input value={this.state.nodeId} disabled/> </th>
</tr>
<tr>
<th align="left"> <label>HTTP port</label> </th>
<th align="left"> <input type='number' min='1024' max='65536' className='input-number' value={this.state.httpPort} disabled/> </th>
</tr>
<tr>
<th align="left"> <label>RTSP port</label> </th>
<th align="left"> <input type='number' min='1024' max='65536' className='input-number' value={this.state.rtspPort} onChange={e => this.setState({rtspPort: e.target.value, rtspPortErr: !e.currentTarget.checkValidity()})} required/> </th>
</tr>
<tr>
<th align="left"> <label>RTP base address</label> </th>
<th align="left"> <input type="text" minLength="7" maxLength="15" size="15" pattern="^2(?:2[4-9]|3\d)(?:\.(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]\d?|0)){3}$" value={this.state.rtpMcastBase} onChange={e => this.setState({rtpMcastBase: e.target.value, rtpMcastBaseErr: !e.currentTarget.checkValidity()})} required/> </th>

5
webui/src/Services.js
View File

@ -31,7 +31,7 @@ const source = '/source';
const sdp = '/sdp';
const sink = '/sink';
const status = '/status';
const browseSources = '/browse/sources';
const browseSources = '/browse/sources/all';
const defaultParams = {
credentials: 'same-origin',
@ -76,7 +76,7 @@ export default class RestAPI {
});
}
static setConfig(log_severity, syslog_proto, syslog_server, rtp_mcast_base, rtp_port, playout_delay, tic_frame_size_at_1fs, sample_rate, max_tic_frame_size, sap_mcast_addr, sap_interval, mdns_enabled) {
static setConfig(log_severity, syslog_proto, syslog_server, rtp_mcast_base, rtp_port, rtsp_port, playout_delay, tic_frame_size_at_1fs, sample_rate, max_tic_frame_size, sap_mcast_addr, sap_interval, mdns_enabled) {
return this.doFetch(config, {
body: JSON.stringify({
log_severity: parseInt(log_severity, 10),
@ -84,6 +84,7 @@ export default class RestAPI {
syslog_server: syslog_server,
rtp_mcast_base: rtp_mcast_base,
rtp_port: parseInt(rtp_port, 10),
rtsp_port: parseInt(rtsp_port, 10),
playout_delay: parseInt(playout_delay, 10),
tic_frame_size_at_1fs: parseInt(tic_frame_size_at_1fs, 10),
sample_rate: parseInt(sample_rate, 10),