Cover paused sources and duplicate relay controls during startup
This commit is contained in:
1 parent
8206e53b5c
commit
eda8dfbc91
3 files changed
+46
-7
No files matched your search
@@ -54,8 +54,8 @@ static int automatic_drop_relay;
|
||||
static int automatic_drop_tail;
|
||||
static etcp_recv_fn automatic_handlers[4];
|
||||
static uint8_t automatic_request[STREAM_WIRE_HEADER + 1];
|
||||
static etcp_recv_fn relay_raw_handlers[4], relay_route_handler;
|
||||
static unsigned relay_data[3], relay_ends[3], relay_requests_seen;
|
||||
static etcp_recv_fn relay_raw_handlers[4], relay_route_handler, relay_status_handler;
|
||||
static unsigned relay_data[3], relay_ends[3], relay_requests_seen, relay_statuses_seen;
|
||||
static int relay_drop_source;
|
||||
static uint8_t relay_request_wire[STREAM_WIRE_HEADER];
|
||||
static struct {
|
||||
@@ -1054,6 +1054,13 @@ static void relay_test_routed(struct ETCP_CONN *conn, struct ll_entry *entry) {
|
||||
relay_route_handler(conn, entry);
|
||||
}
|
||||
|
||||
/* Счётчик STATUS выявляет ответы на дубликаты, не меняя подписанное управление и его порядок. */
|
||||
static void relay_test_status(struct ETCP_CONN *conn, struct ll_entry *entry) {
|
||||
if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
|
||||
&& entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_STATUS) relay_statuses_seen++;
|
||||
relay_status_handler(conn, entry);
|
||||
}
|
||||
|
||||
/* Внутренний приём не вызывает GUI callback; публичный viewer на том же экземпляре вызывает его. */
|
||||
static void relay_test_event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
|
||||
(void)arg; unsigned index = 0; while (index < 3 && inst != nodes[index]) index++; CHECK(index < 3);
|
||||
@@ -1100,15 +1107,22 @@ static void test_supernode_relay(void) {
|
||||
}
|
||||
relay_route_handler = nodes[1]->router_bindings.callbacks[ETCP_RT_ID_STREAM];
|
||||
CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_test_routed) == 0);
|
||||
relay_status_handler = nodes[2]->router_bindings.callbacks[ETCP_RT_ID_STREAM];
|
||||
CHECK(etcp_router_bind(nodes[2], ETCP_RT_ID_STREAM, relay_test_status) == 0);
|
||||
struct stream_parameters parameters = {.n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE};
|
||||
struct STREAM_HANDLE *publisher = NULL, *viewers[3] = {0};
|
||||
CHECK(stream_publish_open(nodes[3], GROUP2, ¶meters, &publisher) == STREAM_OK);
|
||||
struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
|
||||
for (unsigned i = 0; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
|
||||
CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
|
||||
test_clock_run_for(ua, 10000);
|
||||
struct stream_local local;
|
||||
CHECK(!relay_requests_seen && !relay_data[2]);
|
||||
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
|
||||
stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 1000);
|
||||
CHECK(stream_publish_block(publisher, expected, sizeof(expected)) == STREAM_OK);
|
||||
CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
|
||||
run_until(&relay_data[2], 1); test_clock_run_for(ua, 60000);
|
||||
struct stream_local local;
|
||||
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT && !relay_requests_seen);
|
||||
stream_close(viewers[2]); viewers[2] = NULL;
|
||||
test_clock_run_for(ua, 10000);
|
||||
@@ -1149,6 +1163,11 @@ static void test_supernode_relay(void) {
|
||||
CHECK(relay_race.helper_subscribes && !relay_race.source_subscribes);
|
||||
relay_release_info(); test_clock_run_for(ua, 1000);
|
||||
CHECK(!relay_race.source_subscribes && relay_data[2] == 3);
|
||||
struct stream_message duplicate; CHECK(stream_wire_parse(relay_request_wire, sizeof(relay_request_wire), &duplicate) == STREAM_OK);
|
||||
CHECK(duplicate.receiver_epoch);
|
||||
unsigned statuses_before = relay_statuses_seen;
|
||||
for (unsigned i = 0; i < 3; i++) { relay_test_control(&duplicate); test_clock_run_for(ua, 300); }
|
||||
CHECK(relay_statuses_seen == statuses_before);
|
||||
relay_race.check_starts = 0;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] bootstrap helper wins: elapsed_ms=%llu, n=8 r=2, common_start=%llu, stale direct INFO ignored",
|
||||
(unsigned long long)connected_ms, (unsigned long long)relay_race.start);
|
||||
@@ -1236,6 +1255,7 @@ static void test_supernode_relay(void) {
|
||||
stream_close(viewers[2]); stream_close(publisher);
|
||||
relay_drop_source = 0; relay_role(1, 0, 101);
|
||||
CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_route_handler) == 0);
|
||||
CHECK(etcp_router_bind(nodes[2], ETCP_RT_ID_STREAM, relay_status_handler) == 0);
|
||||
for (unsigned i = 0; i < 4; i++) {
|
||||
CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_raw_handlers[i]) == 0);
|
||||
if (i < 3) stream_set_event_callback(nodes[i], i < 2 ? on_stream : NULL, NULL);
|
||||
|
||||
Reference in new issue
Block a user