jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Expose verified stream sources and channel watch events

master
evgeny 7 days ago
parent
90d3206071
commit
f91894aa1b
3 changed files with 124 additions and 1 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 1
      src/chat/chat_event.h
  2. 118
      src/stream/stream.c
  3. 6
      src/stream/stream.h

1
src/chat/chat_event.h
Unescape View File

@ -79,6 +79,7 @@ extern "C" {
#define CHAT_EVT_RADIO_MUTE_RESULT 55 /* [group_id:8LE][node_id:8LE][success:1] — локальный переключатель рации */
#define CHAT_EVT_MSG_SUBMITTED 56 /* [request_id:8LE][success:1] — результат локального сохранения сообщения */
#define CHAT_EVT_CPU_LOAD 57 /* [valid:1][average_percent:1] — CPU потока uasync, среднее за секунду, 0..100 */
#define CHAT_EVT_STREAM_SOURCES 58 /* [group_id:8BE][valid:1][JSON:utf8] — доступные публикации выбранного канала */
typedef void (*chat_event_handler_fn)(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len);

118
src/stream/stream.c
Unescape View File

@ -1,12 +1,16 @@
#include <stdio.h>
#include <string.h>
#include <limits.h>
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "../../lib/platform_compat.h"
#include "../../lib/u_async.h"
#include "../../lib/strbuf.h"
#include "../../lib/sqlite3.h"
#include "../utun_instance.h"
#include "../chat/chat_core.h"
#include "../chat/chat_event.h"
#include "stream_member.h"
#include "stream_source.h"
#include "stream_priv.h"
@ -50,12 +54,112 @@ struct stream_service {
stream_event_fn callback;
void *arg;
void *reconcile;
void *sources_pending;
uint64_t sources_group;
char *sources_snapshot;
int sources_valid;
int stopping, dispatching, registered, initializing;
};
static void session_deliver(void *arg);
static void service_reconcile(void *arg);
/* Общий каталог не выделяет media-ресурсов: профиль проверяется тем же кодом, что и subscribe. */
int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out) {
if (!inst || !out || *out || !group_id) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list invalid arguments"); return STREAM_ERR_ARGUMENT;
}
struct TOPO_GROUP *group = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
if (!group || group->stopping || group->group_type != TOPO_GROUP_TYPE_CHAT) return STREAM_WAIT;
for (const char *p = group->channel_id; *p; p++) if (*p < '0' || *p > '9') {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source namespace invalid group=%llx", (unsigned long long)group_id);
return STREAM_ERR_STATE;
}
sqlite3 *db = chat_core_get_db(inst);
if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list missing DB"); return STREAM_ERR_STATE; }
char query[320];
snprintf(query, sizeof(query), "SELECT p.node_id,COALESCE(NULLIF(p.local_nick,''),n.name,'') "
"FROM \"peers_%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id ORDER BY p.node_id", group->channel_id);
struct strbuf json = strbuf_new();
sqlite3_stmt *statement = NULL;
int result = STREAM_OK, count = 0;
if (strbuf_addf(&json, "[") < 0) result = STREAM_ERR_MEMORY;
if (result == STREAM_OK && sqlite3_prepare_v2(db, query, -1, &statement, NULL) != SQLITE_OK) result = STREAM_ERR_STATE;
int step = SQLITE_DONE;
while (result == STREAM_OK && (step = sqlite3_step(statement)) == SQLITE_ROW) {
uint64_t node = (uint64_t)sqlite3_column_int64(statement, 0);
if (!stream_member_online(inst, group_id, node)) continue;
struct stream_member_state state;
int loaded = stream_member_load(inst, group_id, node, &state, NULL);
if (loaded == STREAM_WAIT || loaded == STREAM_ERR_NOT_MEMBER) continue;
if (loaded < 0) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "source excluded group=%llx node=%llx profile_result=%d",
(unsigned long long)group_id, (unsigned long long)node, loaded); continue;
}
if (!state.published) continue;
if (strbuf_addf(&json, "%s{\"id\":\"%llu\",\"epoch\":\"%016llx\",\"self\":%d,\"name\":", count ? "," : "",
(unsigned long long)node, (unsigned long long)state.source.source_epoch, node == inst->node_id) < 0
|| strbuf_add_json_string(&json, (const char *)sqlite3_column_text(statement, 1)) < 0
|| strbuf_addf(&json, "}") < 0) result = STREAM_ERR_MEMORY;
count++;
}
if (result == STREAM_OK && step != SQLITE_DONE) result = STREAM_ERR_STATE;
sqlite3_finalize(statement);
if (result == STREAM_OK && strbuf_addf(&json, "]") < 0) result = STREAM_ERR_MEMORY;
if (result == STREAM_OK) *out = strbuf_detach(&json);
strbuf_free(&json);
if (result != STREAM_OK) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list failed group=%llx result=%d DB=%s",
(unsigned long long)group_id, result, sqlite3_errmsg(db));
else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source list group=%llx count=%d", (unsigned long long)group_id, count);
return result;
}
/* Один отложенный снимок объединяет Merkle/BGP-события; неизменный список повторно не отправляется. */
static void sources_deliver(void *arg) {
struct stream_service *service = arg;
service->sources_pending = NULL;
if (service->stopping || !service->sources_group) return;
char *json = NULL;
int result = stream_sources_json(service->inst, service->sources_group, &json);
int valid = result == STREAM_OK || result == STREAM_WAIT;
const char *list = json ? json : "[]";
if (service->sources_snapshot && valid == service->sources_valid && !strcmp(service->sources_snapshot, list)) {
u_free(json); return;
}
size_t length = strlen(list);
uint8_t *event = length <= INT_MAX - 9 ? u_malloc((uint32_t)(9 + length)) : NULL;
if (!event) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source event allocation failed bytes=%zu", length); u_free(json); return;
}
stream_write_be(event, service->sources_group, 8); event[8] = (uint8_t)valid;
memcpy(event + 9, list, length);
char *snapshot = json ? json : u_strdup(list);
if (!snapshot) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source snapshot allocation failed");
u_free(service->sources_snapshot); service->sources_snapshot = snapshot; service->sources_valid = valid;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream sources changed group=%llx valid=%d bytes=%zu",
(unsigned long long)service->sources_group, valid, length);
chat_event_post(service->inst, CHAT_EVT_STREAM_SOURCES, event, (int)(length + 9));
u_free(event);
}
/* Наблюдение принадлежит сервису и снимается до освобождения ядра. */
int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id) {
struct stream_service *service = inst ? inst->stream : NULL;
if (!service || service->stopping) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream source watch without service group=%llx", (unsigned long long)group_id);
return STREAM_ERR_STATE;
}
service->sources_group = group_id;
u_free(service->sources_snapshot); service->sources_snapshot = NULL;
if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending);
service->sources_pending = group_id ? uasync_call_soon(inst->ua, service, sources_deliver) : NULL;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source watch group=%llx", (unsigned long long)group_id);
if (group_id && !service->sources_pending) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source watch scheduling failed"); return STREAM_ERR_MEMORY;
}
return STREAM_OK;
}
/* Observer выполняет только учёт; пользовательские callbacks и отправка остаются отложенными. */
static void cache_event(void *arg, const uint8_t *wire, size_t length, int added) {
struct stream_session *s = arg;
@ -268,7 +372,13 @@ static void service_reconcile(void *arg) {
static void member_changed(uint64_t node_id, const char *tags, const char *channel, void *arg) {
(void)tags;
struct stream_service *service = arg;
if (service->stopping || !service->sessions) return;
if (service->stopping) return;
struct TOPO_GROUP *watched = service->sources_group ? stream_group(service->inst, service->sources_group) : NULL;
if (watched && channel && !strcmp(watched->channel_id, channel) && !service->sources_pending) {
service->sources_pending = uasync_call_soon(service->inst->ua, service, sources_deliver);
if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source refresh scheduling failed");
}
if (!service->sessions) return;
stream_network_membership_changed(service->network, node_id, channel);
for (struct stream_session *s = service->sessions; s; s = s->next) {
int own = node_id == service->inst->node_id;
@ -344,6 +454,8 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) {
stream_network_destroy(service->network); service->network = NULL;
if (service->registered) member_sync_remove_props_cbk(inst, member_changed, service);
if (service->reconcile) uasync_call_soon_cancel(inst->ua, service->reconcile);
if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending);
u_free(service->sources_snapshot);
while (service->sessions) {
struct stream_session *s = service->sessions;
if (!s->closing) session_close(s, STREAM_CLOSE_SERVICE, 1, 0);
@ -360,6 +472,10 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) {
void stream_cancel_group(struct UTUN_INSTANCE *inst, uint64_t group_id) {
struct stream_service *service = inst ? inst->stream : NULL;
if (!service) return;
if (service->sources_group == group_id && !service->sources_pending) {
service->sources_pending = uasync_call_soon(inst->ua, service, sources_deliver);
if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream cancelled group source refresh failed");
}
struct stream_session *s = service->sessions;
while (s) {
struct stream_session *next = s->next;

6
src/stream/stream.h
Unescape View File

@ -60,6 +60,12 @@ int stream_service_init(struct UTUN_INSTANCE *inst);
void stream_service_destroy(struct UTUN_INSTANCE *inst);
void stream_cancel_group(struct UTUN_INSTANCE *inst, uint64_t group_id);
void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callback, void *arg);
/* Проверенные online-публикации: JSON с id(decimal string), epoch(hex string), name, self.
* Только uasync; *out заранее NULL, результат освобождается u_free(). WAIT — группа недоступна. */
int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out);
/* Один выбранный канал frontend: начальный снимок и изменения через CHAT_EVT_STREAM_SOURCES.
* group_id=0 снимает наблюдение. Не создаёт подписку/буфер и не удерживает ACTIVE. Только uasync. */
int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id);
/* Локальный owner удерживает эффективный ACTIVE; желаемый frontend-режим ядро восстанавливает после последнего close. */
int stream_service_needs_activity(const struct UTUN_INSTANCE *inst);

Write Preview
Loading…
Cancel
Save