/*

* SRT - Secure, Reliable, Transport

* Copyright (c) 2018 Haivision Systems Inc.

*

* This Source Code Form is subject to the terms of the Mozilla Public

* License, v. 2.0. If a copy of the MPL was not distributed with this

* file, You can obtain one at http://mozilla.org/MPL/2.0/.

*

*/

// Medium concretizations

// Just for formality. This file should be used

#include <iostream>

#include <fstream>

#include <sstream>

#include <string>

#include <stdexcept>

#include <iterator>

#include <map>

#include <chrono>

#include <thread>

#include <srt.h>

#if !defined(_WIN32)

#include <sys/ioctl.h>

#endif

// SRT protected includes

#define REQUIRE_CXX11 1

#include "srt_attr_defs.h"

#include "netinet_any.h"

#include "common.h"

#include "api.h"

#include "logging.h"

#include "ofmt.h"

#include "utilities.h"

#include "apputil.hpp"

#include "socketoptions.hpp"

#include "uriparser.hpp"

#include "testmedia.hpp"

#include "hvu_compat.h"

#include "verbose.hpp"

using namespace std;

using namespace srt;

using namespace hvu;

using srt::KmStateStr;

using srt::SockStatusStr;

using srt::MemberStatusStr;

srt::sync::atomic<bool> transmit_throw_on_interrupt {false};

srt::sync::atomic<bool> transmit_int_state {false};

int transmit_bw_report = 0;

unsigned transmit_stats_report = 0;

size_t transmit_chunk_size = SRT_LIVE_DEF_PLSIZE;

bool transmit_printformat_json = false;

srt_listen_callback_fn* transmit_accept_hook_fn = nullptr;

void* transmit_accept_hook_op = nullptr;

bool transmit_use_sourcetime = false;

int transmit_retry_connect = 0;

bool transmit_retry_always = false;

struct CloseReasonMap

{

map<SRT_CLOSE_REASON, string> at;

CloseReasonMap()

{

at[SRT_CLS_UNKNOWN] = "Unset";

at[SRT_CLS_INTERNAL] = "Closed by internal reasons during connection attempt";

at[SRT_CLS_PEER] = "Received SHUTDOWN message from the peer";

at[SRT_CLS_RESOURCE] = "Problem with resource allocation";

at[SRT_CLS_ROGUE] = "Received wrong data in the packet";

at[SRT_CLS_OVERFLOW] = "Emergency close due to receiver buffer overflow";

at[SRT_CLS_IPE] = "Internal program error";

at[SRT_CLS_API] = "The application called srt_close()";

at[SRT_CLS_FALLBACK] = "The peer doesn't support close reason feature";

at[SRT_CLS_LATE] = "Accepted-socket late-rejection or in-handshake rollback";

at[SRT_CLS_CLEANUP] = "All sockets are being closed due to srt_cleanup() call";

at[SRT_CLS_DEADLSN] = "This was an accepted socket off a dead listener";

at[SRT_CLS_PEERIDLE] = "Peer didn't send any packet for a time of SRTO_PEERIDLETIMEO";

at[SRT_CLS_UNSTABLE] = "Requested to be broken as unstable in Backup group";

}

string operator[](SRT_CLOSE_REASON rval)

{

int reason = int(rval);

if (reason >= SRT_CLSC_USER)

{

string extra;

if (reason == SRT_CLSC_USER)

extra = " - Application exit due to interrupted transmission";

if (reason == SRT_CLSC_USER + 1)

extra = " - Error during configuration, transmission not started";

return fmtcat("User-defined reason #", reason - SRT_CLSC_USER, extra);

}

auto p = at.find(rval);

if (p == at.end())

return "UNDEFINED";

return p->second;

}

} g_close_reason;

// Do not unblock. Copy this to an app that uses applog and set appropriate name.

// app_logger_config must be also declared in the same place (it must be

// initialized before this object can be initialized).

//hvu::logging::Logger applog("app", app_logger_config, true, "srt-test");

std::shared_ptr<SrtStatsWriter> transmit_stats_writer;

string DirectionName(SRT_EPOLL_T direction)

