You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

411 lines
22 KiB

#include <pthread.h>
#include <string.h>
#include "../../../lib/debug_config.h"
#include "../../../lib/mem.h"
#include "../../../lib/u_async.h"
#include "../../../src/utun_instance.h"
#include "../../../src/chat/chat_core.h"
#include "../../../src/stream/stream.h"
#include "../../../src/stream/stream_member.h"
#include "../../../src/stream/stream_media.h"
#include "../../../src/stream/stream_network.h"
#include "../../../src/stream/stream_publisher.h"
#include "stream_bridge.h"
#define RX_LIMIT (4u * 1024u * 1024u)
#define RX_RECORDS 256
struct media_record {
struct media_record *next;
uint64_t id, pts, frame_id, created;
size_t length, offset;
int key;
uint8_t data[];
};
struct media_queue { struct media_record *head, *tail; size_t bytes; unsigned count; };
struct android_media_session {
uint64_t id, group, source, epoch, expected_epoch, next_frame, last_pts[2], drain_until;
struct STREAM_HANDLE *handle;
struct stream_media_format format;
int opening, stopping, finished, need_key, seen[2], rx_overflow, reason;
int want_read, want_video, waiting, resync, listen_audio;
int64_t position, audio_ahead;
};
static pthread_mutex_t media_lock = PTHREAD_MUTEX_INITIALIZER;
static struct UTUN_INSTANCE *media_inst;
static struct posted_task *media_post;
static void *media_timer;
static struct stream_publisher *transmitter;
static struct android_media_session publisher, viewer;
static struct media_queue output_queue;
static struct stream_media_status viewer_status;
static uint64_t viewer_status_id;
static uint64_t watch_group;
static int watch_pending;
static void media_process(void *arg);
static void media_close(struct android_media_session *s, int kind);
int android_stream_tx_snapshot(uint64_t id, struct android_stream_tx_stats *out) {
if (!out) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "TX stats output missing"); return -1; }
pthread_mutex_lock(&media_lock);
struct stream_publisher_status status;
int result = stream_publisher_snapshot(transmitter, id, &status);
if (!result) *out = (struct android_stream_tx_stats){status.id, status.sample_us, status.video_bytes, status.audio_bytes,
status.stream_bytes, status.transport_bytes, status.inflight_bytes, status.acked_1s, status.total_bytes, status.connections};
pthread_mutex_unlock(&media_lock); return result;
}
/* Очереди обмена pthread: LL_QUEUE принадлежит uasync и сюда не подходит. */
static void queue_append(struct media_queue *q, struct media_record *record) {
record->next = NULL;
if (q->tail) q->tail->next = record; else q->head = record;
q->tail = record; q->bytes += record->length; q->count++;
}
static struct media_record *queue_take(struct media_queue *q) {
struct media_record *record = q->head;
if (!record) return NULL;
q->head = record->next; if (!q->head) q->tail = NULL;
q->bytes -= record->length; q->count--; record->next = NULL;
return record;
}
static void queue_clear(struct media_queue *q) {
struct media_record *record;
while ((record = queue_take(q))) u_free(record);
}
/* RX pressure удаляет media/GAP, сохраняя подтверждения открытия/закрытия owners. */
static void media_drop_output(void) {
struct media_record **link = &output_queue.head, *previous = NULL;
while (*link) {
struct media_record *r = *link;
if (r->data[0] <= 3 || (r->id != publisher.id && r->id != viewer.id)) {
*link = r->next; if (output_queue.tail == r) output_queue.tail = previous;
output_queue.bytes -= r->length; output_queue.count--; u_free(r);
} else { previous = r; link = &r->next; }
}
}
/* Все обращения к ua, включая post, защищены тем же mutex, что и detach перед destroy. */
static int media_schedule(void) {
if (media_post || media_timer) return 0;
if (!media_inst || !uasync_is_running(media_inst->ua)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream schedule: core detached/stopped"); return -1;
}
struct posted_task *task = u_calloc(1, sizeof(*task));
if (!task) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream post allocation failed"); return -1; }
task->callback = media_process;
media_post = task; uasync_post_reserved(media_inst->ua, task);
return 0;
}
/* Выдача владеет копией: lifetime scratch stream callback не выходит в Kotlin. */
static void media_output(uint64_t id, int kind, const struct stream_media_format *format, uint64_t pts,
int flags, const uint8_t *data, size_t size) {
if (!id || size > STREAM_MEDIA_MAX_FRAME) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android output invalid id/size=%zu", size); return; }
if (output_queue.count >= RX_RECORDS || size + ANDROID_STREAM_HEADER_SIZE > RX_LIMIT - output_queue.bytes) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android stream RX full id=%llu bytes=%zu count=%u",
(unsigned long long)id, output_queue.bytes, output_queue.count);
viewer.rx_overflow = 1;
media_drop_output();
if (kind == 1 || kind == 2) { media_output(id, 3, NULL, 0, 0, NULL, 0); return; }
}
struct media_record *record = u_calloc(1, sizeof(*record) + size + ANDROID_STREAM_HEADER_SIZE);
if (!record) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream RX allocation failed bytes=%zu", size); viewer.rx_overflow = 1; return;
}
record->id = id; record->length = size + ANDROID_STREAM_HEADER_SIZE; record->created = get_time_us();
uint8_t *p = record->data; p[0] = (uint8_t)kind; p[1] = (uint8_t)flags;
if (format) {
stream_write_be(p + 2, format->width, 2); stream_write_be(p + 4, format->height, 2);
stream_write_be(p + 6, format->rotation, 2); stream_write_be(p + 20, format->revision, 4);
p[24] = format->video_codec;
}
stream_write_be(p + 8, pts, 8); stream_write_be(p + 16, size, 4);
if (size) memcpy(p + ANDROID_STREAM_HEADER_SIZE, data, size);
queue_append(&output_queue, record);
}
/* Callback receiver работает под media_lock, декодирование видео/аудио выполняет внешний worker. */
static void media_frame(void *arg, const struct stream_media_format *format,
const struct stream_media_video *video, const struct stream_media_audio *audio) {
struct android_media_session *s = arg;
if (s->rx_overflow) return;
if (video) media_output(s->id, 1, format, video->pts_us, video->flags, video->data, video->length);
if (audio) media_output(s->id, 2, format, audio->pts_us, 0, audio->data, audio->length);
}
/* Состояния потока не передаются через chat SharedFlow; здесь только владельцы выбранного viewer. */
static void media_event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
(void)arg;
pthread_mutex_lock(&media_lock);
if (inst == media_inst && viewer.handle && event->descriptor.group_id == viewer.group
&& event->descriptor.source_node_id == viewer.source && event->descriptor.source_epoch == viewer.epoch
&& !viewer.stopping && (event->type != STREAM_EVENT_CLOSED || !stream_handle_active(viewer.handle))) {
if (event->type == STREAM_EVENT_CLOSED) {
viewer.reason = event->reason;
media_close(&viewer, 6);
} else if (event->type == STREAM_EVENT_AVAILABLE) {
viewer.waiting = 0;
if (media_schedule() < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media ready scheduling failed");
}
}
pthread_mutex_unlock(&media_lock);
}
/* Publisher держит конечное окно выдачи после FIN; abrupt detach ничего не обещает получателям. */
static void media_close(struct android_media_session *s, int kind) {
if (kind) media_output(s->id, kind, NULL, 0, kind == 6 ? s->reason : 0, NULL, 0);
if (s->handle) stream_close(s->handle);
memset(s, 0, sizeof(*s));
}
/* Один actor обслуживает команды и один ответ на запрос; idle не копирует поток из кольца. */
static void media_process(void *arg) {
(void)arg;
pthread_mutex_lock(&media_lock);
media_post = NULL; media_timer = NULL;
if (!media_inst) { pthread_mutex_unlock(&media_lock); return; }
if (watch_pending) {
watch_pending = 0;
if (stream_watch_sources(media_inst, watch_group) != STREAM_OK)
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source watch failed group=%llx", (unsigned long long)watch_group);
}
struct android_media_session *sessions[] = {&viewer};
for (unsigned i = 0; i < 1; i++) {
struct android_media_session *s = sessions[i];
if (s->opening) {
if (s->stopping) { media_close(s, 5); continue; }
s->opening = 0;
int result = STREAM_OK;
s->reason = STREAM_CLOSE_FAILED;
if (s->source && s->expected_epoch) {
struct stream_member_state state;
result = stream_member_load(media_inst, s->group, s->source, &state, NULL);
if (result == STREAM_OK && (!state.published || state.source.source_epoch != s->expected_epoch)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android selected source changed group=%llx source=%llx expected=%llx actual=%llx",
(unsigned long long)s->group, (unsigned long long)s->source, (unsigned long long)s->expected_epoch,
(unsigned long long)(state.published ? state.source.source_epoch : 0));
result = STREAM_ERR_CONFLICT; s->reason = STREAM_CLOSE_SOURCE_CHANGED;
}
}
if (result == STREAM_OK) result = s->source ? stream_subscribe_open(media_inst, s->group, s->source, &s->handle)
: stream_publish_open(media_inst, s->group, NULL, &s->handle);
if (result == STREAM_OK) result = stream_media_enable(s->handle,
s->source ? STREAM_MEDIA_TRACK_VIDEO | (s->listen_audio ? STREAM_MEDIA_TRACK_AUDIO : 0) : 0);
struct stream_info info;
if (result == STREAM_OK) {
result = stream_get_info(s->handle, &info);
if (result == STREAM_OK) {
s->epoch = info.descriptor.source_epoch;
if (s->expected_epoch && s->epoch != s->expected_epoch) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android viewer opened stale epoch id=%llu", (unsigned long long)s->id);
result = STREAM_ERR_CONFLICT; s->reason = STREAM_CLOSE_SOURCE_CHANGED;
}
}
}
if (result != STREAM_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open failed id=%llu group=%llx source=%llx result=%d",
(unsigned long long)s->id, (unsigned long long)s->group, (unsigned long long)s->source, result);
media_close(s, 6);
} else {
if (s->source) { viewer_status_id = s->id; memset(&viewer_status, 0, sizeof(viewer_status)); }
media_output(s->id, 4, NULL, 0, 0, NULL, 0);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream opened id=%llu publisher=%d group=%llx source=%llx",
(unsigned long long)s->id, !s->source, (unsigned long long)s->group, (unsigned long long)s->source);
}
}
}
if (viewer.stopping && viewer.id) media_close(&viewer, 5);
if (viewer.handle && (viewer.resync || viewer.rx_overflow)) {
viewer.resync = viewer.rx_overflow = 0; viewer.position = -1; viewer.waiting = 0;
if (stream_media_resync(viewer.handle) != STREAM_OK) media_close(&viewer, 6);
}
if (viewer.handle) {
int result = stream_media_get_status(viewer.handle, &viewer_status);
if (result < 0) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media index failed result=%d", result); media_close(&viewer, 6); }
}
unsigned queued_viewer = 0;
for (struct media_record *r = output_queue.head; r; r = r->next) if (r->id == viewer.id) queued_viewer++;
if (viewer.handle && viewer.want_read && !queued_viewer
&& output_queue.bytes + STREAM_MEDIA_MAX_FRAME + ANDROID_STREAM_HEADER_SIZE <= RX_LIMIT) {
viewer.want_read = 0;
struct stream_media_packet packet;
int result = stream_media_read(viewer.handle, viewer.position, viewer.want_video, viewer.audio_ahead, &packet);
if (stream_media_get_status(viewer.handle, &viewer_status) < 0) result = STREAM_ERR_FORMAT;
if (result == STREAM_OK) {
if (packet.kind == STREAM_MEDIA_READ_END) media_close(&viewer, 5);
else media_frame(&viewer, &packet.format, packet.kind == STREAM_MEDIA_READ_VIDEO ? &packet.video : NULL,
packet.kind == STREAM_MEDIA_READ_AUDIO ? &packet.audio : NULL);
} else if (result == STREAM_MORE) {
viewer.want_read = 1;
if (media_schedule() < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media assembly scheduling failed");
} else if (result == STREAM_WAIT) viewer.want_read = viewer.waiting = 1;
else if (result == STREAM_GAP || result == STREAM_EXPIRED) {
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android playback resync id=%llu result=%d", (unsigned long long)viewer.id, result);
stream_media_resync(viewer.handle); viewer.position = -1; media_output(viewer.id, 3, NULL, 0, 0, NULL, 0);
memset(&viewer_status, 0, sizeof(viewer_status));
} else { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android playback read failed result=%d", result); media_close(&viewer, 6); }
}
pthread_mutex_unlock(&media_lock);
}
/* Attach/detach — обязательная граница lifetime ядра. */
int android_stream_attach(struct UTUN_INSTANCE *inst) {
pthread_mutex_lock(&media_lock);
if (!inst || !inst->stream || (media_inst && media_inst != inst)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream attach rejected inst=%p owner=%p", (void *)inst, (void *)media_inst);
pthread_mutex_unlock(&media_lock); return -1;
}
if (!transmitter && stream_publisher_create(&transmitter, inst) < 0) {
pthread_mutex_unlock(&media_lock); return -1;
}
media_inst = inst; stream_set_event_callback(inst, media_event, NULL);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream bridge attached");
pthread_mutex_unlock(&media_lock);
return 0;
}
void android_stream_detach(struct UTUN_INSTANCE *inst) {
pthread_mutex_lock(&media_lock);
if (media_inst == inst) {
if (media_post) uasync_cancel_post(inst->ua, media_post);
if (media_timer) uasync_cancel_timeout(inst->ua, media_timer);
media_post = NULL; media_timer = NULL;
stream_watch_sources(inst, 0);
stream_set_event_callback(inst, NULL, NULL);
queue_clear(&output_queue);
stream_publisher_destroy(transmitter); transmitter = NULL; memset(&publisher, 0, sizeof(publisher));
if (viewer.id) media_close(&viewer, 0);
media_inst = NULL;
watch_pending = 0; watch_group = 0;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream bridge detached; producers rejected");
}
pthread_mutex_unlock(&media_lock);
}
/* Команда содержит идентификаторы, а не borrowed pointer ядра. */
int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height,
unsigned rotation, int audio, uint64_t expected_epoch, unsigned codec) {
if (!id || !group || (source && !expected_epoch)
|| (!source && (!width || width > 4096 || !height || height > 4096 || rotation > 270 || rotation % 90))) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open invalid id=%llu group=%llx source=%llx epoch=%llx dimensions=%ux%u",
(unsigned long long)id, (unsigned long long)group, (unsigned long long)source,
(unsigned long long)expected_epoch, width, height); return -1;
}
pthread_mutex_lock(&media_lock);
struct android_media_session *s = source ? &viewer : &publisher;
int result = -1;
if (codec < STREAM_MEDIA_H264 || codec > STREAM_MEDIA_H266 || s->id || id == publisher.id || id == viewer.id
|| !media_inst || !uasync_is_running(media_inst->ua)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open busy/detached id=%llu", (unsigned long long)id);
} else if ((!source || media_schedule() == 0)) {
*s = (struct android_media_session){.id = id, .group = group, .source = source, .expected_epoch = expected_epoch,
.opening = 1, .need_key = 1, .listen_audio = !!audio, .position = -1,
.audio_ahead = STREAM_MEDIA_READ_AHEAD_US,
.format = {(uint16_t)width, (uint16_t)height, (uint16_t)rotation, (uint8_t)codec, audio ? STREAM_MEDIA_OPUS_STEREO : 0, 1}};
result = source ? 0 : stream_publisher_open(transmitter, id, group, &s->format);
if (result < 0) memset(s, 0, sizeof(*s));
}
pthread_mutex_unlock(&media_lock); return result;
}
/* Последняя выбранная группа заменяет ещё не обработанную команду, указатели ядра не покидают mutex. */
int android_stream_watch(uint64_t group) {
pthread_mutex_lock(&media_lock);
int result = media_inst && uasync_is_running(media_inst->ua) ? media_schedule() : -1;
if (result == 0) { watch_group = group; watch_pending = 1; }
else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android stream watch rejected group=%llx", (unsigned long long)group);
pthread_mutex_unlock(&media_lock); return result;
}
/* Перегрузка удаляет целые видео-кадры и блокирует P-кадры до нового KEY. */
int android_stream_feed(uint64_t id, int video, uint64_t pts, int key, const uint8_t *data, size_t length,
const struct stream_media_format *format) {
pthread_mutex_lock(&media_lock);
int result = stream_publisher_feed(transmitter, id, video, pts, key, data, length, format);
pthread_mutex_unlock(&media_lock); return result;
}
/* Stop идемпотентен; publisher дренирует принятую очередь, viewer закрывается сразу в actor. */
void android_stream_stop(uint64_t id) {
pthread_mutex_lock(&media_lock);
struct android_media_session *s = publisher.id == id ? &publisher : viewer.id == id ? &viewer : NULL;
if (s == &publisher) { s->stopping = 1; stream_publisher_stop(transmitter, id); }
else if (s) { s->stopping = 1; if (media_schedule() < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream stop schedule failed"); }
pthread_mutex_unlock(&media_lock);
}
/* Выбрать только события текущего owner; старые owner-события имеют ограниченный TTL. */
uint8_t *android_stream_pull(uint64_t id, size_t *length) {
if (!length) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream pull missing length"); return NULL; }
*length = 0; pthread_mutex_lock(&media_lock);
if (publisher.id) {
struct stream_publisher_status status;
if (stream_publisher_snapshot(transmitter, publisher.id, &status) == 0) {
if (publisher.opening && status.state >= STREAM_PUBLISHER_LIVE && status.state <= STREAM_PUBLISHER_DRAINING) {
publisher.opening = 0; publisher.epoch = status.epoch; media_output(publisher.id, 4, NULL, 0, 0, NULL, 0);
}
if (status.state == STREAM_PUBLISHER_ENDED || status.state == STREAM_PUBLISHER_FAILED) {
media_output(publisher.id, status.state == STREAM_PUBLISHER_ENDED ? 5 : 6, NULL, 0, status.reason, NULL, 0);
memset(&publisher, 0, sizeof(publisher));
}
}
}
struct media_record **link = &output_queue.head, *previous = NULL;
uint8_t *data = NULL; uint64_t now = get_time_us();
while (*link) {
struct media_record *r = *link;
if (r->id == id || now - r->created > 5000000) {
*link = r->next; if (output_queue.tail == r) output_queue.tail = previous;
output_queue.bytes -= r->length; output_queue.count--;
if (r->id == id) {
data = u_malloc((uint32_t)r->length);
if (data) { memcpy(data, r->data, r->length); *length = r->length; }
else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream pull allocation failed");
u_free(r); break;
}
u_free(r);
} else { previous = r; link = &r->next; }
}
pthread_mutex_unlock(&media_lock); return data;
}
int android_stream_read_request(uint64_t id, int64_t position, int want_video, int64_t audio_ahead) {
if ((want_video != 0 && want_video != 1) || audio_ahead < -1 || audio_ahead > 60000000) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media demand invalid video=%d audio_ahead=%lld",
want_video, (long long)audio_ahead); return -1;
}
pthread_mutex_lock(&media_lock);
int result = 0;
if (viewer.id == id) {
int queued = 0;
for (struct media_record *r = output_queue.head; r; r = r->next) if (r->id == id) queued = 1;
if (!queued) {
int changed = viewer.position != position || viewer.want_video != want_video || viewer.audio_ahead != audio_ahead;
viewer.position = position; viewer.want_video = want_video; viewer.audio_ahead = audio_ahead;
viewer.want_read = want_video || audio_ahead >= 0;
if (viewer.want_read && (!viewer.waiting || changed)) { viewer.waiting = 0; result = media_schedule(); }
}
}
pthread_mutex_unlock(&media_lock); return result;
}
int android_stream_get_status(uint64_t id, struct stream_media_status *status) {
if (!status) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media status: missing output"); return -1; }
pthread_mutex_lock(&media_lock);
int result = viewer_status_id == id ? 0 : -1;
if (!result) *status = viewer_status;
pthread_mutex_unlock(&media_lock); return result;
}
int android_stream_resync(uint64_t id) {
pthread_mutex_lock(&media_lock);
int result = -1;
if (viewer.id == id && viewer.handle) {
media_drop_output(); viewer.resync = viewer.want_read = 1; viewer.waiting = 0; viewer.position = -1;
memset(&viewer_status, 0, sizeof(viewer_status));
result = media_schedule();
}
pthread_mutex_unlock(&media_lock); return result;
}