Martino FerrariandClaude Opus 4.6
50d85363a3
Add TCPStreamer/TCPStreamerClient (TCP transport behind the UDPSServerI/UDPSClientI seam), an optional single in-process TCPStreamService serving many streams over worker threads, multiplexed SUBSCRIBE (comma list, stream index in fragmentIdx), and TCPStreamJoin, which merges several streams into one sample-and-hold stream paced by a master. StreamHub takes a comma list of streams over one connection. Includes GTests, E2E scenarios s58-s73 and docs. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
140 lines
4.9 KiB
C++
140 lines
4.9 KiB
C++
#ifndef TCPSTREAMJOIN_H_
|
|
#define TCPSTREAMJOIN_H_
|
|
|
|
/**
|
|
* @file TCPStreamJoin.h
|
|
* @brief Publishes one new stream of a TCPStreamService that merges several other
|
|
* streams of the same service.
|
|
*
|
|
* @code
|
|
* +Svc = { Class = TCPStreamService Port = 44500 }
|
|
* +Join = { Class = TCPStreamJoin Service = "Svc" StreamName = "all"
|
|
* Sources = "fast,slow" Master = "fast" }
|
|
* @endcode
|
|
* Declare the join after the service. Consumers subscribe to "all" like to any stream.
|
|
*
|
|
* The merged stream is emitted at the pace of the Master source: each Master DATA frame
|
|
* produces one merged frame holding the Master's signals and the last values received
|
|
* from every other source (sample-and-hold). Nothing is emitted until every source has
|
|
* delivered one DATA frame. No thread is added: the merged frame is built and queued on
|
|
* the Master producer's thread.
|
|
*
|
|
* @par CONFIG
|
|
* Concatenation of the source descriptors in Sources order, named
|
|
* "<stream>_<signal>" (at most 63 characters), time-signal indexes rebased, publish mode
|
|
* Strict. All sources must share the same HRT frequency (an unknown one matches any).
|
|
*
|
|
* @par DATA
|
|
* [master timestamp][source0 body][source1 body]... where a body is that source's last
|
|
* DATA payload without its 8-byte timestamp. A held value is therefore stamped with the
|
|
* Master's time unless the source carries its own time signal.
|
|
*
|
|
* @par Limits
|
|
* Only Strict and Decimate sources are accepted (Accumulate batches have varying
|
|
* length); a source whose frame size changes without a new CONFIG is ignored; the merged
|
|
* frame must fit MaxPayloadSize. The output stream is registered while every source is
|
|
* registered and removed as soon as one leaves; it is rebuilt when a source re-registers
|
|
* or changes its CONFIG.
|
|
*
|
|
* @par Parameters
|
|
* | Parameter | Type | Default | Description |
|
|
* |----------------|--------|-------------|-------------|
|
|
* | Service | string | (required) | Name of the TCPStreamService object. |
|
|
* | StreamName | string | object name | Name of the merged stream. |
|
|
* | Sources | string | (required) | Comma separated source stream names (1..16). |
|
|
* | Master | string | first source| Source whose frames pace the merged stream. |
|
|
* | MaxPayloadSize | uint32 | 60000 | Largest merged frame payload. |
|
|
*
|
|
* @par Locking
|
|
* Producer-side: source tapMutex, then the join mutex, then the output stream mutex, then
|
|
* the connection queue mutex. Service notifications run outside its topology lock and
|
|
* are serialised.
|
|
*/
|
|
|
|
#include "FastPollingMutexSem.h"
|
|
#include "Object.h"
|
|
#include "ReferenceT.h"
|
|
#include "StreamString.h"
|
|
#include "StructuredDataI.h"
|
|
#include "TCPSServiceStream.h"
|
|
#include "TCPSStreamTapI.h"
|
|
#include "TCPStreamService.h"
|
|
|
|
namespace MARTe {
|
|
|
|
class TCPStreamJoin : public Object, public TCPSStreamTapI, public TCPSStreamObserverI {
|
|
public:
|
|
CLASS_REGISTER_DECLARATION()
|
|
|
|
/** Most source streams one join can merge. */
|
|
static const uint32 TCPSJ_MAX_SOURCES = 16u;
|
|
|
|
TCPStreamJoin();
|
|
virtual ~TCPStreamJoin();
|
|
|
|
virtual bool Initialise(StructuredDataI &data);
|
|
|
|
virtual void OnTapConfig(TCPSServiceStream *source, const uint8 *payload, uint32 payloadSize);
|
|
virtual void OnTapData(TCPSServiceStream *source, uint32 counter, const uint8 *payload,
|
|
uint32 payloadSize);
|
|
virtual void OnStreamRegistered(TCPSServiceStream *stream);
|
|
virtual void OnStreamUnregistered(TCPSServiceStream *stream);
|
|
|
|
/** @return true while the merged stream is registered with the service. */
|
|
bool IsActive() const;
|
|
/** @return true once a merged CONFIG has been published for the current sources. */
|
|
bool HasConfig() const;
|
|
/** @return merged DATA frames emitted. */
|
|
uint64 GetFramesEmitted() const;
|
|
uint32 GetNumSources() const;
|
|
|
|
private:
|
|
struct Source {
|
|
Source();
|
|
~Source();
|
|
StreamString name;
|
|
TCPSServiceStream *stream; ///< non-NULL while registered and tapped
|
|
bool configOk;
|
|
uint8 *cfg;
|
|
uint32 cfgBytes;
|
|
uint32 numSigs;
|
|
uint64 hrt;
|
|
uint8 *body;
|
|
uint32 bodyCap;
|
|
uint32 bodySize;
|
|
bool haveBody;
|
|
};
|
|
|
|
int32 FindByName(const char8 *name) const;
|
|
int32 FindByStream(const TCPSServiceStream *s) const;
|
|
bool AllRegistered() const;
|
|
void Evaluate();
|
|
void PublishConfigLocked();
|
|
bool ParseConfig(Source &s, const uint8 *payload, uint32 n);
|
|
|
|
ReferenceT<TCPStreamService> service;
|
|
TCPSServiceStream *out;
|
|
bool observing;
|
|
StreamString outName;
|
|
Source *src;
|
|
uint32 numSrc;
|
|
uint32 master;
|
|
uint32 maxPayload;
|
|
|
|
FastPollingMutexSem mtx;
|
|
bool active;
|
|
bool configPublished;
|
|
bool sizeLogged;
|
|
uint32 outCounter;
|
|
volatile uint64 emitted;
|
|
uint8 *scratch;
|
|
uint32 scratchCap;
|
|
|
|
TCPStreamJoin(const TCPStreamJoin &);
|
|
TCPStreamJoin &operator=(const TCPStreamJoin &);
|
|
};
|
|
|
|
} // namespace MARTe
|
|
|
|
#endif // TCPSTREAMJOIN_H_
|