{

string dir_name;

if (direction & ~SRT_EPOLL_ERR)

{

if (direction & SRT_EPOLL_IN)

{

dir_name = "source";

}

if (direction & SRT_EPOLL_OUT)

{

if (!dir_name.empty())

dir_name = "relay";

else

dir_name = "target";

}

if (direction & SRT_EPOLL_ERR)

{

dir_name += "+error";

}

}

else

{

// stupid name for a case of IPE

dir_name = "stone";

}

return dir_name;

}

static string RejectReasonStr(int id)

{

if (id < SRT_REJC_PREDEFINED)

return srt_rejectreason_str(id);

if (id < SRT_REJC_USERDEFINED)

return srt_rejectreasonx_str(id);

ostringstream sout;

sout << "User-defined reason code " << id;

return sout.str();

}

template<class FileBase> inline

bytevector FileRead(FileBase& ifile, size_t chunk, const string& filename)

{

bytevector data(chunk);

ifile.read(data.data(), chunk);

size_t nread = ifile.gcount();

if (nread < data.size())

data.resize(nread);

if (data.empty())

throw Source::ReadEOF(filename);

return data;

}

class FileSource: public virtual Source

{

ifstream ifile;

string filename_copy;

public:

FileSource(const string& path): ifile(path, ios::in | ios::binary), filename_copy(path)

{

if (!ifile)

throw std::runtime_error(path + ": Can't open file for reading");

}

MediaPacket Read(size_t chunk) override { return FileRead(ifile, chunk, filename_copy); }

bool IsOpen() override { return bool(ifile); }

bool End() override { return ifile.eof(); }

//~FileSource() { ifile.close(); }

};

class FileTarget: public virtual Target

{

ofstream ofile;

public:

FileTarget(const string& path): ofile(path, ios::out | ios::trunc | ios::binary) {}

void Write(const MediaPacket& data) override

{

ofile.write(data.payload.data(), data.payload.size());

#ifdef PLEASE_LOG

applog.Debug() << "FileTarget::Write: " << data.payload.size() << " written to a file";

#endif

}

bool IsOpen() override { return !!ofile; }

bool Broken() override { return !ofile.good(); }

//~FileTarget() { ofile.close(); }

void Close() override

{

#ifdef PLEASE_LOG

applog.Debug() << "FileTarget::Close";

#endif

ofile.close();

}

};

// Can't base this class on FileSource and FileTarget classes because they use two

// separate fields, which makes it unable to reliably define IsOpen(). This would

// require to use 'fstream' type field in some kind of FileCommon first. Not worth

// a shot.

class FileRelay: public Relay

{

fstream iofile;

string filename_copy;

public:

FileRelay(const string& path):

iofile(path, ios::in | ios::out | ios::binary), filename_copy(path)

{

if (!iofile)

throw std::runtime_error(path + ": Can't open file for reading");

}

MediaPacket Read(size_t chunk) override { return FileRead(iofile, chunk, filename_copy); }

void Write(const MediaPacket& data) override

{

iofile.write(data.payload.data(), data.payload.size());

}

bool IsOpen() override { return !!iofile; }

bool End() override { return iofile.eof(); }

bool Broken() override { return !iofile.good(); }

void Close() override { iofile.close(); }

};

template <class Iface> struct File;

template <> struct File<Source> { typedef FileSource type; };

template <> struct File<Target> { typedef FileTarget type; };

template <> struct File<Relay> { typedef FileRelay type; };

template <class Iface>

Iface* CreateFile(const string& name) { return new typename File<Iface>::type (name); }

void SrtCommon::InitParameters(string host, string path, map<string,string> par)

