1 Commits
5 changed files with 66 additions and 3 deletions

No files matched your search

+31 -1
View File
@@ -246,6 +246,32 @@ DebugServiceI *svc = DebugServiceI::GetInstance(); // broker wrapper retrieves i
5. Applies quantization (`QuantizedType`, `RangeMin`, `RangeMax`) per signal
6. Supports unicast and multicast UDP delivery (configurable per instance)
### TCP variant
`TCPStreamer` / `TCPStreamerClient` subclass the UDP DataSources and replace the transport
(`TCPSServer` / `TCPSClient`) behind the `UDPSServerI` / `UDPSClientI` seam
(`UDPSTransportI.h`). Framing is unchanged (`totalFragments` is always 1); each client has
a bounded queue, so a slow client drops only its own frames.
`TCPStreamService` (`Source/Components/Interfaces/TCPStreamService/`) is an optional
in-process server: one acceptor plus `Workers` I/O threads serve many `TCPSServiceStream`s
(one per `TCPStreamer` with `Service=`/`StreamName=`). Producers only copy frames into
per-connection queues and wake the worker via a self-pipe. Lock order: topology, stream,
connection. A client picks streams with a `SUBSCRIBE` frame (type 5):
- **Multiplexing**: the payload may be a comma separated list; each server frame carries the
stream's list index in `fragmentIdx`; queues are accounted per stream. `TCPSClient`
demultiplexes into per-stream listeners (`GetStream(i)` facades), and StreamHub creates
one session per stream over one shared client.
- **Joining**: `TCPStreamJoin` taps several streams (`TCPSStreamTapI`), watches the service
for (un)registration (`TCPSStreamObserverI`) and publishes a merged Strict stream: CONFIG =
concatenated descriptors named `<stream>_<signal>`; one DATA frame per master frame with
the other sources held at their last value. No extra thread. Accumulate sources and
differing HRT frequencies are refused. Lock order: source tap, join, output stream,
connection queue.
See `Docs/TCPStreamer.md`.
### Packed Signals (Accumulate mode)
When a signal has `NumberOfElements > 1` and `SamplingRate` is configured:
@@ -387,9 +413,13 @@ Hub = {
}
```
A TCP source sets `Transport = "tcp"` and, for a `TCPStreamService`, `StreamName = "<stream>"`;
a comma list (`"a,b"`) takes several streams over one connection and creates one session per
stream (label `<Label>_<stream>`, id `<node>_<stream>`).
Sources added at runtime (`addSource`) get generated ids `s1, s2, …`;
`saveSources` persists them to `SourcesFile` (JSON array of
`{label, addr, multicastGroup?, dataPort?}`), reloaded at start-up.
`{label, addr, multicastGroup?, dataPort?, transport?, streamName?}`), reloaded at start-up.
`AllowedOrigins` is a comma/space-separated allowlist of `scheme://host[:port]`
values accepted in the WebSocket `Origin` header (max 8 entries, 128 chars
+5
View File
@@ -63,6 +63,11 @@ Full streaming pipeline: MARTe2 (FileReader -> IOGAM -> UDPStreamer) -> StreamHu
(live, zoom, window, trigger). The validator compares the recording against the analytic
ground truth (fidelity, sine shape fit, continuity) and optionally a fed-reference tap.
Chain scenarios `s60`-`s73` use TCP transports (`network = "tcp"` or `"tcp_service"`). The
optional `hub` field adds `{"mode": "mux"}` (the hub takes all sources over one multiplexed
connection, `s72`) or `{"mode": "join"}` (a `TCPStreamJoin` merges the sources and the hub
subscribes to the merge, `s73`).
### direct
MARTe2 FileReader -> UDPStreamer -> UDPStreamerClient -> FileWriter round-trip. Validates that
+4
View File
@@ -33,6 +33,9 @@ Reply (unicast): `{"type":"pong"}`.
- `addr` — `host:port` of the UDPStreamer control port.
- `multicastGroup` / `dataPort` — optional; if present the hub joins the
multicast group for DATA and keeps a TCP control connection for CONNECT/CONFIG.
- `transport` (`"udp"` default, or `"tcp"`) and `streamName` — for a TCP source, the
`TCPStreamService` stream to subscribe to. A comma list (`"a,b"`) multiplexes several
streams over one connection and adds one source per stream, labelled `<label>_<stream>`.
- The hub assigns ids `s1`, `s2`, … and broadcasts an updated `sources` event.
### `removeSource`
@@ -421,6 +424,7 @@ warning.
[
{"label": "wave", "addr": "127.0.0.1:44500"},
{"label": "mc", "addr": "127.0.0.1:44501", "multicastGroup": "239.0.0.1", "dataPort": 44502},
{"label": "svc", "addr": "127.0.0.1:44500", "transport": "tcp", "streamName": "fast,slow"},
{"source": "wave", "signal": "Adc", "scale": 0.00030518, "offset": -1.25, "unit": "V"}
]
```
+4
View File
@@ -58,6 +58,10 @@ Useful flags: `-p <port>` WS port, `-n <pts>` ring size, `-s` skip rebuild,
./Client/streamhub/build/StreamHubClient -host <hub-host> -port 8090
```
A source can use TCP instead of UDP: add `Transport = "tcp"` (and `StreamName = "<stream>"`
for a `TCPStreamService`). `StreamName = "a,b"` takes several streams over one connection,
shown as separate sources `<Label>_a`, `<Label>_b`. See `Docs/TCPStreamer.md`.
Sources can also be added at runtime from either client ("Add source": label +
`host:port`, optional multicast group/data port). "Save sources" persists the
runtime list to `streamhub_sources.json` so the hub reloads it on restart.
+22 -2
View File
@@ -12,6 +12,7 @@ This repository integrates two complementary capabilities:
| Capability | Component | Purpose |
| --------------------- | ------------------------ | ------------------------------------------------------------------------------------------------------------- |
| **Signal streaming** | `UDPStreamer` DataSource | Continuously stream selected signals to a browser-based oscilloscope over UDP |
| **Signal streaming (TCP)** | `TCPStreamer` / `TCPStreamerClient` DataSources, `TCPStreamService`, `TCPStreamJoin` | Same UDPS stream over reliable TCP; one in-process server for many streams; multiplexed connections and merged streams |
| **Signal debugging** | `DebugService` Interface | On-demand signal tracing, value forcing, and conditional breakpoints — zero application code changes required |
| **Sine generation** | `SineArrayGAM` | Generate continuous sine-wave arrays for testing and simulation |
| **Time stamping** | `TimeArrayGAM` | Provide time-reference arrays aligned to an RT cycle |
@@ -30,8 +31,11 @@ MARTe_Integrated_components/
│ └── Client/go/ Go packages: udpsprotocol, wshub
├── Source/Components/
│ ├── DataSources/UDPStreamer/ Real-time UDP signal streaming DataSource
│ ├── DataSources/TCPStreamer/ TCP variant of UDPStreamer (plus TCPStreamerClient)
│ ├── GAMs/SineArrayGAM/ Sine-wave array generator GAM
│ ├── GAMs/TimeArrayGAM/ Time-reference array GAM
│ ├── Interfaces/UDPStream/ UDPS server/client transports (UDP and TCP)
│ ├── Interfaces/TCPStreamService/ In-process multi-stream TCP server + TCPStreamJoin
│ ├── Interfaces/DebugService/ Signal tracing/forcing/breakpoint Interface
│ └── Interfaces/TCPLogger/ TCP log-forwarding LoggerConsumerI
├── Test/
@@ -59,6 +63,21 @@ sending a `CONNECT` packet; the server then sends `CONFIG` (signal metadata) and
See `Docs/UDPStreamer.md` and `Docs/Protocol.md`.
### TCPStreamer / TCPStreamService / TCPStreamJoin
`TCPStreamer` and `TCPStreamerClient` are the TCP counterparts of the UDP DataSources: same
UDPS framing, per-client bounded queues (a slow client only drops its own frames).
`TCPStreamService` is an optional single in-process server (acceptor + worker threads)
shared by many `TCPStreamer`s; clients pick a stream by name with a `SUBSCRIBE` frame.
- **Multiplexing**: a `SUBSCRIBE` may list several streams (`"fast,slow"`) to carry them over
one socket, each with its own CONFIG and rate. StreamHub does this for a comma separated
`StreamName`.
- **Joining**: `TCPStreamJoin` publishes a new stream merging several streams of the service
(`<stream>_<signal>`, sample-and-hold, paced by a master source).
See `Docs/TCPStreamer.md`.
### SineArrayGAM
Generates a continuous float32 sine-wave array every RT cycle. Used as a signal
@@ -119,7 +138,7 @@ clients on a configurable port. Works as a sidecar to `DebugService`.
### StreamHub Application
Headless C++ hub (`Source/Applications/StreamHub/`) that aggregates multiple
UDPStreamer sources and serves them to oscilloscope clients over WebSocket
UDPStreamer / TCPStreamer sources (`Transport = "tcp"`, optional `StreamName`) and serves them to oscilloscope clients over WebSocket
(port 8090): ring buffers with wall-clock time calibration, LTTB decimation,
hub-side trigger engine, per-window zoom. Clients: browser SPA
(`Client/webui` + `Client/udpstreamer/static`) and native ImGui desktop client
@@ -132,7 +151,7 @@ See `Docs/StreamHub-UserGuide.md`, `Docs/StreamHub-API.md` and
### UDPS Protocol
The `Common/UDP/UDPSProtocol.h` header defines the shared binary wire format used by
both `UDPStreamer` and `DebugService`. It is intentionally free of MARTe2-specific
`UDPStreamer`, `TCPStreamer` and `DebugService`. It is intentionally free of MARTe2-specific
dependencies so it can also be used by Go clients (via `Common/Client/go/udpsprotocol`).
See `Docs/Protocol.md`.
@@ -244,6 +263,7 @@ Open `http://localhost:9090`, explore the object tree, trace signals, force valu
| ----------------------------- | -------------------------------------------------------------- |
| `Docs/Protocol.md` | UDPS binary wire protocol specification |
| `Docs/UDPStreamer.md` | UDPStreamer DataSource configuration reference |
| `Docs/TCPStreamer.md` | TCPStreamer, TCPStreamService, multiplexing and TCPStreamJoin |
| `Docs/UDPS-C-Client.md` | Standalone C/C++ UDPS receiver library (`Common/Client/c`) |
| `Docs/SineArrayGAM.md` | SineArrayGAM configuration reference |
| `Docs/DebugService.md` | DebugService TCP API and architecture |