Distinguish relay-only stream subscriptions in member profiles
This commit is contained in:
1 parent
2cc9834e4c
commit
8883002e3f
2 files changed
+21
-4
No files matched your search
@@ -3,6 +3,7 @@
|
|||||||
|
|
||||||
#include "../../lib/debug_config.h"
|
#include "../../lib/debug_config.h"
|
||||||
#include "../../lib/json_flat.h"
|
#include "../../lib/json_flat.h"
|
||||||
|
#include "../../lib/mem.h"
|
||||||
#include "../utun_instance.h"
|
#include "../utun_instance.h"
|
||||||
#include "../routing_layer/topo_node_sqlite.h"
|
#include "../routing_layer/topo_node_sqlite.h"
|
||||||
#include "stream_block.h"
|
#include "stream_block.h"
|
||||||
@@ -70,9 +71,10 @@ static int member_option(const char *key, const char *value, void *arg) {
|
|||||||
const char *p = value;
|
const char *p = value;
|
||||||
while (*p) {
|
while (*p) {
|
||||||
if (parse->state.count == STREAM_MEMBER_MAX_SUBSCRIPTIONS) return -1;
|
if (parse->state.count == STREAM_MEMBER_MAX_SUBSCRIPTIONS) return -1;
|
||||||
struct stream_subscription_key entry;
|
struct stream_subscription_key entry = {0};
|
||||||
if (member_hex(&p, &entry.source_node_id) != STREAM_OK || *p++ != ':'
|
if (member_hex(&p, &entry.source_node_id) != STREAM_OK || *p++ != ':'
|
||||||
|| member_hex(&p, &entry.source_epoch) != STREAM_OK) return -1;
|
|| member_hex(&p, &entry.source_epoch) != STREAM_OK) return -1;
|
||||||
|
if (*p == ':') { p++; if (*p++ != 'r') return -1; entry.relay_only = 1; }
|
||||||
if (parse->state.count) {
|
if (parse->state.count) {
|
||||||
const struct stream_subscription_key *previous = &parse->state.subscriptions[parse->state.count - 1];
|
const struct stream_subscription_key *previous = &parse->state.subscriptions[parse->state.count - 1];
|
||||||
if (entry.source_node_id < previous->source_node_id
|
if (entry.source_node_id < previous->source_node_id
|
||||||
@@ -132,8 +134,9 @@ int stream_member_format(const struct stream_member_state *state, char source[MS
|
|||||||
&& entry->source_epoch <= state->subscriptions[i - 1].source_epoch)))) {
|
&& entry->source_epoch <= state->subscriptions[i - 1].source_epoch)))) {
|
||||||
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member format invalid subscription index=%zu", i); return STREAM_ERR_ARGUMENT;
|
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member format invalid subscription index=%zu", i); return STREAM_ERR_ARGUMENT;
|
||||||
}
|
}
|
||||||
int length = snprintf(encoded_subscriptions + used, sizeof(encoded_subscriptions) - used, "%s%016llx:%016llx",
|
int length = snprintf(encoded_subscriptions + used, sizeof(encoded_subscriptions) - used, "%s%016llx:%016llx%s",
|
||||||
i ? "," : "", (unsigned long long)entry->source_node_id, (unsigned long long)entry->source_epoch);
|
i ? "," : "", (unsigned long long)entry->source_node_id, (unsigned long long)entry->source_epoch,
|
||||||
|
entry->relay_only ? ":r" : "");
|
||||||
if (length < 0 || (size_t)length >= sizeof(encoded_subscriptions) - used) {
|
if (length < 0 || (size_t)length >= sizeof(encoded_subscriptions) - used) {
|
||||||
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member subscriptions exceed limit count=%zu", state->count); return STREAM_ERR_ARGUMENT;
|
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "member subscriptions exceed limit count=%zu", state->count); return STREAM_ERR_ARGUMENT;
|
||||||
}
|
}
|
||||||
@@ -217,3 +220,14 @@ int stream_member_online(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t
|
|||||||
struct TOPO_GROUP_NODE *node = topo_node_find_by_id(group, member_id);
|
struct TOPO_GROUP_NODE *node = topo_node_find_by_id(group, member_id);
|
||||||
return node && topo_node_best_path(node) != NULL;
|
return node && topo_node_best_path(node) != NULL;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* Роль относится к каналу, не к устройству или глобальному media_delivery. */
|
||||||
|
int stream_member_supernode(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t member_id) {
|
||||||
|
struct TOPO_GROUP *group = inst && inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
|
||||||
|
if (!group || group->stopping || group->group_type != TOPO_GROUP_TYPE_CHAT) return 0;
|
||||||
|
struct topo_connect_peer *peers = NULL; size_t count = 0;
|
||||||
|
if (topo_node_sqlite_get_connect_peers(inst->topo_sqlite_db, group->channel_id, &peers, &count) < 0) return -1;
|
||||||
|
int supernode = 0;
|
||||||
|
for (size_t i = 0; i < count; i++) if (peers[i].node_id == member_id) { supernode = peers[i].supernode; break; }
|
||||||
|
u_free(peers); return supernode;
|
||||||
|
}
|
||||||
@@ -1,6 +1,6 @@
|
|||||||
/* Строковые опции подписанного блока мембера, без отдельного протокола объявлений.
|
/* Строковые опции подписанного блока мембера, без отдельного протокола объявлений.
|
||||||
* stream_source = epoch(16 hex):n:r:RS-size; отсутствие/пустое значение — не публикует.
|
* stream_source = epoch(16 hex):n:r:RS-size; отсутствие/пустое значение — не публикует.
|
||||||
* stream_subscriptions = source(16 hex):epoch(16 hex),...; сортировка по source/epoch.
|
* stream_subscriptions = source(16 hex):epoch(16 hex)[:r],...; :r — только помощь, без локального зрителя.
|
||||||
* Указанная в stream_source собственная трансляция также должна быть в subscriptions.
|
* Указанная в stream_source собственная трансляция также должна быть в subscriptions.
|
||||||
* load проверяет подпись профиля и принадлежность ключа источника, BGP-доступность — отдельно.
|
* load проверяет подпись профиля и принадлежность ключа источника, BGP-доступность — отдельно.
|
||||||
* parse/format не меняют выход при ошибке. Все вызовы — один uasync-поток. */
|
* parse/format не меняют выход при ошибке. Все вызовы — один uasync-поток. */
|
||||||
@@ -23,6 +23,7 @@ struct UTUN_INSTANCE;
|
|||||||
struct stream_subscription_key {
|
struct stream_subscription_key {
|
||||||
uint64_t source_node_id;
|
uint64_t source_node_id;
|
||||||
uint64_t source_epoch;
|
uint64_t source_epoch;
|
||||||
|
int relay_only;
|
||||||
};
|
};
|
||||||
|
|
||||||
struct stream_member_state {
|
struct stream_member_state {
|
||||||
@@ -43,6 +44,8 @@ int stream_member_has_subscription(const struct stream_member_state *state, uint
|
|||||||
int stream_member_load(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t member_id,
|
int stream_member_load(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t member_id,
|
||||||
struct stream_member_state *state, uint8_t public_key[SC_PUBKEY_SIZE]);
|
struct stream_member_state *state, uint8_t public_key[SC_PUBKEY_SIZE]);
|
||||||
int stream_member_online(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t member_id);
|
int stream_member_online(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t member_id);
|
||||||
|
/* Канальная роль из принятого блока владельца; -1 — ошибка чтения, 0/1 — отсутствие/наличие роли. */
|
||||||
|
int stream_member_supernode(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t member_id);
|
||||||
|
|
||||||
#ifdef __cplusplus
|
#ifdef __cplusplus
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user