{

// Application-specific options: mode, blocking, timeout, adapter

if ( Verbose::on && !par.empty())

{

Verb() << "SRT parameters specified:\n";

for (map<string,string>::iterator i = par.begin(); i != par.end(); ++i)

{

Verb() << "\t" << i->first << " = '" << i->second << "'\n";

}

}

if (path != "")

{

// Special case handling of an unusual specification.

if (path.substr(0, 2) != "//")

{

Error("Path specification not supported for SRT (use // in front for special cases)");

}

path = path.substr(2);

if (path == "group")

{

// Group specified, check type.

m_group_type = par["type"];

if (m_group_type == "")

{

Error("With //group, the group 'type' must be specified.");

}

vector<string> parts;

Split(m_group_type, '/', back_inserter(parts));

if (parts.size() == 0 || parts.size() > 2)

{

Error("Invalid specification for 'type' parameter");

}

if (parts.size() == 2)

{

m_group_type = parts[0];

m_group_config = parts[1];

}

vector<string> nodes;

Split(par["nodes"], ',', back_inserter(nodes));

if (nodes.empty())

{

Error("With //group, 'nodes' must specify comma-separated host:port specs.");

}

int token = 1;

// Check if correctly specified

for (string& hostport: nodes)

{

if (hostport == "")

continue;

// The attribute string, as it was embedded in another URI,

// must have had replaced the & character with another ?, so

// now all ? character, except the first one, must be now

// restored so that UriParser interprets them correctly.

size_t atq = hostport.find('?');

if (atq != string::npos)

{

while (atq+1 < hostport.size())

{

size_t next = hostport.find('?', atq+1);

if (next == string::npos)

break;

hostport[next] = '&';

atq = next;

}

}

UriParser check(hostport, UriParser::EXPECT_HOST);

if (check.host() == "" || check.port() == "")

{

Error("With //group, 'nodes' must specify comma-separated host:port specs.");

}

if (check.portno() <= 1024)

{

Error("With //group, every node in 'nodes' must have port >1024");

}

Connection cc(check.host(), check.portno());

if (check.parameters().count("weight"))

{

cc.weight = stoi(check.queryValue("weight"));

}

if (check.parameters().count("source"))

{

UriParser sourcehp(check.queryValue("source"), UriParser::EXPECT_HOST);

cc.source = CreateAddr(sourcehp.host(), sourcehp.portno());

}

// Check if there's a key with 'srto.' prefix.

UriParser::query_it start = check.parameters().lower_bound("srto.");

SRT_SOCKOPT_CONFIG* config = nullptr;

bool all_clear = true;

vector<string> fails;

map<string, string> options;

if (start != check.parameters().end())

{

for (; start != check.parameters().end(); ++start)

{

auto& y = *start;

if (y.first.substr(0, 5) != "srto.")

break;

options[y.first.substr(5)] = y.second;

}

}

if (!options.empty())

{

config = srt_create_config();

for (auto o: srt_options)

{

if (!options.count(o.name))

continue;

string value = options.at(o.name);

bool ok = o.apply<SocketOption::SRT>(config, value);

if ( !ok )

{

fails.push_back(o.name);

all_clear = false;

}

}

if (!all_clear)

{

srt_delete_config(config);

Error("With //group, failed to set options: " + Printable(fails));

}

cc.options = config;

}

cc.token = token++;

m_group_nodes.push_back(std::move(cc));

}

par.erase("type");

par.erase("nodes");

// For a group-connect specification, it's

// always the caller mode.

// XXX change it here if maybe rendezvous is also

// possible in future.

par["mode"] = "caller";

}

}

if (par.count("bind"))

{

string bindspec = par.at("bind");

UriParser u (bindspec, UriParser::EXPECT_HOST);

if ( u.scheme() != ""

|| u.path() != ""

|| !u.parameters().empty()

|| u.portno() == 0)

{

Error("Invalid syntax in 'bind' option");

}

if (u.host() != "")

par["adapter"] = u.host();

par["port"] = u.port();

par.erase("bind");

}

string adapter;

if (par.count("adapter"))

{

adapter = par.at("adapter");

}

m_mode = "default";

if (par.count("mode"))

{

m_mode = par.at("mode");

}

size_t max_payload_size = 0;

// Try to interpret host and adapter first

sockaddr_any host_sa, adapter_sa;

if (host != "")

{

host_sa = CreateAddr(host);

if (host_sa.family() == AF_UNSPEC)

Error("Failed to interpret 'host' spec: " + host);

if (host_sa.family() == AF_INET)

max_payload_size = SRT_MAX_PLSIZE_AF_INET;

}

if (adapter != "")

{

adapter_sa = CreateAddr(adapter);

if (adapter_sa.family() == AF_UNSPEC)

Error("Failed to interpret 'adapter' spec: " + adapter);

if (host_sa.family() != AF_UNSPEC && host_sa.family() != adapter_sa.family())

{

Error("Both host and adapter specified and they use different IP versions");

}

if (max_payload_size == 0 && host_sa.family() == AF_INET)

max_payload_size = SRT_MAX_PLSIZE_AF_INET;

}

if (!max_payload_size)

max_payload_size = SRT_MAX_PLSIZE_AF_INET6;

SocketOption::Mode mode = SrtInterpretMode(m_mode, host, adapter);

if (mode == SocketOption::FAILURE)

{

Error("Invalid mode");

}

if (!m_group_nodes.empty() && mode != SocketOption::CALLER)

{

Error("Group node specification is only available in caller mode");

}

// Fix the mode name after successful interpretation

m_mode = SocketOption::mode_names[mode];

par.erase("mode");

if (par.count("blocking"))

{

m_blocking_mode = !false_names.count(par.at("blocking"));

par.erase("blocking");

}

if (par.count("timeout"))

{

m_timeout = stoi(par.at("timeout"), 0, 0);

par.erase("timeout");

}

if (par.count("adapter"))

{

m_adapter = adapter;

par.erase("adapter");

}

else if (m_mode == "listener")

{

// For listener mode, adapter is taken from host,

// if 'adapter' parameter is not given

m_adapter = host;

}

if (par.count("port"))

{

m_outgoing_port = stoi(par.at("port"), 0, 0);

par.erase("port");

}

// Assigning group configuration from a special "groupconfig" attribute.

// This is the only way how you can set up this configuration at the listener side.

if (par.count("groupconfig"))

{

m_group_config = par.at("groupconfig");

par.erase("groupconfig");

}

// -----------------

// Fixing socket options, if needed (keys remain in the map)

// -----------------

if (par.count("tsbpd") && false_names.count(par.at("tsbpd")))

{

m_tsbpdmode = false;

}

// That's kinda clumsy, but it must rely on the defaults.

// Default mode is live, so check if the file mode was enforced

if ((par.count("transtype") == 0 || par["transtype"] != "file")

&& transmit_chunk_size > SRT_LIVE_DEF_PLSIZE)

{

if (transmit_chunk_size > max_payload_size)

throw std::runtime_error(fmtcat("Chunk size in live mode exceeds ", max_payload_size, " bytes; this is not supported"));

par["payloadsize"] = fmts(transmit_chunk_size);

}

else

{

// set it so without making sure that it was set to "file".

// worst case it will be rejected in settings

m_transtype = SRTT_FILE;

}

// Fix Minversion, if specified as string

if (par.count("minversion"))

{

string v = par["minversion"];

if (v.find('.') != string::npos)

{

int version = srt::SrtParseVersion(v.c_str());

if (version == 0)

{

throw std::runtime_error(fmtcat("Value for 'minversion' doesn't specify a valid version: ", v));

}

par["minversion"] = fmts(version);

Verb("\tFIXED: minversion = 0x", fmt(version, fmtc().hex().fillzero().width(8)));

}

}

// Assign the others here.

m_options = par;

m_options["mode"] = m_mode;

}

