mirror of https://github.com/nostar/urfd.git
Merge 9cc9d2dd81 into bd0c114e43
Dave Behnke
1 month ago
committed by
GitHub
commit
a8016763ce
@ -0,0 +1,145 @@
|
||||
# NNG Event System Documentation
|
||||
|
||||
This document describes the real-time event system in `urfd`, which uses NNG (nanomsg next gen) to broadcast system state and activity as JSON.
|
||||
|
||||
## Architecture Overview
|
||||
|
||||
The `urfd` reflector acts as an NNG **Publisher** (PUB). Any number of subscribers (e.g., a middle-tier service or dashboard) can connect as **Subscribers** (SUB) to receive the event stream.
|
||||
|
||||
```mermaid
|
||||
graph TD
|
||||
subgraph "urfd Core"
|
||||
CR["CReflector"]
|
||||
CC["CClients"]
|
||||
CU["CUsers"]
|
||||
PS["CPacketStream"]
|
||||
end
|
||||
|
||||
subgraph "Publishing Layer"
|
||||
NP["g_NNGPublisher"]
|
||||
end
|
||||
|
||||
subgraph "Network"
|
||||
ADDR["tcp://0.0.0.0:5555"]
|
||||
end
|
||||
|
||||
subgraph "External"
|
||||
MT["Middle Tier / Dashboard"]
|
||||
end
|
||||
|
||||
%% Internal Flows
|
||||
CC -- "client_connect / client_disconnect" --> NP
|
||||
CU -- "hearing / closing" --> NP
|
||||
CR -- "periodic state report" --> NP
|
||||
PS -- "IsActive status" --> CR
|
||||
|
||||
%% Network Flow
|
||||
NP --> ADDR
|
||||
ADDR -.-> MT
|
||||
```
|
||||
|
||||
## Messaging Protocols
|
||||
|
||||
Events are sent as serialized JSON strings. Each message contains a `type` field to identify the payload structure.
|
||||
|
||||
### 1. State Broadcast (`state`)
|
||||
|
||||
Sent periodically based on `DashboardInterval` (default 10s). It provides a full snapshot of the reflector's configuration and status.
|
||||
|
||||
**Payload Structure:**
|
||||
|
||||
```json
|
||||
{
|
||||
"type": "state",
|
||||
"Configure": {
|
||||
"Key": "Value",
|
||||
...
|
||||
},
|
||||
"Peers": [
|
||||
{
|
||||
"Callsign": "XLX123",
|
||||
"Modules": "ABC",
|
||||
"Protocol": "D-Extra",
|
||||
"ConnectTime": "2023-10-27T10:00:00Z"
|
||||
}
|
||||
],
|
||||
"Clients": [
|
||||
{
|
||||
"Callsign": "N7TAE",
|
||||
"OnModule": "A",
|
||||
"Protocol": "DMR",
|
||||
"ConnectTime": "2023-10-27T10:05:00Z"
|
||||
}
|
||||
],
|
||||
"Users": [
|
||||
{
|
||||
"Callsign": "G4XYZ",
|
||||
"Repeater": "GB3NB",
|
||||
"OnModule": "B",
|
||||
"ViaPeer": "XLX456",
|
||||
"LastHeard": "2023-10-27T10:10:00Z"
|
||||
}
|
||||
],
|
||||
"ActiveTalkers": [
|
||||
{
|
||||
"Module": "A",
|
||||
"Callsign": "N7TAE"
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
### 2. Client Connectivity (`client_connect` / `client_disconnect`)
|
||||
|
||||
Triggered immediately when a client (Repeater, Hotspot, or Mobile App) links or unlinks from a module.
|
||||
|
||||
**Payload Structure:**
|
||||
|
||||
```json
|
||||
{
|
||||
"type": "client_connect",
|
||||
"callsign": "N7TAE",
|
||||
"ip": "1.2.3.4",
|
||||
"protocol": "DMR",
|
||||
"module": "A"
|
||||
}
|
||||
```
|
||||
|
||||
### 3. Voice Activity (`hearing`)
|
||||
|
||||
Triggered when the reflector "hears" an active transmission. This event is sent for every "tick" or heartbeat of voice activity processed by the reflector.
|
||||
|
||||
**Payload Structure:**
|
||||
|
||||
```json
|
||||
{
|
||||
"type": "hearing",
|
||||
"my": "G4XYZ",
|
||||
"ur": "CQCQCQ",
|
||||
"rpt1": "GB3NB",
|
||||
"rpt2": "XLX123 A",
|
||||
"module": "A",
|
||||
"protocol": "M17"
|
||||
}
|
||||
```
|
||||
|
||||
### 4. Transmission End (`closing`)
|
||||
|
||||
Triggered when a transmission stream is closed (user stops talking).
|
||||
|
||||
**Payload Structure:**
|
||||
|
||||
```json
|
||||
{
|
||||
"type": "closing",
|
||||
"my": "G4XYZ",
|
||||
"module": "A",
|
||||
"protocol": "M17"
|
||||
}
|
||||
```
|
||||
|
||||
## Middle Tier Design Considerations
|
||||
|
||||
1. **Late Joining**: The `state` message is broadcast periodically to ensure a middle-tier connecting at any time (or reconnecting) can synchronize its internal state without waiting for new events.
|
||||
2. **Active Talkers**: The `ActiveTalkers` array in the `state` message identifies who is currently keyed up. Real-time transitions (start/stop) are driven by the `hearing` events and the absence of such events over a timeout (typically 2-3 seconds).
|
||||
3. **Deduplication**: The `state` report is a snapshot. If the middle-tier is already tracking events, it can use the `state` report to "re-base" its state and clear out stale data.
|
||||
|
After Width: | Height: | Size: 40 KiB |
@ -0,0 +1,66 @@
|
||||
#include "NNGPublisher.h"
|
||||
#include "Global.h"
|
||||
#include <iostream>
|
||||
|
||||
CNNGPublisher::CNNGPublisher()
|
||||
: m_started(false)
|
||||
{
|
||||
m_sock.id = 0;
|
||||
}
|
||||
|
||||
CNNGPublisher::~CNNGPublisher()
|
||||
{
|
||||
Stop();
|
||||
}
|
||||
|
||||
bool CNNGPublisher::Start(const std::string &addr)
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
if (m_started) return true;
|
||||
|
||||
int rv;
|
||||
if ((rv = nng_pub0_open(&m_sock)) != 0) {
|
||||
std::cerr << "NNG: Failed to open pub socket: " << nng_strerror(rv) << std::endl;
|
||||
return false;
|
||||
}
|
||||
|
||||
if ((rv = nng_listen(m_sock, addr.c_str(), nullptr, 0)) != 0) {
|
||||
std::cerr << "NNG: Failed to listen on " << addr << ": " << nng_strerror(rv) << std::endl;
|
||||
nng_close(m_sock);
|
||||
return false;
|
||||
}
|
||||
|
||||
m_started = true;
|
||||
std::cout << "NNG: Publisher started at " << addr << std::endl;
|
||||
return true;
|
||||
}
|
||||
|
||||
void CNNGPublisher::Stop()
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
if (!m_started) return;
|
||||
|
||||
nng_close(m_sock);
|
||||
m_started = false;
|
||||
std::cout << "NNG: Publisher stopped" << std::endl;
|
||||
}
|
||||
|
||||
void CNNGPublisher::Publish(const nlohmann::json &event)
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
if (!m_started) return;
|
||||
|
||||
if (m_sock.id == 0) {
|
||||
std::cerr << "NNG debug: Cannot publish, socket not initialized." << std::endl;
|
||||
return;
|
||||
}
|
||||
std::string msg = event.dump();
|
||||
if (g_Configure.GetBoolean(g_Keys.dashboard.debug))
|
||||
std::cout << "NNG debug: Attempting to publish message of size " << msg.size() << ": " << msg << std::endl;
|
||||
int rv = nng_send(m_sock, (void *)msg.c_str(), msg.size(), NNG_FLAG_NONBLOCK);
|
||||
if (rv == 0) {
|
||||
std::cout << "NNG: Published event: " << event["type"] << std::endl;
|
||||
} else if (rv != NNG_EAGAIN) {
|
||||
std::cerr << "NNG: Send error: " << nng_strerror(rv) << std::endl;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,24 @@
|
||||
#pragma once
|
||||
|
||||
#include <string>
|
||||
#include <mutex>
|
||||
#include <nlohmann/json.hpp>
|
||||
#include <nng/nng.h>
|
||||
#include <nng/protocol/pubsub0/pub.h>
|
||||
|
||||
class CNNGPublisher
|
||||
{
|
||||
public:
|
||||
CNNGPublisher();
|
||||
~CNNGPublisher();
|
||||
|
||||
bool Start(const std::string &addr);
|
||||
void Stop();
|
||||
|
||||
void Publish(const nlohmann::json &event);
|
||||
|
||||
private:
|
||||
nng_socket m_sock;
|
||||
std::mutex m_mutex;
|
||||
bool m_started;
|
||||
};
|
||||
Loading…
Reference in new issue