Description
Shows how to create a device by integrating a 3rd party library.
Prebuilt Binaries
Source Code
./examples/demo_adtfplugins/communication_with_foreign_application/
Common
asio_helpers.h
#pragma once
#include <asio.hpp>
#include <string>
#include <chrono>
asio::ip::address resolve_address(const std::string& strHostOrAddress, bool prefere_v6 = false);
template<typename SocketType>
bool wait_for_data(SocketType& oSocket, std::chrono::milliseconds tmTimeout)
{
asio::error_code ec;
if (asio::detail::socket_ops::poll_error(oSocket.native_handle(), 0, 0, ec) != 0)
{
return true;
}
return asio::detail::socket_ops::poll_read(oSocket.native_handle(), 0, static_cast<int>(tmTimeout.count()), ec) != 0;
}
asio_helpers.cpp
#include "asio_helpers.h"
asio::ip::address resolve_address(const std::string& strHostOrAddress, bool prefere_v6)
{
if (strHostOrAddress.empty())
{
if (prefere_v6)
{
return asio::ip::address_v6();
}
else
{
return asio::ip::address_v4();
}
}
try
{
return asio::ip::make_address(strHostOrAddress);
}
catch(...)
{
asio::io_service io_service;
asio::ip::udp::resolver resolver(io_service);
asio::ip::udp::resolver::iterator current = resolver.resolve(strHostOrAddress, "");
auto result_address = current->endpoint().address();
for (;current != asio::ip::udp::resolver::iterator(); ++current)
{
if (current->endpoint().address().is_v6() == prefere_v6)
{
result_address = current->endpoint().address();
break;
}
}
return result_address;
}
}
sink_to_non_adtf.h
#pragma once
{
public:
cSinkToNonADTFSystem();
protected:
virtual void Send(const void* pData, size_t nDataSize) = 0;
private:
};
A_UTILS_NS::cResult tResult
For backwards compatibility and to bring latest version into scope.
Definition result.h:736
Memory buffer class to encapsulate and manage raw contiguously memory.
Definition memorybuffer.h:23
Property Variable template for the given T. A Property Variable will store a copy of a property value...
Definition configuration.h:808
virtual tResult ProcessInput(base::flash::tNanoSeconds tmTrigger, streaming::flash::ISampleReader *pReader)
virtual tResult AcceptType(streaming::flash::ISampleReader *pReader, const ucom::ant::iobject_ptr< const streaming::ant::IStreamType > &pType)
Definition sample_streaming_sink.h:49
Definition samplestreamer_intf.h:188
Definition codec_factory.h:31
ant::iobject_ptr< T > iobject_ptr
Alias always bringing the latest version of ant::iobject_ptr into scope.
Definition object_ptr_intf.h:437
sink_to_non_adtf.cpp
#include "sink_to_non_adtf.h"
using namespace adtf::util;
cSinkToNonADTFSystem::cSinkToNonADTFSystem()
{
m_bSerializeViaMediaDescription.SetDescription("If activated, incoming samples will be serialized as defined "
"by the media description of the stream type before sending it via the socket.");
RegisterPropertyVariable("serialize_via_media_description", m_bSerializeViaMediaDescription);
m_pInputReader = CreateInputPin("input");
SetDescription("input", "Incoming data to transmit");
}
{
if (!m_bSerializeViaMediaDescription)
{
return cSampleStreamingSink::AcceptType(pReader, pType);
}
if (!m_oCodecFactory.isValid())
{
RETURN_ERROR_DESC(ERR_INVALID_ARG,
"Serialization requires a stream type that has a media description set.");
}
}
{
if (!m_bSerializeViaMediaDescription)
{
Send(oRawData.GetDataPtr(), oRawData.GetDataSize());
}
else
{
RETURN_IF_FAILED(pSample->Lock(pBuffer));
auto oDecoder = m_oCodecFactory.makeDecoderFor(pBuffer->GetPtr(), pBuffer->GetSize());
RETURN_IF_FAILED(TO_ADTF_RESULT(oDecoder.isValid()));
pBuffer = nullptr;
Send(m_oSerializationBuffer.getPtr(), m_oSerializationBuffer.getSize());
}
}
#define RETURN_ERROR_DESC(_code,...)
Same as RETURN_ERROR(_error) using a printf like parameter list for detailed error description.
Definition result.h:44
#define RETURN_NOERROR
Return status ERR_NOERROR, which requires the calling function's return type to be tResult.
Definition result.h:29
Definition sample_data.h:429
virtual T * Get() const =0
Get raw pointer to shared object.
Definition lockedobject_intf.h:273
Namespace for the ADTF Base SDK.
Definition adtf_base_type_traits.h:13
Namespace for the ADTF Streaming SDK.
Definition bindingproxyinport.h:14
Namespace for the ADTF uCOM SDK.
Definition adtf_system.h:324
a_util::result::Result transformToBuffer(const codec::Decoder &decoder, a_util::memory::MemoryBuffer &buffer, bool zero=false)
TCP
TCP Wrapper
tcp_socket_wrapper.h
#pragma once
#include <asio.hpp>
#include "tcp_socket_wrapper_intf.h"
#include <mutex>
#ifdef WIN32
#undef GetObject
#endif
{
public:
cTCPSocketWrapper();
void EnableTcpNoDelay() override;
tResult Connect(
const char* strRemoteHost, uint16_t nRemotePort,
bool bReconnect)
override;
tResult Send(
const void* pData,
size_t nDataSize)
override;
tResult Read(
void* pDestination,
size_t nBytes,
size_t* pBytesRead)
override;
void Close() override;
private:
std::string m_strRemoteHost;
uint16_t m_nRemotePort = 0;
bool m_bReconnect = false;
std::mutex m_oConnectMutex;
asio::io_context m_oIoContext;
asio::ip::tcp::socket m_oTcpSocket;
bool m_bTcpNoDelay = false;
};
tcp_socket_wrapper.cpp
#include "tcp_socket_wrapper.h"
#include <asio_helpers.h>
constexpr std::chrono::milliseconds SOCKET_READ_TIMEOUT{50};
cTCPSocketWrapper::cTCPSocketWrapper():
m_oTcpSocket(m_oIoContext)
{
}
tResult cTCPSocketWrapper::Connect(
const char* strRemoteHost, uint16_t nRemotePort,
bool bReconnect)
{
m_strRemoteHost = strRemoteHost;
m_nRemotePort = nRemotePort;
m_bReconnect = bReconnect;
return Connect();
}
tResult cTCPSocketWrapper::Connect()
{
std::lock_guard<std::mutex> oGuard(m_oConnectMutex);
m_oTcpSocket.close();
try
{
m_oTcpSocket.connect(asio::ip::tcp::endpoint(resolve_address(m_strRemoteHost), m_nRemotePort));
}
catch (...)
{
if (m_bReconnect)
{
}
}
if (m_bTcpNoDelay)
{
}
}
void cTCPSocketWrapper::EnableTcpNoDelay()
{
m_bTcpNoDelay = true;
}
bool is_connection_error(const asio::system_error& oError)
{
return oError.code() == asio::error::not_connected ||
oError.code() == asio::error::connection_aborted ||
oError.code() == asio::error::connection_refused ||
oError.code() == asio::error::connection_reset ||
oError.code() == asio::error::eof ||
oError.code() == asio::error::broken_pipe;
}
tResult cTCPSocketWrapper::Send(
const void* pData,
size_t nDataSize)
{
auto pPosition = static_cast<const uint8_t*>(pData);
size_t nDataLeft = nDataSize;
while (nDataLeft)
{
size_t nDataSent = 0;
try
{
nDataSent = m_oTcpSocket.send(asio::buffer(pPosition, nDataSize));
}
catch (const asio::system_error& oError)
{
if (is_connection_error(oError) &&
m_bReconnect)
{
{
pPosition = static_cast<const uint8_t*>(pData);
nDataLeft = nDataSize;
continue;
}
}
}
pPosition += nDataSent;
nDataLeft -= static_cast<size_t>(nDataSent);
}
}
tResult cTCPSocketWrapper::Read(
void* pDestination,
size_t nBytes,
size_t* pBytesRead)
{
if (!wait_for_data<>(m_oTcpSocket, SOCKET_READ_TIMEOUT))
{
}
try
{
*pBytesRead = m_oTcpSocket.receive(asio::buffer(pDestination, nBytes));
}
catch (const asio::system_error& oError)
{
if (is_connection_error(oError) &&
m_bReconnect)
{
Connect();
}
}
}
void cTCPSocketWrapper::Close()
{
m_oTcpSocket.close();
}
{
}
#define RETURN_CURRENT_EXCEPTION()
returns the current exception as a tResult, use it in a catch block.
Definition result.h:111
#define RETURN_IF_THROWS(s)
if the expression throws an exception, returns a tResult containing the exception information.
Definition result.h:123
#define RETURN_ERROR(code)
Return specific error code, which requires the calling function's return type to be tResult.
Definition result.h:42
#define IS_OK(s)
Check if result is OK.
Definition result.h:17
Definition object_ptr.h:384
object_ptr< Implementation > make_object_ptr(Args &&... args)
Alias always bringing the latest version of ant::make_object_ptr() into scope.
Definition object_ptr_utilities.h:129
TCP Receiver
tcp_source_from_non_adtf.h
#pragma once
#include "tcp_socket_wrapper_intf.h"
#ifndef ADTF_EXAMPLES_CID
#define ADTF_EXAMPLES_CID ".local.cid"
#endif
{
public:
"demo_foreign_application_tcp_receiver.streaming_source" ADTF_EXAMPLES_CID,
"TCP Receiver From Non ADTF Application");
cTcpSourceFromNonADTFSystem();
~cTcpSourceFromNonADTFSystem() override;
private:
void ReadThread();
size_t ReadFromSocket(void* pBuffer, size_t nBufferSize, bool bFillBuffer);
private:
std::vector<uint8_t> m_oReadBuffer;
bool m_bReadDirectlyIntoSample = true;
};
#define REQUIRE_INTERFACE(_interface)
Macro usable with ADTF_CLASS_DEPENDENCIES() to require mandatory interfaces.
Definition class_dependencies.h:36
#define ADTF_CLASS_DEPENDENCIES(...)
Add interface ids (string literals,.
Definition class_dependencies.h:61
#define ADTF_CLASS_ID_NAME(_class, _strcid, _strclabel)
Definition class_id.h:33
Definition sample_streaming_source.h:50
tResult StopStreaming() override
tResult StartStreaming() override
Definition runner_fallback.h:25
Definition kernel_intf.h:390
Definition reference_clock_intf.h:782
virtual tResult Construct()
Definition streaming_source.h:69
Definition samplestreamer_intf.h:234
tcp_source_from_non_adtf.cpp
#include "tcp_source_from_non_adtf.h"
#include <asio_helpers.h>
constexpr const tTimeStamp g_tmSocketTimeOut = 50000;
constexpr const size_t g_nDefaultReadBufferSize = 0xFFFF;
using namespace adtf::util;
cTcpSourceFromNonADTFSystem::cTcpSourceFromNonADTFSystem()
{
m_strRemoteHost.SetDescription("The hostname or ip address that the source should connect to.");
RegisterPropertyVariable("remote_host", m_strRemoteHost);
m_nRemotePort.SetDescription("The port on the remote host that the source should connect to.");
m_nRemotePort.SetValidRange(1, 65535);
RegisterPropertyVariable("remote_port", m_nRemotePort);
m_nFixedPacketSize.SetDescription("If non-zero, the source will wait to read this amount of bytes before putting them into a Sample.");
RegisterPropertyVariable("fixed_packet_size", m_nFixedPacketSize);
m_strDDLStructName.SetDescription(
"If set, the stream type will contain the media description of the given struct, retrieved from the "
"description file specified within property 'ddl_description_file' if a valid filepath is set. If no file is "
"set, it will be retrieved from the (Deprecated) Media Description Service.");
RegisterPropertyVariable("ddl_struct_name", m_strDDLStructName);
m_strDescriptionFile.SetDescription(
"A description file containing the DDL description. If empty, the media description specified within the "
"(deprecated) ADTF Media Description Service will be used.");
RegisterPropertyVariable("ddl_description_file", m_strDescriptionFile);
m_bDeserializeViaMediaDescription.SetDescription("If activated, incoming data will be deserialized as defined "
"by the media description of the struct selected via ddl_struct_name before writing it into output samples.");
RegisterPropertyVariable("deserialize_via_media_description", m_bDeserializeViaMediaDescription);
m_bMarkSamplesAsDeserialized.SetDescription("In case that 'deserialize_via_media_description' is disabled, this will force the output stream type to have the deserialized flag set.");
RegisterPropertyVariable("mark_samples_as_deserialized", m_bMarkSamplesAsDeserialized);
m_bEnableAutomaticReconnection.SetDescription("If enabled, connection loss is not treated as an error, but the source tries to reconnect.");
RegisterPropertyVariable("enable_automatic_reconnection", m_bEnableAutomaticReconnection);
m_pOutputWriter = CreateOutputPin("output");
SetDescription("output", "Provides the sample data received from a Non-ADTF instance over TCP");
m_pSocket = ITCPSocketPrivate::Create();
SetDescription("socket", "Interface server to provide a shared socket for communication");
SetDescription("receive_data",
"Connect a Thread Runner that will provide the context for receiving all data. "
"If this is not connected, the source will create a thread on its own.");
SetDescription("Use this streaming source to receive sample data from a Non-ADTF Application using the TCP protocol");
SetHelpLink("$(ADTF_DIR)/doc/html/page_tcp_receiver_from_non_adtf_application.html");
}
cTcpSourceFromNonADTFSystem::~cTcpSourceFromNonADTFSystem() = default;
tResult cTcpSourceFromNonADTFSystem::Construct()
{
RETURN_IF_FAILED(cSampleStreamingSource::Construct());
if (m_strDDLStructName->IsNotEmpty())
{
{
const auto oDataRep = m_bDeserializeViaMediaDescription || m_bMarkSamplesAsDeserialized ?
if (m_strDescriptionFile->IsNotEmpty())
{
oDataDefinition, oDataRep);
}
else
{
ADTF3_IGNORE_DEPRECATION_WARNING_BEGIN
oDataRep);
ADTF3_IGNORE_DEPRECATION_WARNING_END
}
}();
m_pOutputWriter->ChangeType(pType);
if (m_bDeserializeViaMediaDescription)
{
m_oCodecFactory =
}
if (*m_nFixedPacketSize == 0)
{
if (m_bDeserializeViaMediaDescription)
{
}
else
{
}
}
}
else
{
if (m_bDeserializeViaMediaDescription)
{
RETURN_ERROR_DESC(ERR_INVALID_ARG,
"Deserialization requires a ddl struct name to be set.");
}
}
}
tResult cTcpSourceFromNonADTFSystem::Init()
{
RETURN_IF_FAILED(cSampleStreamingSource::Init());
RETURN_IF_FAILED(_runtime->GetObject(m_pClock));
if ((*m_strRemoteHost).IsEmpty())
{
}
if (*m_nRemotePort == 0)
{
}
m_bReadDirectlyIntoSample = m_nFixedPacketSize > 0 && !m_bDeserializeViaMediaDescription;
if (!m_bReadDirectlyIntoSample)
{
if (m_nFixedPacketSize > 0)
{
m_oReadBuffer.resize(m_nFixedPacketSize);
}
else
{
m_oReadBuffer.resize(g_nDefaultReadBufferSize);
}
}
}
tResult cTcpSourceFromNonADTFSystem::StartStreaming()
{
RETURN_IF_FAILED(cSampleStreamingSource::StartStreaming());
RETURN_IF_FAILED_DESC(m_pSocket->Connect(*m_strRemoteHost, m_nRemotePort, m_bEnableAutomaticReconnection),
"Unable to connect to %s:%" PRIu16, m_strRemoteHost->GetPtr(), *m_nRemotePort);
RETURN_IF_FAILED(m_oReadThread.Activate());
}
tResult cTcpSourceFromNonADTFSystem::StopStreaming()
{
m_oReadThread.Deactivate();
m_pSocket->Close();
return cSampleStreamingSource::StopStreaming();
}
void cTcpSourceFromNonADTFSystem::ReadThread()
{
try
{
if (m_bReadDirectlyIntoSample)
{
pSample = ReadIntoSample();
}
else
{
if (m_bDeserializeViaMediaDescription)
{
pSample = ReadAndDeserialize();
}
else
{
pSample = ReadIntoBuffer();
}
}
if (pSample)
{
m_pOutputWriter->Write(pSample);
}
}
catch (...)
{
}
}
{
const auto nBytesRead = ReadFromSocket(pBuffer->GetPtr(), pBuffer->GetSize(), true);
if (nBytesRead == 0)
{
return nullptr;
}
pSample->SetTime(m_pClock->GetStreamTimeNs());
return pSample;
}
{
const auto nBytesRead = ReadFromSocket(m_oReadBuffer.data(), m_oReadBuffer.size(), false);
if (nBytesRead == 0)
{
return nullptr;
}
return pSample;
}
{
const auto nBytesRead = ReadFromSocket(m_oReadBuffer.data(), m_oReadBuffer.size(), true);
if (nBytesRead == 0)
{
return nullptr;
}
return pSample;
}
size_t cTcpSourceFromNonADTFSystem::ReadFromSocket(void* pBuffer, size_t nBufferSize, bool bFillBuffer)
{
auto pBufferWritePosition = static_cast<uint8_t*>(pBuffer);
size_t nOverallBytesRead = 0;
do
{
size_t nLastBytesRead = 0;
auto oResult = m_pSocket->Read(pBufferWritePosition, static_cast<int>(nBufferSize), &nLastBytesRead);
{
if (oResult == ERR_TIMEOUT)
{
if (nOverallBytesRead == 0)
{
return 0;
}
continue;
}
if (oResult == ERR_RETRY)
{
std::this_thread::sleep_for(std::chrono::milliseconds(100));
return 0;
}
throw oResult;
}
nOverallBytesRead += nLastBytesRead;
pBufferWritePosition += nLastBytesRead;
nBufferSize -= nLastBytesRead;
}
while (bFillBuffer && nBufferSize > 0);
return nOverallBytesRead;
}
#define IS_FAILED(s)
Check if result is failed.
Definition result.h:20
#define CURRENT_EXCEPTION()
converts the current exception to a tResult
Definition result.h:108
#define THROW_IF_FAILED(s)
throws if the expression returns a failed tResult
Definition result.h:88
#define RETURN_IF_FAILED_DESC(s,...)
Definition result.h:168
Template class implementation for the IRawMemory interface.
Definition rawmemory_base.h:216
Definition lockedobject_intf.h:207
static dd::DataDefinition fromXMLFile(const std::string &xml_filepath, bool strict=false)
Read a file containing a data definiton in XML.
tResult alloc_sample(ucom::ant::iobject_ptr< ucom::ant::IObject > &pSample)
base::flash::tNanoSeconds get_sample_time(const ucom::ant::iobject_ptr< const ant::ISample > &pSample)
Namespace for the ADTF System SDK.
Definition adtf_service.h:14
object_ptr< T > ucom_object_ptr_cast(object_ptr< T > oCasted)
Alias always bringing the latest version of ant::ucom_object_ptr_cast() into scope.
Definition object_ptr_utilities.h:104
a_util::result::Result transform(const DECODER &decoder, ENCODER &encoder, const TransformOption transform_option)
Definition serialization.h:57
@ Deserialized
alias names for legacy reasons
Definition data_representation.h:29
@ Serialized
alias names for legacy reasons
Definition data_representation.h:28
TCP Sender
tcp_sink_to_non_adtf.h
#pragma once
#include <sink_to_non_adtf.h>
#include "tcp_socket_wrapper_intf.h"
#ifndef ADTF_EXAMPLES_CID
#define ADTF_EXAMPLES_CID ".local.cid"
#endif
class cTcpSinkToNonADTFSystem final : public cSinkToNonADTFSystem
{
public:
"demo_foreign_application_tcp_sender.streaming_sink" ADTF_EXAMPLES_CID,
"TCP Sender To Non ADTF Application");
cTcpSinkToNonADTFSystem();
~cTcpSinkToNonADTFSystem() override;
protected:
void Send(const void* pData, size_t nDataSize) override;
private:
ITCPSocket& GetSocket();
};
Definition graph_object.h:66
tcp_sink_to_non_adtf.cpp
#include "tcp_sink_to_non_adtf.h"
using namespace adtf::util;
cTcpSinkToNonADTFSystem::cTcpSinkToNonADTFSystem()
{
m_strRemoteHost.SetDescription(
"This is only taken into account when the sink is not connected to a source via the socket interface binding.\n"
"If set to 'localhost' the IPv4 localhost address '127.0.0.1' is used.\n"
"IPv4 Example: '127.0.0.1'\n"
"IPv6 Example: '::1'\n");
RegisterPropertyVariable("remote_host", m_strRemoteHost);
m_nRemotePort.SetDescription(
"This is only taken into account when the sink is not connected to a source via the socket interface binding.");
m_nRemotePort.SetValidRange(1, 65535);
RegisterPropertyVariable("remote_port", m_nRemotePort);
m_bNoTCPDelay.SetDescription("If activated, the TCP_NODELAY option will be set for the socket.");
RegisterPropertyVariable("no_tcp_delay", m_bNoTCPDelay);
m_bEnableAutomaticReconnection.SetDescription(
"If enabled, connection loss is not treated as an error, but the source tries to reconnect."
" This is only taken into account when the sink is not connected to a source via the socket interface "
"binding.");
RegisterPropertyVariable("enable_automatic_reconnection", m_bEnableAutomaticReconnection);
m_oSocketClient = CreateInterfaceClient<ITCPSocket>("socket");
SetDescription("socket", "Interface client to use a shared socket for communication");
SetDescription("Use this streaming sink to transmit sample data to a Non-ADTF Application using the TCP protocol");
SetHelpLink("$(ADTF_DIR)/doc/html/page_tcp_sender_to_non_adtf_application.html");
}
cTcpSinkToNonADTFSystem::~cTcpSinkToNonADTFSystem() = default;
tResult cTcpSinkToNonADTFSystem::Init()
{
RETURN_IF_FAILED(cSampleStreamingSink::Init());
if (!m_oSocketClient.IsValid())
{
if (m_strRemoteHost->IsEmpty())
{
"Remote host can not be empty if socket interface is not connected to a sink.");
}
if (*m_nRemotePort == 0)
{
"Remote host can not be 0 if socket interface is not connected to a sink.");
}
m_pFallbackSocket = ITCPSocketPrivate::Create();
}
}
tResult cTcpSinkToNonADTFSystem::StartStreaming()
{
RETURN_IF_FAILED(cSampleStreamingSink::StartStreaming());
if (m_pFallbackSocket)
{
m_pFallbackSocket->Connect(*m_strRemoteHost, m_nRemotePort, m_bEnableAutomaticReconnection),
"Unable to connect to %s:%" PRIu16, m_strRemoteHost->GetPtr(), *m_nRemotePort);
}
if (m_bNoTCPDelay)
{
GetSocket().EnableTcpNoDelay();
}
}
void cTcpSinkToNonADTFSystem::Send(const void* pData, size_t nDataSize)
{
const auto oResult = GetSocket().Send(pData, nDataSize);
if (!m_bEnableAutomaticReconnection || oResult != ERR_RETRY)
{
}
}
ITCPSocket& cTcpSinkToNonADTFSystem::GetSocket()
{
return m_pFallbackSocket ? *m_pFallbackSocket : m_oSocketClient.Get();
}
TCP plugin
tcp_plugin.cpp
#include "tcp_source_from_non_adtf.h"
#include "tcp_sink_to_non_adtf.h"
cTcpSourceFromNonADTFSystem,
cTcpSinkToNonADTFSystem);
#define ADTF_PLUGIN(__plugin_identifier,...)
Definition adtf_plugin.h:29
UDP
UDP Wrapper
udp_socket_wrapper.h
#pragma once
#include <asio.hpp>
#include "udp_socket_wrapper_intf.h"
#include <optional>
#include <array>
#include <adtf_base.h>
#include <adtf_filtersdk.h>
#include <adtf_systemsdk.h>
#ifdef _WIN32
#undef GetObject
#endif
#if defined(__GLIBC__) && (__GLIBC__ > 2 || (__GLIBC__ == 2 && __GLIBC_MINOR__ >= 12))
#define OS_HAS_RECVMMSG 1
#endif
{
public:
cUdpSocketWrapper(
std::function<void(const void*, size_t)> fnOnReceive = {});
~cUdpSocketWrapper() override;
tResult Open(
const char* strInterface,
uint16_t nLocalPort,
const char* strRemoteAddress,
size_t nReceiveBufferSize,
size_t nSocketReceiveBufferSize) override;
tResult SetRemote(
const char* strRemoteAddress, uint16_t nRemotePort)
override;
tResult Send(
const void* pData,
size_t nDataSize)
override;
private:
#if defined(ASIO_HAS_IOCP) || defined(ASIO_HAS_IO_URING_AS_DEFAULT) || defined(OS_HAS_RECVMMSG)
static constexpr size_t m_nQueueDepth = 64;
#else
static constexpr size_t m_nQueueDepth = 1;
#endif
struct tPacketHandler
{
tPacketHandler(size_t nHandlerId, size_t nBufferSize): nHandlerId(nHandlerId), oBuffer(nBufferSize)
{
}
size_t nHandlerId;
std::vector<std::byte> oBuffer;
asio::mutable_registered_buffer oRegisteredBuffer;
asio::ip::udp::endpoint oEndpoint;
};
#if defined(OS_HAS_RECVMMSG)
void ScheduleReadMany() noexcept;
void ReadMany(const std::error_code& nError
) noexcept;
#endif
void ScheduleReadOne(const std::shared_ptr<tPacketHandler>& oBuffer) noexcept;
void OnReceiveOne(const std::error_code& nError,
std::size_t nBytesTransferred,
const std::shared_ptr<tPacketHandler>& oBuffer) noexcept;
asio::io_context m_oIoContext;
asio::ip::udp::socket m_oUdpSocket;
std::optional<asio::ip::udp::endpoint> m_oUdpRemoteEndpoint;
std::optional<asio::ip::udp::endpoint> m_oLastReceivedUdpRemoteEndpoint;
std::unique_ptr<asio::buffer_registration<std::vector<asio::mutable_buffer>>> m_pAsioBufferRegistration;
const std::function<void(const void*, size_t)> m_fnOnReceive;
std::array<std::shared_ptr<tPacketHandler>, m_nQueueDepth> m_oHandlers;
bool m_bRunning = false;
std::mutex m_oRunningMutex;
std::condition_variable m_oRunningCv;
};
udp_socket_wrapper.cpp
#include "udp_socket_wrapper.h"
#include <asio_helpers.h>
#include <iostream>
#include <cinttypes>
cUdpSocketWrapper::cUdpSocketWrapper(std::function<void(const void*, size_t)> fnOnReceive):
m_oUdpSocket(m_oIoContext), m_fnOnReceive(std::move(fnOnReceive))
{
m_oIoContext.stop();
}
cUdpSocketWrapper::~cUdpSocketWrapper() = default;
tResult cUdpSocketWrapper::Open(
const char* strInterface,
uint16_t nLocalPort,
const char* strRemoteAddress,
size_t nReceiveBufferSize,
size_t nSocketReceiveBufferSize)
{
if (m_oUdpSocket.is_open())
{
RETURN_ERROR_DESC(ERR_INVALID_STATE,
"Can't reconfigure socket - socket is already open.");
}
const auto oBindAddress = resolve_address(strInterface);
asio::ip::udp::socket::protocol_type protocol =
oBindAddress.is_v4() ? asio::ip::udp::socket::protocol_type::v4() : asio::ip::udp::socket::protocol_type::v6();
RETURN_IF_THROWS(m_oUdpSocket.set_option(asio::ip::udp::socket::reuse_address(
true)));
if (nSocketReceiveBufferSize != 0)
{
try
{
asio::ip::udp::socket::receive_buffer_size oRcvBufferSizeOptionGet;
m_oUdpSocket.get_option(oRcvBufferSizeOptionGet);
m_oUdpSocket.set_option(
asio::ip::udp::socket::receive_buffer_size(static_cast<int>(nSocketReceiveBufferSize)));
LOG_INFO(
"Sockets buffersize set from %d to %zu bytes.", oRcvBufferSizeOptionGet.value(),
nSocketReceiveBufferSize);
}
catch (const asio::error_code&)
{
}
}
const auto oMulticastAddress = resolve_address(strRemoteAddress);
if (!oBindAddress.is_unspecified() && !oMulticastAddress.is_unspecified() &&
oMulticastAddress.is_v4() != oBindAddress.is_v4())
{
ERR_INVALID_ARG,
"Both the multicast address and the bind interface address need to be of the same ip version.");
}
if (!oMulticastAddress.is_multicast())
{
"Binding IPv4 socket to %s:%" PRIu16 " failed.", oBindAddress.to_string().c_str(), nLocalPort);
}
else
{
if (oBindAddress.is_v4())
{
#ifdef _WIN32
"Binding IPv4 MC socket to %s:%" PRIu16 " failed.", oBindAddress.to_string().c_str(),
nLocalPort);
#else
"Binding IPv4 MC socket to %s:%" PRIu16 " failed.", oBindAddress.to_string().c_str(),
nLocalPort);
#endif
}
else
{
"Binding IPv6 socket to %s:%" PRIu16 " failed.", oBindAddress.to_string().c_str(),
nLocalPort);
}
asio::error_code oSetOutboundInterfaceError;
if (oBindAddress.is_v4())
{
m_oUdpSocket.set_option(asio::ip::multicast::outbound_interface(oBindAddress.to_v4()),
oSetOutboundInterfaceError);
}
else
{
m_oUdpSocket.set_option(asio::ip::multicast::outbound_interface(oBindAddress.to_v6().scope_id()),
oSetOutboundInterfaceError);
}
if (oSetOutboundInterfaceError)
{
LOG_INFO(
"Setting multicast outbound interface is not supported on this platform: %s.",
oSetOutboundInterfaceError.message().c_str());
}
if (m_fnOnReceive)
{
if (!oSetOutboundInterfaceError)
{
if (oMulticastAddress.is_v4())
{
asio::ip::multicast::join_group(oMulticastAddress.to_v4(), oBindAddress.to_v4())));
}
else
{
asio::ip::multicast::join_group(oMulticastAddress.to_v6(), oBindAddress.to_v6().scope_id())));
}
}
else
{
if (oMulticastAddress.is_v4())
{
m_oUdpSocket.set_option(asio::ip::multicast::join_group(oMulticastAddress.to_v4())));
}
else
{
m_oUdpSocket.set_option(asio::ip::multicast::join_group(oMulticastAddress.to_v6())));
}
}
}
}
if (m_fnOnReceive)
{
std::vector<asio::mutable_buffer> oOriginalBuffers;
for (size_t i = 0; i < m_nQueueDepth; ++i)
{
m_oHandlers[i] = std::make_shared<tPacketHandler>(i, nReceiveBufferSize);
oOriginalBuffers.emplace_back(m_oHandlers[i]->oBuffer.data(), m_oHandlers[i]->oBuffer.size());
}
m_pAsioBufferRegistration = std::make_unique<asio::buffer_registration<std::vector<asio::mutable_buffer>>>(
asio::register_buffers(m_oIoContext, oOriginalBuffers));
for (size_t i = 0; i < m_nQueueDepth; ++i)
{
m_oHandlers[i]->oRegisteredBuffer = (*m_pAsioBufferRegistration)[i];
}
#if defined(OS_HAS_RECVMMSG) && !defined(ASIO_HAS_IO_URING_AS_DEFAULT)
ScheduleReadMany();
#else
for (auto& oBuffer : m_oHandlers)
{
ScheduleReadOne(oBuffer);
}
#endif
}
else
{
#ifdef _WIN32
#endif
}
}
tResult cUdpSocketWrapper::SetRemote(
const char* strRemoteAddress, uint16_t nRemotePort)
{
RETURN_IF_THROWS(m_oUdpRemoteEndpoint = asio::ip::udp::endpoint(resolve_address(strRemoteAddress), nRemotePort));
}
tResult cUdpSocketWrapper::Send(
const void* pData,
size_t nDataSize)
{
asio::ip::udp::endpoint oDestination;
if (m_oUdpRemoteEndpoint)
{
oDestination = *m_oUdpRemoteEndpoint;
}
else if (m_oLastReceivedUdpRemoteEndpoint)
{
oDestination = *m_oLastReceivedUdpRemoteEndpoint;
}
else
{
RETURN_ERROR_DESC(ERR_INVALID_STATE,
"Unable to send data via UDP, remote endpoint is not yet known.");
}
size_t nDataSent = 0;
"Send to %s:%" PRIuLEAST16 " failed.", oDestination.address().to_string().c_str(),
oDestination.port());
if (static_cast<size_t>(nDataSent) != nDataSize)
{
RETURN_ERROR_DESC(ERR_INVALID_ARG,
"Unable to send data via UDP, data size is too large: %" PRIu64, nDataSize);
}
}
{
std::unique_lock oLock(m_oRunningMutex);
if (m_bRunning)
{
}
if (m_oIoContext.stopped())
{
}
m_bRunning = true;
m_oRunningCv.notify_all();
}
{
std::unique_lock oLock(m_oRunningMutex);
if (m_oRunningCv.wait_for(oLock, std::chrono::milliseconds(100), [this]() { return m_bRunning; }))
{
oLock.unlock();
m_oIoContext.run();
oLock.lock();
m_bRunning = false;
m_oRunningCv.notify_all();
}
else
{
}
}
{
{
std::unique_lock oLock(m_oRunningMutex);
m_oIoContext.stop();
m_oRunningCv.wait(oLock, [this]() -> bool { return !m_bRunning; });
}
}
#if defined(OS_HAS_RECVMMSG)
void cUdpSocketWrapper::ScheduleReadMany() noexcept
{
m_oUdpSocket.async_wait(asio::ip::udp::socket::wait_read,
std::bind(&cUdpSocketWrapper::ReadMany, this, std::placeholders::_1));
}
void cUdpSocketWrapper::ReadMany(const std::error_code& nError) noexcept
{
if (!nError)
{
do
{
std::array<sockaddr_storage, m_nQueueDepth> oAddresses = {0};
std::array<iovec, m_nQueueDepth> oBuffers = {0};
std::array<mmsghdr, m_nQueueDepth> oMessages = {0};
for (size_t i = 0; i < m_nQueueDepth; ++i)
{
oMessages[i].msg_hdr.msg_name = &oAddresses[i];
oMessages[i].msg_hdr.msg_namelen = sizeof(sockaddr_storage);
oBuffers[i].iov_base = m_oHandlers[i]->oBuffer.data();
oBuffers[i].iov_len = m_oHandlers[i]->oBuffer.size();
oMessages[i].msg_hdr.msg_iov = &oBuffers[i];
oMessages[i].msg_hdr.msg_iovlen = 1;
oMessages[i].msg_len = 0;
}
const auto nReceived =
::recvmmsg(m_oUdpSocket.native_handle(), oMessages.data(), oMessages.size(), MSG_DONTWAIT, nullptr);
if (nReceived > 0)
{
for (int i = 0; i < nReceived; ++i)
{
const auto& oMessage = oMessages[i];
if (m_fnOnReceive && oMessage.msg_hdr.msg_iovlen > 0)
{
m_fnOnReceive(oMessage.msg_hdr.msg_iov[0].iov_base, oMessage.msg_len);
}
}
const auto& oLastMessage = oMessages[nReceived - 1];
const auto pAddress = static_cast<const sockaddr_storage*>(oLastMessage.msg_hdr.msg_name);
if (pAddress)
{
if (pAddress->ss_family == AF_INET)
{
const auto pIPV4 = reinterpret_cast<const sockaddr_in*>(pAddress);
m_oLastReceivedUdpRemoteEndpoint = {asio::ip::address_v4(pIPV4->sin_addr.s_addr),
pIPV4->sin_port};
}
else if (pAddress->ss_family == AF_INET6)
{
const auto pIPV6 = reinterpret_cast<const sockaddr_in6*>(pAddress);
m_oLastReceivedUdpRemoteEndpoint = {
asio::ip::address_v6(
reinterpret_cast<const asio::ip::address_v6::bytes_type&>(pIPV6->sin6_addr.s6_addr),
pIPV6->sin6_scope_id),
pIPV6->sin6_port};
}
}
}
else
{
break;
}
} while (true);
}
if (nError != asio::error::operation_aborted)
{
ScheduleReadMany();
}
}
#endif
void cUdpSocketWrapper::ScheduleReadOne(const std::shared_ptr<tPacketHandler>& oBuffer) noexcept
{
m_oUdpSocket.async_receive_from(
oBuffer->oRegisteredBuffer, oBuffer->oEndpoint,
std::bind(&cUdpSocketWrapper::OnReceiveOne, this, std::placeholders::_1, std::placeholders::_2, oBuffer));
}
void cUdpSocketWrapper::OnReceiveOne(const std::error_code& nError,
std::size_t nBytesTransferred,
const std::shared_ptr<tPacketHandler>& oBuffer) noexcept
{
if (!nError)
{
m_oLastReceivedUdpRemoteEndpoint = oBuffer->oEndpoint;
if (m_fnOnReceive)
{
m_fnOnReceive(oBuffer->oBuffer.data(), nBytesTransferred);
}
}
if (nError != asio::error::operation_aborted)
{
ScheduleReadOne(oBuffer);
}
}
{
}
#define LOG_INFO(...)
Logs an info message.
Definition log.h:388
#define RETURN_IF_THROWS_DESC(s,...)
Definition result.h:137
UDP Receiver
udp_source_from_non_adtf.h
#pragma once
#include <adtf_base.h>
#include <adtf_filtersdk.h>
#include <adtf_systemsdk.h>
#include "udp_socket_wrapper_intf.h"
#ifndef ADTF_EXAMPLES_CID
#define ADTF_EXAMPLES_CID ".local.cid"
#endif
{
public:
"demo_foreign_application_udp_receiver.streaming_source" ADTF_EXAMPLES_CID,
"UDP Receiver From Non ADTF Application");
cUdpSourceFromNonADTFSystem();
~cUdpSourceFromNonADTFSystem() override;
private:
void SocketThread();
void PackToSample(const void* pData, size_t nBytes);
};
udp_source_from_non_adtf.cpp
#include "udp_source_from_non_adtf.h"
#include <functional>
using namespace adtf::util;
cUdpSourceFromNonADTFSystem::cUdpSourceFromNonADTFSystem()
{
m_strInterface.SetDescription("If set, bind the socket to this local interface.");
RegisterPropertyVariable("interface", m_strInterface);
m_nListeningPort.SetDescription("The UDP port to bind to and listen on.");
m_nListeningPort.SetValidRange(1, 65535);
RegisterPropertyVariable("port", m_nListeningPort);
m_strMulticastGroup.SetDescription("If set, join the given multicast group.\n"
"Depending on the platform, only multicast traffic from this specific group can "
"be received when this option is provided!");
RegisterPropertyVariable("multicast_group", m_strMulticastGroup);
m_strDDLStructName.SetDescription(
"If set, the stream type will contain the media description of the given struct, retrieved from the "
"description file specified within property 'ddl_description_file' if a valid filepath is set. If no file is "
"set, it will be retrieved from the (Deprecated) Media Description Service.");
RegisterPropertyVariable("ddl_struct_name", m_strDDLStructName);
m_strDescriptionFile.SetDescription(
"A description file containing the DDL description. If empty, the media description specified within the "
"(deprecated) ADTF Media Description Service will be used.");
RegisterPropertyVariable("ddl_description_file", m_strDescriptionFile);
m_bDeserializeViaMediaDescription.SetDescription(
"If activated, incoming data will be deserialized as defined "
"by the media description of the struct selected via ddl_struct_name before writing it into output samples.");
RegisterPropertyVariable("deserialize_via_media_description", m_bDeserializeViaMediaDescription);
m_bMarkSamplesAsDeserialized.SetDescription("In case that 'deserialize_via_media_description' is disabled, this "
"will force the output stream type to have the deserialized flag set.");
RegisterPropertyVariable("mark_samples_as_deserialized", m_bMarkSamplesAsDeserialized);
m_nReceiveBufferSize.SetDescription("Maximum payload size in bytes this filter can accept. Choose according to "
"configured MTU. Values up to 64kB are plausible from UDP side.");
m_nReceiveBufferSize.SetValidRange(1, 0xFFFF);
RegisterPropertyVariable("receive_buffer_size", m_nReceiveBufferSize);
m_nSocketReceiveBufferSize.SetDescription(
"Sockets receive buffer size in bytes (min: 4098, Default: 0 - uses system default)");
RegisterPropertyVariable("socket_receive_buffer_size", m_nSocketReceiveBufferSize);
m_pOutputWriter = CreateOutputPin("output");
SetDescription("output", "Provides the sample data received from a Non-ADTF instance over UDP");
m_pSocket = IUdpSocketPrivate::Create(
std::bind(&cUdpSourceFromNonADTFSystem::PackToSample, this, std::placeholders::_1, std::placeholders::_2));
SetDescription("socket", "Interface server to provide a shared socket for communication");
m_oSocketThread =
SetDescription("socket_thread",
"Connect a Thread Runner that will provide the context for receiving all data from the socket. "
"If this is not connected, the source will create a thread on its own.");
SetDescription(
"Use this streaming source to receive sample data from a Non-ADTF Application using the UDP protocol");
SetHelpLink("$(ADTF_DIR)/doc/html/page_demo_non_adtf_application_sender_receiver.html");
}
cUdpSourceFromNonADTFSystem::~cUdpSourceFromNonADTFSystem() = default;
tResult cUdpSourceFromNonADTFSystem::Construct()
{
RETURN_IF_FAILED(cSampleStreamingSource::Construct());
if (!m_strDDLStructName->empty())
{
{
const auto oDataRep = m_bDeserializeViaMediaDescription || m_bMarkSamplesAsDeserialized ?
if (m_strDescriptionFile->IsNotEmpty())
{
oDataDefinition, oDataRep);
}
else
{
ADTF3_IGNORE_DEPRECATION_WARNING_BEGIN
oDataRep);
ADTF3_IGNORE_DEPRECATION_WARNING_END
}
}();
m_pOutputWriter->ChangeType(pType);
if (m_bDeserializeViaMediaDescription)
{
m_oCodecFactory =
}
}
else
{
if (m_bDeserializeViaMediaDescription)
{
RETURN_ERROR_DESC(ERR_INVALID_ARG,
"Deserialization requires a ddl struct name to be set.");
}
}
}
tResult cUdpSourceFromNonADTFSystem::Init()
{
RETURN_IF_FAILED(cSampleStreamingSource::Init());
RETURN_IF_FAILED(_runtime->GetObject(m_pClock));
if (*m_nListeningPort == 0)
{
}
RETURN_IF_FAILED(m_pSocket->Open(m_strInterface->c_str(), *m_nListeningPort, m_strMulticastGroup->c_str(),
*m_nReceiveBufferSize,
*m_nSocketReceiveBufferSize));
}
tResult cUdpSourceFromNonADTFSystem::StartStreaming()
{
RETURN_IF_FAILED(cSampleStreamingSource::StartStreaming());
RETURN_IF_FAILED(m_pSocket->Start());
RETURN_IF_FAILED(m_oSocketThread.Activate());
}
tResult cUdpSourceFromNonADTFSystem::StopStreaming()
{
RETURN_IF_FAILED(m_pSocket->Stop());
m_oSocketThread.Deactivate();
return cSampleStreamingSource::StopStreaming();
}
void cUdpSourceFromNonADTFSystem::SocketThread()
{
const tResult nResult = m_pSocket->Run();
{
}
}
void cUdpSourceFromNonADTFSystem::PackToSample(const void* pData, size_t nBytes)
{
const auto tmNow = m_pClock->GetStreamTimeNs();
if (m_bDeserializeViaMediaDescription)
{
(TO_ADTF_RESULT(oDecoder.isValid()));
auto oCodec =
}
else
{
}
m_pOutputWriter->Write(pSample);
m_pOutputWriter->ManualTrigger(tmNow);
}
#define LOG_RESULT(code)
Log result to console.
Definition log.h:15
UDP Sender
udp_sink_to_non_adtf.h
#pragma once
#include <sink_to_non_adtf.h>
#include "udp_socket_wrapper_intf.h"
class cUdpSinkToNonADTFSystem final : public cSinkToNonADTFSystem
{
public:
"demo_foreign_application_udp_sender.streaming_sink" ADTF_EXAMPLES_CID,
"UDP Sender To Non ADTF Application");
cUdpSinkToNonADTFSystem();
~cUdpSinkToNonADTFSystem() override;
private:
void Send(const void* pData, size_t nDataSize) override;
IUdpSocket& GetSocket();
};
udp_sink_to_non_adtf.cpp
#include "udp_sink_to_non_adtf.h"
#include "udp_socket_wrapper.h"
using namespace adtf::util;
using namespace asio::ip;
cUdpSinkToNonADTFSystem::cUdpSinkToNonADTFSystem()
{
m_strInterface.SetDescription("If set, bind the socket to this local interface.\n"
"Requires source_port to be set too.\n"
"Ignored when socket interface client is connected.");
RegisterPropertyVariable("source_interface", m_strInterface);
m_nSourcePort.SetDescription(
"The UDP port to bind to as source. When left at 0, the port may be randomized. Ignored when socket interface client is connected.");
m_nSourcePort.SetValidRange(0, 65535);
RegisterPropertyVariable("source_port", m_nSourcePort);
m_strRemoteHost.SetDescription(
"If set, send data to this host. This is required if the sink is not connected to a source.\n"
"If this property is empty, packets will be sent to the last endpoint that data was received from.\n"
"If set to 'localhost' the IPv4 localhost address '127.0.0.1' is used.\n"
"IPv4 Example: '127.0.0.1'\n"
"IPv6 Example: '::1'\n");
RegisterPropertyVariable("remote_host", m_strRemoteHost);
m_nRemotePort.SetDescription("This is only taken into account when remote_host is set as well.");
m_nRemotePort.SetValidRange(1, 65535);
RegisterPropertyVariable("port", m_nRemotePort);
m_oUdpSocketClient = CreateInterfaceClient<IUdpSocket>("socket");
SetDescription("socket", "Interface client to use a shared socket for communication.\n"
"The connected socket provides the local host and port to bound as source.");
SetDescription("Use this streaming sink to transmit sample data to a Non-ADTF Application using the UDP protocol");
SetHelpLink("$(ADTF_DIR)/doc/html/page_udp_receiver_from_non_adtf_application.html");
}
cUdpSinkToNonADTFSystem::~cUdpSinkToNonADTFSystem() = default;
tResult cUdpSinkToNonADTFSystem::Init()
{
RETURN_IF_FAILED(cSampleStreamingSink::Init());
if (!m_oUdpSocketClient.IsValid())
{
if (m_strRemoteHost->empty())
{
"Remote host can not be empty if socket interface is not connected to a sink.");
}
RETURN_IF_FAILED(pSocket->Open(m_strInterface->c_str(), *m_nSourcePort, m_strRemoteHost->c_str(), 0, 0));
m_pFallbackSocket = pSocket;
}
if (!m_strRemoteHost->empty())
{
if (*m_nRemotePort != 0)
{
RETURN_IF_FAILED(GetSocket().SetRemote(m_strRemoteHost->c_str(), *m_nRemotePort));
}
else
{
RETURN_ERROR_DESC(ERR_INVALID_ARG,
"Remote port can never be 0 when remote host is specified.");
}
}
}
tResult cUdpSinkToNonADTFSystem::StartStreaming()
{
RETURN_IF_FAILED(cSampleStreamingSink::StartStreaming());
}
void cUdpSinkToNonADTFSystem::Send(const void* pData, size_t nDataSize)
{
const auto oResult = GetSocket().Send(pData, nDataSize);
{
}
}
IUdpSocket& cUdpSinkToNonADTFSystem::GetSocket()
{
return m_pFallbackSocket ? *m_pFallbackSocket : m_oUdpSocketClient.Get();
}
UDP Plugin
udp_plugin.cpp
#include "udp_source_from_non_adtf.h"
#include "udp_sink_to_non_adtf.h"
cUdpSourceFromNonADTFSystem,
cUdpSinkToNonADTFSystem);
Test
serialization.description
<?xml version="1.0" encoding="iso-8859-1" standalone="no"?>
<adtf:ddl xmlns:adtf="adtf">
<header>
<language_version>4.00</language_version>
<author>AUDI Electronics Venture GmbH</author>
<date_creation>20130703</date_creation>
<date_change />
<description>ADTF Common Description File</description>
</header>
<units/>
<datatypes/>
<structs>
<struct alignment="4" name="tTestSerialization" version="1" ddlversion="2.0">
<element name="nValue1" type="tUInt32" arraysize="1">
<serialized byteorder="LE" bytepos="4" bitpos="0" numbits="32"/>
<deserialized alignment="4"/>
</element>
<element name="nValue2" type="tUInt32" arraysize="1">
<serialized byteorder="LE" bytepos="0" bitpos="0" numbits="32"/>
<deserialized alignment="4"/>
</element>
</struct>
</structs>
<streams/>
<enums/>
<streammetatypes/>
</adtf:ddl>
test_non_adtf_sinks_and_sources.cpp
#ifdef _WIN32
#define WIN32_LEAN_AND_MEAN
#define NOMINMAX
#define NOGDI
#define NOUSER
#endif
#include <asio.hpp>
#ifdef CreateService
#undef CreateService
#endif
using namespace adtf::util;
#ifndef ADTF_EXAMPLES_CID
#define ADTF_EXAMPLES_CID ".local.cid"
#endif
constexpr char CID_TCP_SOURCE[] = "demo_foreign_application_tcp_receiver.streaming_source" ADTF_EXAMPLES_CID;
constexpr char CID_TCP_SINK[] = "demo_foreign_application_tcp_sender.streaming_sink" ADTF_EXAMPLES_CID;
constexpr char CID_UDP_SOURCE[] = "demo_foreign_application_udp_receiver.streaming_source" ADTF_EXAMPLES_CID;
constexpr char CID_UDP_SINK[] = "demo_foreign_application_udp_sender.streaming_sink" ADTF_EXAMPLES_CID;
{
cMyTestSystem(bool bWithLogging = false, bool bUseMds = false): cTestSystem({}, {}, false, bWithLogging)
{
if (bUseMds)
{
tADTFRunLevel::RL_System,
{{"media_description_files", ADTF_TESTING_SOURCE_DIR "/test_serialization.description"}});
configure_media_description_service();
}
}
~cMyTestSystem()
{
}
void configure_media_description_service()
{
object_ptr<adtf::services::IMediaDescriptionService> pService;
IConfiguration* pConfig = ucom_cast<IConfiguration*>(pService.Get());
set_property<cString>(*pConfig, "media_description_files",
ADTF_TESTING_SOURCE_DIR "/serialization.description");
}
};
struct cMyTestSystemWithLog : public cMyTestSystem
{
cMyTestSystemWithLog(): cMyTestSystem(true)
{
}
};
struct tTestGraph
{
~tTestGraph()
{
CHECK_OK(pGraph->SetState(IFilterGraph::tFilterState::State_Constructed));
pGraph.Reset();
}
object_ptr<IFilterGraph> pGraph;
std::unique_ptr<cOutputRecorder> pSourceOutput;
std::unique_ptr<cTestWriter> pSinkInput;
};
class cTestApplication
{
public:
cTestApplication(const cString& strArguments)
{
REQUIRE_OK(adtf::util::cSystem::ChildExecute(&m_nPID, DEMO_APPLICATION, strArguments));
std::this_thread::sleep_for(std::chrono::seconds(1));
}
~cTestApplication()
{
adtf::util::cSystem::ChildTerminate(m_nPID);
}
private:
uint64_t m_nPID;
};
tTestGraph create_graph(const cString& strSourceCid,
const std::map<cString, cString>& oSourceConfig,
const cString& strSinkCid,
const std::map<cString, cString>& oSinkConfig,
bool bConnectSockets = true)
{
pGraph->SetName("graph");
if (bConnectSockets)
{
REQUIRE_OK(pGraph->AddConnection(
"connection1",
"source",
"socket",
"socket",
"", 0,
true));
REQUIRE_OK(pGraph->AddConnection(
"connection2",
"socket",
"",
"sink",
"socket", 0,
true));
}
REQUIRE_OK(pGraph->SetState(IFilterGraph::tFilterGraphState::State_Initialized));
return {pGraph, std::make_unique<cOutputRecorder>(pSource, "output"),
std::make_unique<cTestWriter>(pSink, "input", oType)};
}
void test_raw(tTestGraph& oGraph)
{
REQUIRE_OK(oGraph.pGraph->SetState(IFilterGraph::tFilterState::State_Running));
for (uint64_t nCounter = 123456789; nCounter < 123456789 + 100; ++nCounter)
{
oGraph.pSinkInput->Write(std::chrono::seconds(0), nCounter, true);
INFO("WaitForTrigger on counter = " + std::to_string(nCounter));
REQUIRE(oGraph.pSourceOutput->WaitForTrigger(std::chrono::seconds(1)));
auto oOutput = oGraph.pSourceOutput->GetCurrentOutput();
REQUIRE(!oOutput.GetSamples().empty());
{
REQUIRE(oData == nCounter);
}
}
}
void test_serialization(tTestGraph& oGraph)
{
REQUIRE_OK(oGraph.pGraph->SetState(IFilterGraph::tFilterState::State_Running));
tTestSerialization sTest{1, 2};
oGraph.pSinkInput->Write(std::chrono::seconds(0), sTest, true);
REQUIRE(oGraph.pSourceOutput->WaitForTrigger(std::chrono::seconds(1)));
auto oOutput = oGraph.pSourceOutput->GetCurrentOutput();
REQUIRE(!oOutput.GetSamples().empty());
{
REQUIRE(oData->nValue1 == sTest.nValue2);
REQUIRE(oData->nValue2 == sTest.nValue1);
}
}
TEST_CASE_METHOD(cMyTestSystem, "Test Raw TCP / IPv4", "[req:ACORE-8868][req:ACORE-10086]")
{
const auto nServerPort = GetAvailablePort(TCP_IPv4);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort) + " -tcp");
auto oGraph = create_graph(CID_TCP_SOURCE,
{
{"remote_host", "127.0.0.1"},
{"remote_port", cString::FromType(nServerPort)},
},
CID_TCP_SINK,
{
});
test_raw(oGraph);
}
TEST_CASE_METHOD(cMyTestSystem, "Test Raw UDP / IPv4", "[req:ACORE-8868][req:ACORE-10086]")
{
const auto nServerPort = GetAvailablePort(UDP_IPv4);
const auto nLocalPort = GetAvailablePort(UDP_IPv4);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort));
auto oGraph = create_graph(CID_UDP_SOURCE,
{
{"interface", "127.0.0.1"},
{"port", cString::FromType(nLocalPort)},
},
CID_UDP_SINK,
{
{"remote_host", "127.0.0.1"},
{"port", cString::FromType(nServerPort)},
});
test_raw(oGraph);
}
TEST_CASE_METHOD(cMyTestSystemWithLog, "Test Raw receive buffer size set", "[req:gitlab-#2939]")
{
const auto nServerPort = GetAvailablePort(UDP_IPv4);
const auto nLocalPort = GetAvailablePort(UDP_IPv4);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort));
auto oGraph = create_graph(CID_UDP_SOURCE,
{
{"interface", "127.0.0.1"},
{"port", cString::FromType(nLocalPort)},
{"socket_receive_buffer_size", std::to_string(2 * 1024 * 1024).c_str()},
},
CID_UDP_SINK,
{
{"remote_host", "127.0.0.1"},
{"port", cString::FromType(nServerPort)},
});
test_raw(oGraph);
CHECK(oMessages.ContainsMessage("Sockets buffersize set from"));
CHECK(oMessages.ContainsMessage("to " + std::to_string(2 * 1024 * 1024) + " bytes"));
}
TEST_CASE_METHOD(cMyTestSystem, "Test Raw UDP Multicast / IPv4", "[req:ACORE-11192][!mayfail]")
{
const auto nServerPort = GetAvailablePort(UDP_IPv4);
const auto nLocalPort = GetAvailablePort(UDP_IPv4);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort) + " -m=239.0.2.1 -r=" + cString::FromType(nLocalPort));
auto oGraph = create_graph(CID_UDP_SOURCE,
{
{"interface", "127.0.0.1"},
{"multicast_group", "239.0.2.1"},
{"port", cString::FromType(nLocalPort)},
},
CID_UDP_SINK,
{
{"remote_host", "239.0.2.1"},
{"port", cString::FromType(nServerPort)},
});
test_raw(oGraph);
}
TEST_CASE_METHOD(cMyTestSystem, "Test Raw UDP Multicast / IPv6", "[req:ACORE-11192]")
{
const auto nServerPort = GetAvailablePort(UDP_IPv6);
const auto nLocalPort = GetAvailablePort(UDP_IPv6);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort) + " -m=ff11::114 -ipv6 -r=" + cString::FromType(nLocalPort));
auto oGraph = create_graph(CID_UDP_SOURCE,
{
{"interface", "::1"},
{"multicast_group", "ff11::114"},
{"port", cString::FromType(nLocalPort)},
},
CID_UDP_SINK,
{
{"remote_host", "ff11::114"},
{"port", cString::FromType(nServerPort)},
});
test_raw(oGraph);
}
TEST_CASE_METHOD(cMyTestSystem, "Test Raw TCP / IPv6", "[req:ACORE-8868][req:ACORE-10086]")
{
const auto nServerPort = GetAvailablePort(TCP_IPv6);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort) + " -tcp -ipv6");
auto oGraph = create_graph(CID_TCP_SOURCE,
{
{"remote_host", "::1"},
{"remote_port", cString::FromType(nServerPort)},
},
CID_TCP_SINK, {});
test_raw(oGraph);
}
TEST_CASE_METHOD(cMyTestSystem, "Test Raw UDP / IPv6", "[req:ACORE-8868][req:ACORE-10086]")
{
const auto nLocalPort = GetAvailablePort(UDP_IPv6);
const auto nServerPort = GetAvailablePort(UDP_IPv6);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort) + " -ipv6");
auto oGraph = create_graph(CID_UDP_SOURCE,
{
{"port", cString::FromType(nLocalPort)},
{"interface", "::1"},
},
CID_UDP_SINK,
{
{"remote_host", "::1"},
{"port", cString::FromType(nServerPort)},
});
test_raw(oGraph);
}
void source_deserial_test_tcp_ipv4(bool bUseMds)
{
cMyTestSystem oSystem(false, bUseMds);
const auto nServerPort = oSystem.GetAvailablePort(oSystem.TCP_IPv4);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort) + " -tcp");
auto oGraph = create_graph(CID_TCP_SOURCE,
{
{"remote_host", "127.0.0.1"},
{"remote_port", cString::FromType(nServerPort)},
{"ddl_struct_name", "tTestSerialization"},
{"ddl_description_file", bUseMds ? "" : ADTF_TESTING_SOURCE_DIR "/serialization.description"},
{"deserialize_via_media_description", "true"},
},
test_serialization(oGraph);
}
TEST_CASE("Test Source Deserialization TCP / IPv4", "[req:ACORE-8868][req:ACORE-10086]")
{
const bool bUseMediaDescriptionService = GENERATE(false, true);
source_deserial_test_tcp_ipv4(bUseMediaDescriptionService);
}
TEST_CASE_METHOD(cMyTestSystem, "Test Source Deserialization TCP / IPv6", "[req:ACORE-8868][req:ACORE-10086]")
{
const auto nServerPort = GetAvailablePort(TCP_IPv6);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort) + " -tcp -ipv6");
auto oGraph = create_graph(CID_TCP_SOURCE,
{
{"remote_host", "::1"},
{"remote_port", cString::FromType(nServerPort)},
{"ddl_description_file", ADTF_TESTING_SOURCE_DIR "/serialization.description"},
{"ddl_struct_name", "tTestSerialization"},
{"deserialize_via_media_description", "true"},
},
test_serialization(oGraph);
}
void source_deserial_test_udp_ipv4(bool bUseMds)
{
cMyTestSystem oSystem(false, bUseMds);
const auto nLocalPort = oSystem.GetAvailablePort(oSystem.UDP_IPv4);
const auto nServerPort = oSystem.GetAvailablePort(oSystem.UDP_IPv4);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort));
auto oGraph = create_graph(CID_UDP_SOURCE,
{
{"interface", "127.0.0.1"},
{"port", cString::FromType(nLocalPort)},
{"ddl_struct_name", "tTestSerialization"},
{"ddl_description_file", bUseMds ? "" : ADTF_TESTING_SOURCE_DIR "/serialization.description"},
{"deserialize_via_media_description", "true"},
},
CID_UDP_SINK,
{
{"remote_host", "127.0.0.1"},
{"port", cString::FromType(nServerPort)},
});
test_serialization(oGraph);
}
TEST_CASE("Test Source Deserialization UDP / IPv4", "[req:ACORE-8868][req:ACORE-10086]")
{
const bool bUseMediaDescriptionService = GENERATE(false, true);
source_deserial_test_udp_ipv4(bUseMediaDescriptionService);
}
TEST_CASE_METHOD(cMyTestSystem, "Test Source Deserialization UDP / IPv6", "[req:ACORE-8868][req:ACORE-10086]")
{
const auto nLocalPort = GetAvailablePort(UDP_IPv6);
const auto nServerPort = GetAvailablePort(UDP_IPv6);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort) + " -ipv6");
auto oGraph = create_graph(CID_UDP_SOURCE,
{
{"interface", "::1"},
{"port", cString::FromType(nLocalPort)},
{"ddl_description_file", ADTF_TESTING_SOURCE_DIR "/serialization.description"},
{"ddl_struct_name", "tTestSerialization"},
{"deserialize_via_media_description", "true"},
},
CID_UDP_SINK,
{
{"remote_host", "::1"},
{"port", cString::FromType(nServerPort)},
});
test_serialization(oGraph);
}
TEST_CASE_METHOD(cMyTestSystem, "Test Sink Serialization TCP / IPv4", "[req:ACORE-10086]")
{
const auto nServerPort = GetAvailablePort(TCP_IPv4);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort) + " -tcp");
auto oGraph = create_graph(CID_TCP_SOURCE,
{
{"remote_host", "127.0.0.1"},
{"remote_port", cString::FromType(nServerPort)},
{"ddl_description_file", ADTF_TESTING_SOURCE_DIR "/serialization.description"},
{"ddl_struct_name", "tTestSerialization"},
},
CID_TCP_SINK, {{"serialize_via_media_description", "true"}},
test_serialization(oGraph);
}
TEST_CASE_METHOD(cMyTestSystem, "Test Sink Serialization TCP / IPv6", "[req:ACORE-10086]")
{
const auto nServerPort = GetAvailablePort(TCP_IPv6);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort) + " -tcp -ipv6");
auto oGraph = create_graph(CID_TCP_SOURCE,
{
{"remote_host", "::1"},
{"remote_port", cString::FromType(nServerPort)},
{"ddl_description_file", ADTF_TESTING_SOURCE_DIR "/serialization.description"},
{"ddl_struct_name", "tTestSerialization"},
},
CID_TCP_SINK,
{
{"serialize_via_media_description", "true"},
},
test_serialization(oGraph);
}
TEST_CASE_METHOD(cMyTestSystem, "Test Sink Serialization UDP / IPv4", "[req:ACORE-10086]")
{
const auto nLocalPort = GetAvailablePort(UDP_IPv4);
const auto nServerPort = GetAvailablePort(UDP_IPv4);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort));
auto oGraph = create_graph(CID_UDP_SOURCE,
{
{"interface", "127.0.0.1"},
{"port", cString::FromType(nLocalPort)},
{"ddl_description_file", ADTF_TESTING_SOURCE_DIR "/serialization.description"},
{"ddl_struct_name", "tTestSerialization"},
},
CID_UDP_SINK,
{
{"serialize_via_media_description", "true"},
{"remote_host", "127.0.0.1"},
{"port", cString::FromType(nServerPort)},
},
test_serialization(oGraph);
}
TEST_CASE_METHOD(cMyTestSystem, "Test Sink Serialization UDP / IPv6", "[req:ACORE-10086]")
{
const auto nLocalPort = GetAvailablePort(UDP_IPv6);
const auto nServerPort = GetAvailablePort(UDP_IPv6);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort) + " -ipv6");
auto oGraph = create_graph(CID_UDP_SOURCE,
{
{"interface", "::1"},
{"port", cString::FromType(nLocalPort)},
{"ddl_description_file", ADTF_TESTING_SOURCE_DIR "/serialization.description"},
{"ddl_struct_name", "tTestSerialization"},
},
CID_UDP_SINK,
{
{"serialize_via_media_description", "true"},
{"remote_host", "::1"},
{"port", cString::FromType(nServerPort)},
},
test_serialization(oGraph);
}
TEST_CASE_METHOD(cMyTestSystem, "Test TCP Source Fixed Packet Size", "[req:ACORE-10086]")
{
const auto nServerPort = GetAvailablePort(TCP_IPv4);
cTestApplication oApplication("-p=" + cString::FromType(nServerPort) + " -tcp");
auto oGraph = create_graph(CID_TCP_SOURCE,
{
{"remote_host", "127.0.0.1"},
{"remote_port", cString::FromType(nServerPort)},
{"fixed_packet_size", "8"},
},
CID_TCP_SINK, {});
REQUIRE_OK(oGraph.pGraph->SetState(IFilterGraph::tFilterState::State_Running));
uint32_t nValue1 = 1;
uint32_t nValue2 = 2;
oGraph.pSinkInput->Write(std::chrono::seconds(0), nValue1, true);
REQUIRE(!oGraph.pSourceOutput->WaitForTrigger(std::chrono::seconds(1)));
oGraph.pSinkInput->Write(std::chrono::seconds(0), nValue2, true);
REQUIRE(oGraph.pSourceOutput->WaitForTrigger(std::chrono::seconds(1)));
auto oOutput = oGraph.pSourceOutput->GetCurrentOutput();
REQUIRE(!oOutput.GetSamples().empty());
{
REQUIRE((*oData)[0] == nValue1);
REQUIRE((*oData)[1] == nValue2);
}
}
TEST_CASE_METHOD(cMyTestSystem, "Independent UDP Source and Sink Sockets / IPv4", "[req:ACORE-10086]")
{
const auto nServerPort = GetAvailablePort(UDP_IPv4);
auto oGraph = create_graph(CID_UDP_SOURCE,
{
{"interface", "127.0.0.1"},
{"port", cString::FromType(nServerPort)},
},
CID_UDP_SINK,
{
{"source_interface", "127.0.0.1"},
{"remote_host", "127.0.0.1"},
{"port", cString::FromType(nServerPort)},
},
REQUIRE_OK(oGraph.pGraph->SetState(IFilterGraph::tFilterState::State_Running));
uint32_t nValue1 = 1;
oGraph.pSinkInput->Write(std::chrono::seconds(0), nValue1, true);
REQUIRE(oGraph.pSourceOutput->WaitForTrigger(std::chrono::seconds(1)));
auto oOutput = oGraph.pSourceOutput->GetCurrentOutput();
REQUIRE(!oOutput.GetSamples().empty());
{
REQUIRE(oData == nValue1);
}
}
TEST_CASE_METHOD(cMyTestSystem, "Independent UDP Source and Sink Sockets / IPv6", "[req:ACORE-10086]")
{
const auto nServerPort = GetAvailablePort(UDP_IPv6);
auto oGraph = create_graph(CID_UDP_SOURCE,
{
{"interface", "::1"},
{"port", cString::FromType(nServerPort)},
},
CID_UDP_SINK,
{
{"source_interface", "::1"},
{"remote_host", "::1"},
{"port", cString::FromType(nServerPort)},
},
REQUIRE_OK(oGraph.pGraph->SetState(IFilterGraph::tFilterState::State_Running));
uint32_t nValue1 = 1;
oGraph.pSinkInput->Write(std::chrono::seconds(0), nValue1, true);
REQUIRE(oGraph.pSourceOutput->WaitForTrigger(std::chrono::seconds(1)));
auto oOutput = oGraph.pSourceOutput->GetCurrentOutput();
REQUIRE(!oOutput.GetSamples().empty());
{
REQUIRE(oData == nValue1);
}
}
TEST_CASE_METHOD(cMyTestSystem, "Independent UDP Source and Sink Sockets Multicast / IPv4", "[req:ACORE-11192][!mayfail]")
{
const auto nServerPort = GetAvailablePort(UDP_IPv4);
auto oGraph = create_graph(CID_UDP_SOURCE,
{
{"interface", "127.0.0.1"},
{"port", cString::FromType(nServerPort)},
{"multicast_group", "239.0.2.2"},
},
CID_UDP_SINK,
{
{"source_interface", "127.0.0.1"},
{"remote_host", "239.0.2.2"},
{"port", cString::FromType(nServerPort)},
},
REQUIRE_OK(oGraph.pGraph->SetState(IFilterGraph::tFilterState::State_Running));
uint32_t nValue1 = 1;
oGraph.pSinkInput->Write(std::chrono::seconds(0), nValue1, true);
REQUIRE(oGraph.pSourceOutput->WaitForTrigger(std::chrono::seconds(1)));
auto oOutput = oGraph.pSourceOutput->GetCurrentOutput();
REQUIRE(!oOutput.GetSamples().empty());
{
REQUIRE(oData == nValue1);
}
}
TEST_CASE_METHOD(cMyTestSystem, "Independent UDP Source and Sink Sockets Multicast / IPv6", "[req:ACORE-11192]")
{
const auto nServerPort = GetAvailablePort(UDP_IPv6);
auto oGraph = create_graph(CID_UDP_SOURCE,
{
{"interface", "::1"},
{"port", cString::FromType(nServerPort)},
{"multicast_group", "ff11::114"},
},
CID_UDP_SINK,
{
{"source_interface", "::1"},
{"remote_host", "ff11::114"},
{"port", cString::FromType(nServerPort)},
},
REQUIRE_OK(oGraph.pGraph->SetState(IFilterGraph::tFilterState::State_Running));
uint32_t nValue1 = 1;
oGraph.pSinkInput->Write(std::chrono::seconds(0), nValue1, true);
REQUIRE(oGraph.pSourceOutput->WaitForTrigger(std::chrono::seconds(1)));
auto oOutput = oGraph.pSourceOutput->GetCurrentOutput();
REQUIRE(!oOutput.GetSamples().empty());
{
REQUIRE(oData == nValue1);
}
}
void tcp_server(uint16_t nServerPort, std::promise<void> oListening)
{
cServerSocket oServer;
oListening.set_value();
cStreamSocket oClient1;
cStreamSocket oClient2;
uint8_t nBuffer;
}
TEST_CASE_METHOD(cMyTestSystem, "Independent TCP Source and Sink Sockets", "[req:ACORE-10086]")
{
const auto nServerPort = GetAvailablePort(TCP_IPv4);
auto oGraph = create_graph(CID_TCP_SOURCE,
{
{"remote_host", "127.0.0.1"},
{"remote_port", cString::FromType(nServerPort)},
},
CID_TCP_SINK,
{
{"remote_host", "127.0.0.1"},
{"remote_port", cString::FromType(nServerPort)},
},
std::promise<void> oListening;
auto oListeningStarted = oListening.get_future();
auto oServerResult = std::async(std::launch::async, tcp_server, nServerPort, std::move(oListening));
oListeningStarted.get();
REQUIRE_OK(oGraph.pGraph->SetState(IFilterGraph::tFilterState::State_Running));
uint8_t nValue1 = 123;
oGraph.pSinkInput->Write(std::chrono::seconds(0), nValue1, true);
REQUIRE(oGraph.pSourceOutput->WaitForTrigger(std::chrono::seconds(1)));
auto oOutput = oGraph.pSourceOutput->GetCurrentOutput();
REQUIRE(!oOutput.GetSamples().empty());
{
REQUIRE(oData == nValue1);
}
oServerResult.get();
}
void tcp_server_source(uint16_t nServerPort, std::promise<void> oListening)
{
cServerSocket oServer;
oListening.set_value();
cStreamSocket oClient1;
uint8_t nBuffer = 123;
}
TEST_CASE_METHOD(cMyTestSystem, "Test TCP Source Reconnection", "[req:ACORE-11249]")
{
const auto nServerPort = GetAvailablePort(TCP_IPv4);
REQUIRE_OK(_runtime->CreateInstance(CID_TCP_SOURCE, pSource));
set_property(*pConfiguration,
"remote_port", cString::FromType(nServerPort));
SECTION("disabled reconnection")
{
set_property(*pConfiguration,
"enable_automatic_reconnection",
false);
REQUIRE_OK(pSource->SetState(IStreamingService::tStreamingState::State_Initialized));
REQUIRE_FAILED(pSource->SetState(IStreamingService::tStreamingState::State_Streaming));
}
SECTION("enabled reconnection")
{
set_property(*pConfiguration,
"enable_automatic_reconnection",
true);
REQUIRE_OK(pSource->SetState(IStreamingService::tStreamingState::State_Initialized));
REQUIRE_OK(pSource->SetState(IStreamingService::tStreamingState::State_Streaming));
for (size_t nCounter = 0; nCounter < 5; ++nCounter)
{
REQUIRE(!oRecorder.WaitForTrigger(std::chrono::milliseconds(500)));
std::promise<void> oListening;
auto oListeningStarted = oListening.get_future();
auto oServerResult = std::async(std::launch::async, tcp_server_source, nServerPort, std::move(oListening));
oListeningStarted.get();
REQUIRE(oRecorder.WaitForTrigger(std::chrono::seconds(5)));
REQUIRE(oRecorder.GetCurrentOutput().GetSamples().size() == 1);
oServerResult.get();
}
}
}
void tcp_server_sink(uint16_t nServerPort, std::promise<void> oListening, std::promise<void> oRecievedData)
{
cServerSocket oServer;
oListening.set_value();
cStreamSocket oClient1;
uint8_t nBuffer = 123;
oRecievedData.set_value();
LOG_INFO(
"Recieved data %" PRIu8, nBuffer);
}
TEST_CASE_METHOD(cMyTestSystem, "Test TCP Sink Reconnection", "[req:ACORE-11249]")
{
const auto nServerPort = GetAvailablePort(TCP_IPv4);
REQUIRE_OK(_runtime->CreateInstance(CID_TCP_SINK, pSink));
set_property(*pConfiguration,
"remote_port", cString::FromType(nServerPort));
SECTION("disabled reconnection")
{
set_property(*pConfiguration,
"enable_automatic_reconnection",
false);
REQUIRE_OK(pSink->SetState(IStreamingService::tStreamingState::State_Initialized));
REQUIRE_FAILED(pSink->SetState(IStreamingService::tStreamingState::State_Streaming));
}
SECTION("enabled reconnection")
{
set_property(*pConfiguration,
"enable_automatic_reconnection",
true);
REQUIRE_OK(pSink->SetState(IStreamingService::tStreamingState::State_Initialized));
REQUIRE_OK(pSink->SetState(IStreamingService::tStreamingState::State_Streaming));
for (uint8_t nCounter = 0; nCounter < 5; ++nCounter)
{
std::promise<void> oListening;
auto oListeningStarted = oListening.get_future();
std::promise<void> oRecievedData;
auto oReceivedDataDone = oRecievedData.get_future();
auto oServerResult = std::async(std::launch::async, tcp_server_sink, nServerPort, std::move(oListening),
std::move(oRecievedData));
oListeningStarted.get();
do
{
oWriter.Write(nCounter, true);
} while (oReceivedDataDone.wait_for(std::chrono::milliseconds(100)) == std::future_status::timeout);
oServerResult.get();
}
}
}
#define REQUIRE_OK(...)
Macro for a mandatory check if a function call returned with ERR_NOERROR. If not the test will fail a...
Definition catch_integration.h:75
#define REQUIRE_FAILED(...)
Macro for a mandatory check if a function call did not return with ERR_NOERROR. If it did the test wi...
Definition catch_integration.h:81
#define CHECK_OK(...)
Macro to check if a function call returned with ERR_NOERROR. If not the test will fail and the error ...
Definition catch_integration.h:88
Definition stream_type_helper.h:31
Definition output_recorder.h:36
Definition test_writer.h:40
Definition test_system.h:110
void SetRunLevel(base::ant::tADTFRunLevel eRunlevel)
ucom::ant::object_ptr< ucom::ant::IService > CreateService(const std::string &strClassId, const std::string &strObjectId, base::ant::tADTFRunLevel nRunlevel=base::tADTFRunLevel::RL_System)
void LoadPlugin(const std::string &strPluginFileName, bool bCreateServices=true, base::ant::tADTFRunLevel nServicesRunlevel=base::tADTFRunLevel::RL_System)
Definition test_logger.h:29
tMessages GetCurrentMessages(bool bClear=true)
virtual tResult GetObject(iobject_ptr< IObject > &pObject, const char *strNameOID) const =0
tResult set_property(IConfiguration &oConfiguration, const char *strNameOfValue, VALUETYPE oValue)
Set the property.
Definition configuration.h:327
Namespace for all testing functionality of the ADTF Filter SDK.
Definition create_default_triggers.h:17
tResult add_binding_proxy(ant::cGraph &oGraph, const util::cString &strName, int32_t nOrderNumber, ucom::iobject_ptr< ucom::IObject > &pBindingProxy)
adds a Binding Proxy to a graph and return the created proxy.
Definition graph_utils.h:117
tResult add_graph_object(ant::cGraph &oGraph, const util::cString &strCID, const util::cString &strName, const std::map< util::cString, util::cString > &oProperties, int32_t nOrderNumber, ucom::iobject_ptr< ucom::IObject > &pObject)
Convenience functionality to create and add add graph object to a existing graph.
Definition graph_utils.h:39
InterfacePointerType ucom_cast(ObjectPointerType i_pObject) noexcept
Alias always bringing the latest version of ant::ucom_cast() into scope.
Definition ucom_cast.h:157
Demo Application
demo_non_adtf_application.cpp
#include <a_utils.h>
#include <iostream>
#include <csignal>
#include <vector>
#include <optional>
#include <asio.hpp>
namespace udp
{
struct async_context
{
async_context(
const std::shared_ptr<asio::ip::udp::socket>& socket,
std::optional<asio::ip::udp::endpoint> multicast_endpoint):
socket(socket), multicast_endpoint(multicast_endpoint)
{
buffer.fill(0);
}
std::shared_ptr<asio::ip::udp::socket> socket;
std::array<uint8_t, 9000> buffer;
asio::ip::udp::endpoint endpoint;
std::optional<asio::ip::udp::endpoint> multicast_endpoint;
};
void schedule_read(
const std::error_code& nError,
const std::shared_ptr<async_context>& oContext);
void schedule_echo(
const std::error_code& nError,
size_t nBytesRead,
const std::shared_ptr<async_context>& context)
{
if (!nError)
{
if (context->multicast_endpoint)
{
std::cout << "Multicast echo for " << context->endpoint << " to " << *context->multicast_endpoint
<< std::endl;
context->socket->async_send_to(asio::buffer(context->buffer.data(), nBytesRead),
*context->multicast_endpoint,
std::bind(&schedule_read, std::placeholders::_1, context));
}
else
{
std::cout << "UDP echo to " << context->endpoint << std::endl;
context->socket->async_send_to(asio::buffer(context->buffer.data(), nBytesRead), context->endpoint,
std::bind(&schedule_read, std::placeholders::_1, context));
}
}
else
{
schedule_read(nError, context);
}
}
void schedule_read(
const std::error_code& nError,
const std::shared_ptr<async_context>& oContext)
{
if (nError)
{
std::cerr << nError.message();
}
if (nError != asio::error::operation_aborted)
{
oContext->socket->async_receive_from(
asio::buffer(oContext->buffer.data(), oContext->buffer.size()), oContext->endpoint,
std::bind(&schedule_echo, std::placeholders::_1, std::placeholders::_2, oContext));
}
}
void server(asio::io_context& oIoContext,
uint16_t nListenPort,
bool bIpv6,
const std::string& strMulticastGroup,
uint16_t nMulticastResponsePort)
{
std::cout << "Creating UDP server " << (bIpv6 ? "v6" : "v4") << " at port " << nListenPort
<< " mc: " << strMulticastGroup << std::endl;
std::shared_ptr<asio::ip::udp::socket> oUdpSocket = std::make_shared<asio::ip::udp::socket>(oIoContext);
oUdpSocket->open(bIpv6 ? asio::ip::udp::socket::protocol_type::v6() : asio::ip::udp::socket::protocol_type::v4());
oUdpSocket->set_option(asio::ip::udp::socket::reuse_address(true));
std::optional<asio::ip::udp::endpoint> oMulticastEndpoint;
if (!strMulticastGroup.empty())
{
asio::error_code oSetOutboundInterfaceError;
if (bIpv6)
{
const auto multicastAddress = asio::ip::make_address_v6(strMulticastGroup);
oMulticastEndpoint = asio::ip::udp::endpoint(multicastAddress, nMulticastResponsePort);
oUdpSocket->bind(asio::ip::udp::endpoint(asio::ip::udp::v6(), nListenPort));
oUdpSocket->set_option(asio::ip::multicast::outbound_interface(asio::ip::address_v6::loopback().scope_id()),
oSetOutboundInterfaceError);
if (!oSetOutboundInterfaceError)
{
oUdpSocket->set_option(
asio::ip::multicast::join_group(multicastAddress, asio::ip::address_v6::loopback().scope_id()));
}
else
{
oUdpSocket->set_option(asio::ip::multicast::join_group(multicastAddress));
}
}
else
{
const auto multicastAddress = asio::ip::make_address_v4(strMulticastGroup);
oMulticastEndpoint = asio::ip::udp::endpoint(multicastAddress, nMulticastResponsePort);
#ifdef _WIN32
oUdpSocket->bind(asio::ip::udp::endpoint(asio::ip::address_v4::loopback(), nListenPort));
#else
oUdpSocket->bind(asio::ip::udp::endpoint(multicastAddress, nListenPort));
#endif
oUdpSocket->set_option(asio::ip::multicast::outbound_interface(asio::ip::address_v4::loopback()),
oSetOutboundInterfaceError);
if (!oSetOutboundInterfaceError)
{
oUdpSocket->set_option(
asio::ip::multicast::join_group(multicastAddress, asio::ip::address_v4::loopback()));
}
else
{
oUdpSocket->set_option(asio::ip::multicast::join_group(multicastAddress));
}
}
}
else
{
oUdpSocket = std::make_shared<asio::ip::udp::socket>(
oIoContext,
asio::ip::udp::endpoint(bIpv6 ? static_cast<asio::ip::address>(asio::ip::address_v6::loopback()) :
static_cast<asio::ip::address>(asio::ip::address_v4::loopback()),
nListenPort));
}
for (size_t nBufferCount = 0; nBufferCount < 64; ++nBufferCount)
{
auto oContext = std::make_shared<udp::async_context>(oUdpSocket, oMulticastEndpoint);
schedule_read(std::error_code(), oContext);
}
std::cout << "Server ready" << std::endl;
}
}
namespace tcp
{
struct async_context
{
async_context(
const std::shared_ptr<asio::ip::tcp::socket>& socket): socket(socket)
{
}
std::shared_ptr<asio::ip::tcp::socket> socket;
std::array<uint8_t, 9000> buffer;
};
void schedule_read(
const std::error_code& nError,
const std::shared_ptr<async_context>& oContext);
void on_accept(const std::error_code& nError,
const std::shared_ptr<async_context>& context,
const std::shared_ptr<asio::ip::tcp::acceptor>& acceptor)
{
if (context)
{
schedule_read(nError, context);
}
if (nError != asio::error::operation_aborted)
{
auto nextSocket = std::make_shared<asio::ip::tcp::socket>(acceptor->get_executor());
auto nextContext = std::make_shared<async_context>(nextSocket);
acceptor->async_accept(*nextSocket, std::bind(&on_accept, std::placeholders::_1, nextContext, acceptor));
}
}
void schedule_echo(const std::error_code& nError,
size_t nBytesWritten,
asio::const_buffer oCurrentTxBuffer,
const std::shared_ptr<async_context>& oContext)
{
if (!nError)
{
oCurrentTxBuffer += nBytesWritten;
if (oCurrentTxBuffer.size() > 0)
{
oContext->socket->async_write_some(
oCurrentTxBuffer,
std::bind(&schedule_echo, std::placeholders::_1, std::placeholders::_2, oCurrentTxBuffer, oContext));
}
else
{
schedule_read(nError, oContext);
}
}
else
{
schedule_read(nError, oContext);
}
}
void on_read(
const std::error_code& nError,
size_t nBytesRead,
const std::shared_ptr<async_context>& oContext)
{
if (nError != asio::error::operation_aborted)
{
schedule_echo(nError, 0, asio::buffer(oContext->buffer.data(), nBytesRead), oContext);
}
}
void schedule_read(
const std::error_code& nError,
const std::shared_ptr<async_context>& oContext)
{
if (nError)
{
std::cerr << nError.message();
}
if (nError != asio::error::operation_aborted)
{
oContext->socket->async_read_some(asio::buffer(oContext->buffer.data(), oContext->buffer.size()),
std::bind(&on_read, std::placeholders::_1, std::placeholders::_2, oContext));
}
}
void server(asio::io_context& oIoContext, uint16_t nPort, bool bIpv6)
{
std::cout << "Creating TCP server at " << (bIpv6 ? "v6" : "v4") << " port " << nPort << std::endl;
{
auto oTcpAcceptor = std::make_shared<asio::ip::tcp::acceptor>(
oIoContext,
asio::ip::tcp::endpoint(bIpv6 ? static_cast<asio::ip::address>(asio::ip::address_v6::loopback()) :
static_cast<asio::ip::address>(asio::ip::address_v4::loopback()),
nPort));
oTcpAcceptor->set_option(asio::ip::tcp::acceptor::reuse_address(true));
on_accept(std::error_code(), nullptr, oTcpAcceptor);
}
}
}
namespace
{
asio::io_context oIoContext;
void signal_handler(int nSignalId)
{
oIoContext.stop();
std::signal(nSignalId, SIG_DFL);
}
}
void run_threaded(asio::io_context& context, size_t count)
{
std::list<std::thread> threads;
for (size_t i = 0; i < count; ++i)
{
threads.emplace_back([&]() { context.run(); });
}
while (!threads.empty())
{
threads.front().join();
threads.pop_front();
}
}
int main(int nArgn, const char** pArgv)
{
try
{
adtf::util::cCommandLine arguments(nArgn, pArgv);
std::signal(SIGINT, &signal_handler);
tUInt16 nListenPort = static_cast<uint16_t>(arguments.GetProperty("p", "54321").AsInt32());
tBool bTcp = arguments.GetFlag("tcp");
tBool bIpv6 = arguments.GetFlag("ipv6");
std::string strMulticastGroup = arguments.GetProperty("m").GetPtr();
tUInt16 nMulticastResponsePort = static_cast<uint16_t>(arguments.GetProperty("r", "54321").AsInt32());
if (bTcp)
{
tcp::server(oIoContext, nListenPort, bIpv6);
}
else
{
udp::server(oIoContext, nListenPort, bIpv6, strMulticastGroup, nMulticastResponsePort);
}
run_threaded(oIoContext, 1);
return 0;
}
catch (const asio::system_error& oError)
{
if (oError.code() == asio::error::bad_descriptor)
{
std::cout << "server closed" << std::endl;
return 0;
}
std::cerr << "exception: " << oError.what() << std::endl;
return 1;
}
}