void SrtCommon::PrepareListener(string host, int port, int backlog)

{

m_bindsock = srt_create_socket();

if (m_bindsock == SRT_INVALID_SOCK)

Error("srt_create_socket");

SRTSTATUS stat = ConfigurePre(m_bindsock);

if (stat == SRT_ERROR)

Error("ConfigurePre");

if (!m_blocking_mode)

{

srt_conn_epoll = AddPoller(m_bindsock, SRT_EPOLL_IN);

}

auto sa = CreateAddr(host, port);

Verb() << "Binding a server on " << host << ":" << port << " ...";

stat = srt_bind(m_bindsock, sa.get(), sizeof sa);

if (stat == SRT_ERROR)

{

srt_close(m_bindsock);

Error("srt_bind");

}

Verb() << " listen... " << VerbNoEOL;

stat = srt_listen(m_bindsock, backlog);

if (stat == SRT_ERROR)

{

srt_close(m_bindsock);

Error("srt_listen");

}

}

void SrtCommon::StealFrom(SrtCommon& src)

{

// This is used when SrtCommon class designates a listener

// object that is doing Accept in appropriate direction class.

// The new object should get the accepted socket.

m_direction = src.m_direction;

m_blocking_mode = src.m_blocking_mode;

m_timeout = src.m_timeout;

m_tsbpdmode = src.m_tsbpdmode;

m_options = src.m_options;

m_bindsock = SRT_INVALID_SOCK; // no listener

m_sock = src.m_sock;

src.m_sock = SRT_INVALID_SOCK; // STEALING

}

