Ferrite.GroupCallMedia/MediasoupGroupCallBroadcastPlane.cs
Rama main · réplica de lectura del repositorio original.
// SPDX-License-Identifier: AGPL-3.0-or-later
// Copyright (C) 2022-2026 Aykut Alparslan KOC
using System.Globalization;
using System.Net;
using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;
using Ferrite.Services.Calls;
namespace Ferrite.GroupCallMedia;
public sealed class MediasoupGroupCallBroadcastPlane : IGroupCallBroadcastPlane
{
private readonly HttpClient _httpClient;
private readonly GroupCallMediaWorkerOptions _workerOptions;
private readonly GroupCallBroadcastOptions _broadcastOptions;
public MediasoupGroupCallBroadcastPlane(HttpClient httpClient,
GroupCallMediaWorkerOptions workerOptions,
GroupCallBroadcastOptions broadcastOptions)
{
workerOptions.Validate();
broadcastOptions.Validate();
_httpClient = httpClient;
_workerOptions = workerOptions;
_broadcastOptions = broadcastOptions;
}
public Task StartAsync(CancellationToken cancellationToken = default) =>
Task.CompletedTask;
public Task StopAsync(CancellationToken cancellationToken = default) =>
Task.CompletedTask;
public async ValueTask CreateStreamAsync(long callId, bool rtmpStream,
CancellationToken cancellationToken = default)
{
byte[] body = JsonSerializer.SerializeToUtf8Bytes(new { rtmpStream });
using var request = Build(HttpMethod.Put, StreamPath(callId), body);
using var response = await SendAsync(request, cancellationToken);
await EnsureSuccessAsync(response, cancellationToken);
}
public async ValueTask<bool> EndStreamAsync(long callId,
CancellationToken cancellationToken = default)
{
using var request = Build(HttpMethod.Delete, StreamPath(callId));
using var response = await SendAsync(request, cancellationToken);
await EnsureSuccessAsync(response, cancellationToken);
using JsonDocument document = await ReadJsonAsync(response,
cancellationToken);
return ReadBool(document.RootElement, "ended");
}
public async ValueTask<GroupCallBroadcastCredentials> GetCredentialsAsync(
long callId, bool revoke,
CancellationToken cancellationToken = default)
{
byte[] body = JsonSerializer.SerializeToUtf8Bytes(new { revoke });
using var request = Build(HttpMethod.Post,
$"{StreamPath(callId)}/credentials", body);
using var response = await SendAsync(request, cancellationToken);
await EnsureSuccessAsync(response, cancellationToken);
using JsonDocument document = await ReadJsonAsync(response,
cancellationToken);
JsonElement root = document.RootElement;
return new GroupCallBroadcastCredentials(ReadString(root, "url"),
ReadString(root, "key"), ReadInt32(root, "generation"));
}
public async ValueTask<IReadOnlyList<GroupCallBroadcastChannel>>
GetChannelsAsync(long callId,
CancellationToken cancellationToken = default)
{
using var request = Build(HttpMethod.Get,
$"{StreamPath(callId)}/channels");
using var response = await SendAsync(request, cancellationToken);
await EnsureSuccessAsync(response, cancellationToken);
using JsonDocument document = await ReadJsonAsync(response,
cancellationToken);
if (!document.RootElement.TryGetProperty("channels", out var channels) ||
channels.ValueKind != JsonValueKind.Array)
{
throw InvalidResponse("broadcast channels are missing");
}
var result = new List<GroupCallBroadcastChannel>();
foreach (JsonElement channel in channels.EnumerateArray())
{
result.Add(new GroupCallBroadcastChannel(ReadInt32(channel, "channel"),
ReadInt32(channel, "scale"),
ReadInt64(channel, "lastTimestampMs")));
}
return result;
}
public async ValueTask<ReadOnlyMemory<byte>> ReadSegmentAsync(
GroupCallBroadcastSegmentRequest request,
CancellationToken cancellationToken = default)
{
string query = string.Create(CultureInfo.InvariantCulture,
$"timestamp={request.TimestampMs}&scale={request.Scale}" +
$"&channel={request.Channel}&quality={request.VideoQuality}");
using var httpRequest = Build(HttpMethod.Get,
$"{StreamPath(request.CallId)}/segments?{query}");
using var response = await SendAsync(httpRequest, cancellationToken);
await EnsureSuccessAsync(response, cancellationToken);
long? contentLength = response.Content.Headers.ContentLength;
if (contentLength is <= 0 ||
contentLength > _broadcastOptions.MaxSegmentBytes)
{
throw InvalidResponse("broadcast segment length violates the bound");
}
byte[] result = await response.Content.ReadAsByteArrayAsync(cancellationToken);
if (result.Length == 0 || result.Length > _broadcastOptions.MaxSegmentBytes)
{
throw InvalidResponse("broadcast segment body violates the bound");
}
return result;
}
public async ValueTask<GroupCallBroadcastHealth> HealthAsync(
CancellationToken cancellationToken = default)
{
using var request = Build(HttpMethod.Get, "health");
using var response = await SendAsync(request, cancellationToken,
_workerOptions.HealthTimeout);
await EnsureSuccessAsync(response, cancellationToken);
using JsonDocument document = await ReadJsonAsync(response,
cancellationToken);
if (!document.RootElement.TryGetProperty("broadcast", out var broadcast) ||
broadcast.ValueKind != JsonValueKind.Object)
{
throw InvalidResponse("broadcast health is missing");
}
return new GroupCallBroadcastHealth(ReadBool(broadcast, "healthy"),
ReadInt32(broadcast, "streams"),
ReadInt32(broadcast, "liveStreams"),
ReadInt32(broadcast, "segments"),
ReadInt64(broadcast, "bytes"),
broadcast.TryGetProperty("ffmpegVersion", out var ffmpeg) &&
ffmpeg.ValueKind == JsonValueKind.String
? ffmpeg.GetString()
: null);
}
private HttpRequestMessage Build(HttpMethod method, string relativePath,
byte[]? body = null)
{
var request = new HttpRequestMessage(method,
new Uri(_workerOptions.ControlUrl, relativePath));
request.Headers.Authorization = new AuthenticationHeaderValue("Bearer",
_workerOptions.AuthSecret);
request.Headers.TryAddWithoutValidation(GroupCallMediaProtocol.ProtocolHeader,
_workerOptions.ProtocolVersion);
if (body != null)
{
request.Content = new ByteArrayContent(body);
request.Content.Headers.ContentType = new MediaTypeHeaderValue(
"application/json") { CharSet = "utf-8" };
}
return request;
}
private async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request,
CancellationToken cancellationToken, TimeSpan? timeout = null)
{
using var linked = CancellationTokenSource.CreateLinkedTokenSource(
cancellationToken);
linked.CancelAfter(timeout ?? _workerOptions.RequestTimeout);
HttpResponseMessage? response = null;
try
{
response = await _httpClient.SendAsync(request,
HttpCompletionOption.ResponseHeadersRead, linked.Token);
await response.Content.LoadIntoBufferAsync(
_broadcastOptions.MaxSegmentBytes, linked.Token);
return response;
}
catch (OperationCanceledException e) when (!cancellationToken.IsCancellationRequested)
{
response?.Dispose();
throw new GroupCallBroadcastException(
GroupCallBroadcastFailureKind.Unavailable,
"broadcast worker request timed out", e);
}
catch (Exception e) when (e is HttpRequestException or IOException or
InvalidOperationException)
{
response?.Dispose();
throw new GroupCallBroadcastException(
GroupCallBroadcastFailureKind.Unavailable,
"broadcast worker response is unavailable or exceeds its bound", e);
}
catch
{
response?.Dispose();
throw;
}
}
private static async Task EnsureSuccessAsync(HttpResponseMessage response,
CancellationToken cancellationToken)
{
if (response.IsSuccessStatusCode)
{
return;
}
string message;
try
{
using JsonDocument error = await ReadJsonAsync(response,
cancellationToken);
message = error.RootElement.TryGetProperty("error", out var value) &&
value.ValueKind == JsonValueKind.String
? value.GetString() ?? "broadcast worker rejected the request"
: "broadcast worker rejected the request";
}
catch (GroupCallBroadcastException e) when (
e.Kind == GroupCallBroadcastFailureKind.Unavailable)
{
message = "broadcast worker rejected the request";
}
GroupCallBroadcastFailureKind kind = (int)response.StatusCode == 425
? GroupCallBroadcastFailureKind.NotReady
: response.StatusCode switch
{
HttpStatusCode.NotFound => GroupCallBroadcastFailureKind.Expired,
HttpStatusCode.BadRequest => GroupCallBroadcastFailureKind.Unsupported,
HttpStatusCode.ServiceUnavailable =>
GroupCallBroadcastFailureKind.Unavailable,
_ => GroupCallBroadcastFailureKind.Rejected
};
throw new GroupCallBroadcastException(kind, message);
}
private static async Task<JsonDocument> ReadJsonAsync(
HttpResponseMessage response, CancellationToken cancellationToken)
{
try
{
await using Stream stream = await response.Content
.ReadAsStreamAsync(cancellationToken);
return await JsonDocument.ParseAsync(stream,
cancellationToken: cancellationToken);
}
catch (Exception e) when (e is JsonException or IOException)
{
throw InvalidResponse("body is not valid bounded JSON");
}
}
private static string StreamPath(long callId)
{
if (callId <= 0)
{
throw new ArgumentOutOfRangeException(nameof(callId));
}
return $"broadcast/{callId}";
}
private static bool ReadBool(JsonElement root, string name) =>
root.TryGetProperty(name, out var value) &&
value.ValueKind is JsonValueKind.True or JsonValueKind.False
? value.GetBoolean()
: throw InvalidResponse($"{name} must be a bool");
private static string ReadString(JsonElement root, string name) =>
root.TryGetProperty(name, out var value) &&
value.ValueKind == JsonValueKind.String &&
!string.IsNullOrEmpty(value.GetString())
? value.GetString()!
: throw InvalidResponse($"{name} must be a non-empty string");
private static int ReadInt32(JsonElement root, string name) =>
root.TryGetProperty(name, out var value) && value.TryGetInt32(out int result)
? result
: throw InvalidResponse($"{name} must be an int32");
private static long ReadInt64(JsonElement root, string name) =>
root.TryGetProperty(name, out var value) && value.TryGetInt64(out long result)
? result
: throw InvalidResponse($"{name} must be an int64");
private static GroupCallBroadcastException InvalidResponse(string message) =>
new(GroupCallBroadcastFailureKind.Unavailable,
$"invalid broadcast worker response: {message}");
}