void SrtCommon::AcceptNewClient()

{

sockaddr_any scl;

::transmit_throw_on_interrupt = true;

if (!m_blocking_mode)

{

Verb() << "[ASYNC] (conn=" << srt_conn_epoll << ")";

int len = 2;

SRTSOCKET ready[2];

while (srt_epoll_wait(srt_conn_epoll, ready, &len, 0, 0, 1000, 0, 0, 0, 0) == int(SRT_ERROR))

{

if (::transmit_int_state)

Error("srt_epoll_wait for srt_accept: interrupt");

if (srt_getlasterror(NULL) == SRT_ETIMEOUT)

continue;

Error("srt_epoll_wait(srt_conn_epoll)");

}

Verb() << "[EPOLL: " << len << " sockets] " << VerbNoEOL;

}

Verb() << " accept..." << VerbNoEOL;

m_sock = srt_accept(m_bindsock, (scl.get()), (&scl.len));

if (m_sock == SRT_INVALID_SOCK)

{

srt_close(m_bindsock);

m_bindsock = SRT_INVALID_SOCK;

Error("srt_accept");

}

int maxsize = srt_getmaxpayloadsize(m_sock);

if (maxsize == int(SRT_ERROR))

{

srt_close(m_bindsock);

srt_close(m_sock);

Error("srt_getmaxpayloadsize");

}

if (m_transtype == SRTT_LIVE && transmit_chunk_size > size_t(maxsize))

{

srt_close(m_bindsock);

srt_close(m_sock);

Error(fmtcat("accepted connection's payload size ", maxsize, " is too small for required ", transmit_chunk_size, " chunk size"));

}

if (int32_t(m_sock) & SRTGROUP_MASK)

{

m_listener_group = true;

if (m_group_config != "")

{

// Don't break the connection basing on this, just ignore.

Verb() << " (ignoring setting group config: '" << m_group_config << "') " << VerbNoEOL;

}

// There might be added a poller, remove it.

// We need it work different way.

if (srt_epoll != int(SRT_ERROR))

{

Verb() << "(Group: erasing epoll " << srt_epoll << ") " << VerbNoEOL;

srt_epoll_release(srt_epoll);

}

// Don't add any sockets, they will have to be added

// anew every time again.

srt_epoll = srt_epoll_create();

// Group data must have a size of at least 1

// otherwise the srt_group_data() call will fail

if (m_group_data.empty())

m_group_data.resize(1);

Verb() << " connected(group epoll " << srt_epoll <<").";

}

else

{

sockaddr_any peeraddr(AF_INET6);

string peer = "<?PEER?>";

if (SRT_ERROR != srt_getpeername(m_sock, (peeraddr.get()), (&peeraddr.len)))

{

peer = peeraddr.str();

}

sockaddr_any agentaddr(AF_INET6);

string agent = "<?AGENT?>", dev = "<dev unknown>";

if (SRT_ERROR != srt_getsockname(m_sock, (agentaddr.get()), (&agentaddr.len)))

{

agent = agentaddr.str();

char name[256];

size_t len = 255;

if (srt_getsockdevname(m_sock, name, &len) == SRT_STATUS_OK)

dev.assign(name, len);

}

Verb() << " connected [" << agent << "] <-- " << peer << " [" << dev << "]";

}

::transmit_throw_on_interrupt = false;

// ConfigurePre is done on bindsock, so any possible Pre flags

// are DERIVED by sock. ConfigurePost is done exclusively on sock.

SRTSTATUS stat = ConfigurePost(m_sock);

if (stat == SRT_ERROR)

Error("ConfigurePost");

}

static string PrintEpollEvent(int events, int et_events)

{

static pair<int, const char*> const namemap [] = {

make_pair(SRT_EPOLL_IN, "R"),

make_pair(SRT_EPOLL_OUT, "W"),

make_pair(SRT_EPOLL_ERR, "E"),

make_pair(SRT_EPOLL_UPDATE, "U")

};

ostringstream os;

int N = (int)Size(namemap);

for (int i = 0; i < N; ++i)

{

if (events & namemap[i].first)

{

os << "[";

if (et_events & namemap[i].first)

os << "^";

os << namemap[i].second << "]";

}

}

return os.str();

}

void SrtCommon::Init(string host, int port, string path, map<string,string> par, SRT_EPOLL_OPT dir)

{

m_direction = dir;

InitParameters(host, path, par);

int backlog = 1;

if (m_mode == "listener" && par.count("groupconnect")

&& true_names.count(par["groupconnect"]))

{

backlog = 10;

}

Verb("Opening SRT ", DirectionName(dir), " ",

m_mode, "(", m_blocking_mode ? "" : "non-", "blocking,",

" backlog=", backlog, ") on ", host, ":", port);

try

{

if (m_mode == "caller")

{

if (m_group_nodes.empty())

{

OpenClient(host, port);

}

else

{

OpenGroupClient(); // Source data are in the fields already.

}

}

else if (m_mode == "listener")

OpenServer(m_adapter, port, backlog);

else if (m_mode == "rendezvous")

OpenRendezvous(m_adapter, host, port);

else

{

throw std::invalid_argument("Invalid 'mode'. Use 'client' or 'server'");

}

}

catch (...)

{

// This is an in-constructor-called function, so

// when the exception is thrown, the destructor won't

// close the sockets. This intercepts the exception

// to close them.

Verb() << "Open FAILED - closing SRT sockets";

if (m_bindsock != SRT_INVALID_SOCK)

srt_close(m_bindsock);

if (m_sock != SRT_INVALID_SOCK)

srt_close_withreason(m_sock, SRT_CLSC_USER+1);

m_sock = m_bindsock = SRT_INVALID_SOCK;

throw;

}

int pbkeylen = 0;

SRT_KM_STATE kmstate, snd_kmstate, rcv_kmstate;

int len = sizeof (int);

srt_getsockflag(m_sock, SRTO_PBKEYLEN, &pbkeylen, &len);

srt_getsockflag(m_sock, SRTO_KMSTATE, &kmstate, &len);

srt_getsockflag(m_sock, SRTO_SNDKMSTATE, &snd_kmstate, &len);

srt_getsockflag(m_sock, SRTO_RCVKMSTATE, &rcv_kmstate, &len);

Verb() << "ENCRYPTION status: " << KmStateStr(kmstate)

<< " (SND:" << KmStateStr(snd_kmstate) << " RCV:" << KmStateStr(rcv_kmstate)

<< ") PBKEYLEN=" << pbkeylen;

// Display some selected options on the socket.

if (Verbose::on)

{

int64_t bandwidth = 0;

int latency = 0;

bool blocking_snd = false, blocking_rcv = false;

int dropdelay = 0;

int size_int = sizeof (int), size_int64 = sizeof (int64_t), size_bool = sizeof (bool);

char packetfilter[100] = "";

int packetfilter_size = 100;

srt_getsockflag(m_sock, SRTO_MAXBW, &bandwidth, &size_int64);

srt_getsockflag(m_sock, SRTO_RCVLATENCY, &latency, &size_int);

srt_getsockflag(m_sock, SRTO_RCVSYN, &blocking_rcv, &size_bool);

srt_getsockflag(m_sock, SRTO_SNDSYN, &blocking_snd, &size_bool);

srt_getsockflag(m_sock, SRTO_SNDDROPDELAY, &dropdelay, &size_int);

srt_getsockflag(m_sock, SRTO_PACKETFILTER, (packetfilter), (&packetfilter_size));

Verb() << "OPTIONS: maxbw=" << bandwidth << " rcvlatency=" << latency << boolalpha

<< " blocking{rcv=" << blocking_rcv << " snd=" << blocking_snd

<< "} snddropdelay=" << dropdelay << " packetfilter=" << packetfilter;

}

if (!m_blocking_mode)

{

// Don't add new epoll if already created as a part

// of group management: if (srt_epoll == SRT_ERROR)...

if (m_mode == "caller")

dir = (dir | SRT_EPOLL_UPDATE);

Verb() << "NON-BLOCKING MODE - SUB FOR " << PrintEpollEvent(dir, 0);

srt_epoll = AddPoller(m_sock, dir);

}

}

int SrtCommon::AddPoller(SRTSOCKET socket, int modes)

{

int pollid = srt_epoll_create();

if (pollid == int(SRT_ERROR))

throw std::runtime_error("Can't create epoll in nonblocking mode");

Verb() << "EPOLL: creating eid=" << pollid << " and adding @" << socket

<< " in " << DirectionName(SRT_EPOLL_OPT(modes)) << " mode";

srt_epoll_add_usock(pollid, socket, &modes);

return pollid;

}

SRTSTATUS SrtCommon::ConfigurePost(SRTSOCKET sock)

{

bool yes = m_blocking_mode;

SRTSTATUS result = SRT_STATUS_OK;

if (m_direction & SRT_EPOLL_OUT)

{

Verb() << "Setting SND blocking mode: " << boolalpha << yes << " timeout=" << m_timeout;

result = srt_setsockopt(sock, 0, SRTO_SNDSYN, &yes, sizeof yes);

if (result == SRT_ERROR)

{

#ifdef PLEASE_LOG

extern hvu::logging::Logger applog;

applog.Error() << "ERROR SETTING OPTION: SRTO_SNDSYN";

#endif

return result;

}

if (m_timeout)

result = srt_setsockopt(sock, 0, SRTO_SNDTIMEO, &m_timeout, sizeof m_timeout);

if (result == SRT_ERROR)

{

#ifdef PLEASE_LOG

extern hvu::logging::Logger applog;

applog.Error() << "ERROR SETTING OPTION: SRTO_SNDTIMEO";

#endif

return result;

}

}

if (m_direction & SRT_EPOLL_IN)

{

Verb() << "Setting RCV blocking mode: " << boolalpha << yes << " timeout=" << m_timeout;

result = srt_setsockopt(sock, 0, SRTO_RCVSYN, &yes, sizeof yes);

if (result == SRT_ERROR)

return result;

if (m_timeout)

result = srt_setsockopt(sock, 0, SRTO_RCVTIMEO, &m_timeout, sizeof m_timeout);

else

{

int timeout = 1000;

result = srt_setsockopt(sock, 0, SRTO_RCVTIMEO, &timeout, sizeof timeout);

}

if (result == SRT_ERROR)

return result;

}

// host is only checked for emptiness and depending on that the connection mode is selected.

// Here we are not exactly interested with that information.

vector<string> failures;

SrtConfigurePost(sock, m_options, &failures);

if (!failures.empty())

{

if (Verbose::on)

{

Verb() << "WARNING: failed to set options: ";

copy(failures.begin(), failures.end(), ostream_iterator<string>(*Verbose::cverb, ", "));

Verb();

}

}

return SRT_STATUS_OK;

}

SRTSTATUS SrtCommon::ConfigurePre(SRTSOCKET sock)

{

SRTSTATUS result = SRT_STATUS_OK;

int no = 0;

if (!m_tsbpdmode)

{

result = srt_setsockopt(sock, 0, SRTO_TSBPDMODE, &no, sizeof no);

if (result == SRT_ERROR)

return result;

}

// Let's pretend async mode is set this way.

// This is for asynchronous connect.

int maybe = m_blocking_mode;

result = srt_setsockopt(sock, 0, SRTO_RCVSYN, &maybe, sizeof maybe);

if (result == SRT_ERROR)

return result;

// host is only checked for emptiness and depending on that the connection mode is selected.

// Here we are not exactly interested with that information.

vector<string> failures;

// NOTE: here host = "", so the 'connmode' will be returned as LISTENER always,

// but it doesn't matter here. We don't use 'connmode' for anything else than

// checking for failures.

SocketOption::Mode conmode = SrtConfigurePre(sock, "", m_options, &failures);

if (conmode == SocketOption::FAILURE)

{

if (Verbose::on )

{

Verb() << "WARNING: failed to set options: ";

copy(failures.begin(), failures.end(), ostream_iterator<string>(*Verbose::cverb, ", "));

Verb();

}

return SRT_ERROR;

}

return SRT_STATUS_OK;

}

void SrtCommon::SetupAdapter(const string& host, int port)

{

Verb() << "Binding the caller socket to " << host << ":" << port << " ...";

auto lsa = CreateAddr(host, port);