1111
This commit is contained in:
@@ -0,0 +1,767 @@
|
||||
using System.Diagnostics;
|
||||
using LiveRecorder.Application.Abstractions.Logging;
|
||||
using LiveRecorder.Application.Abstractions.Notifications;
|
||||
using LiveRecorder.Application.Abstractions.Platforms;
|
||||
using LiveRecorder.Application.Abstractions.Settings;
|
||||
using LiveRecorder.Application.Services;
|
||||
using LiveRecorder.Domain.Entities;
|
||||
using LiveRecorder.Domain.Enums;
|
||||
using LiveRecorder.Infrastructure.Persistence;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace LiveRecorder.Infrastructure.Services;
|
||||
|
||||
public sealed partial class FfmpegService
|
||||
{
|
||||
private async Task HandleProcessOutputAsync(SessionProcessRuntime runtime, string? line, bool isError)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(line))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogDebug("ffmpeg[{SessionId}] {Line}", runtime.RecordSessionId, line);
|
||||
if (TryParseSegmentOpenPath(line, out var openedPath))
|
||||
{
|
||||
await HandleSegmentOpenedAsync(runtime, openedPath);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!runtime.HasOpenedFirstSegment && IsOptionCompatibilityFailureLine(line))
|
||||
{
|
||||
runtime.MarkStartupFailure(StartupFailureKind.InputOptionCompatibility, line);
|
||||
}
|
||||
else if (!runtime.HasOpenedFirstSegment && IsRetryableStartupFailureLine(line))
|
||||
{
|
||||
runtime.MarkStartupFailure(StartupFailureKind.StreamHandshake, line);
|
||||
}
|
||||
|
||||
if (line.Contains("error", StringComparison.OrdinalIgnoreCase) ||
|
||||
line.Contains("fail", StringComparison.OrdinalIgnoreCase) ||
|
||||
line.Contains("timed out", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
await PersistFfmpegLineAsync(runtime, line, isError ? SystemLogLevel.Error : SystemLogLevel.Warning);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task PersistFfmpegLineAsync(SessionProcessRuntime runtime, string line, SystemLogLevel level)
|
||||
{
|
||||
try
|
||||
{
|
||||
using var scope = _serviceScopeFactory.CreateScope();
|
||||
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
||||
await logService.WriteAsync(
|
||||
level,
|
||||
"FFmpeg",
|
||||
"ffmpeg reported a warning or error line.",
|
||||
line,
|
||||
runtime.LiveRoomId,
|
||||
runtime.RecordSessionId,
|
||||
runtime.CurrentTaskId);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Persist ffmpeg output failed for session {RecordSessionId}", runtime.RecordSessionId);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleSegmentOpenedAsync(SessionProcessRuntime runtime, string openedPath)
|
||||
{
|
||||
await runtime.Gate.WaitAsync();
|
||||
try
|
||||
{
|
||||
using var scope = _serviceScopeFactory.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
||||
var session = await dbContext.RecordSessions
|
||||
.Include(item => item.RecordTasks)
|
||||
.ThenInclude(item => item.Result)
|
||||
.FirstOrDefaultAsync(item => item.Id == runtime.RecordSessionId);
|
||||
|
||||
if (session is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
var currentTask = session.RecordTasks.FirstOrDefault(item => item.Id == runtime.CurrentTaskId);
|
||||
if (currentTask is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (runtime.SaveMode == RecordSaveMode.SingleFile)
|
||||
{
|
||||
runtime.HasOpenedFirstSegment = true;
|
||||
currentTask.AttachProcess(runtime.ProcessId, now);
|
||||
currentTask.MarkRunning(now);
|
||||
session.AttachProcess(runtime.ProcessId, now);
|
||||
session.MarkRunning(now);
|
||||
session.ActivateSegment(1, now);
|
||||
if (!runtime.HasInitializedDanmaku)
|
||||
{
|
||||
await EnsureDanmakuSegmentAsync(runtime, currentTask, now);
|
||||
runtime.HasInitializedDanmaku = true;
|
||||
}
|
||||
|
||||
await dbContext.SaveChangesAsync();
|
||||
return;
|
||||
}
|
||||
|
||||
var segmentIndex = string.Equals(openedPath, runtime.CurrentOutputFilePath, StringComparison.OrdinalIgnoreCase)
|
||||
? Math.Max(1, runtime.CurrentSegmentIndex)
|
||||
: Math.Max(1, runtime.CurrentSegmentIndex + 1);
|
||||
if (!runtime.HasInitializedDanmaku)
|
||||
{
|
||||
await EnsureDanmakuSegmentAsync(runtime, currentTask, now);
|
||||
runtime.HasInitializedDanmaku = true;
|
||||
}
|
||||
|
||||
if (segmentIndex == runtime.CurrentSegmentIndex)
|
||||
{
|
||||
runtime.HasOpenedFirstSegment = true;
|
||||
currentTask.MarkStarting(runtime.StreamUrl, openedPath, currentTask.StartedAt ?? now);
|
||||
currentTask.AttachProcess(runtime.ProcessId, now);
|
||||
currentTask.MarkRunning(now);
|
||||
session.AttachProcess(runtime.ProcessId, now);
|
||||
session.MarkRunning(now);
|
||||
session.ActivateSegment(segmentIndex, now);
|
||||
runtime.CurrentOutputFilePath = openedPath;
|
||||
await dbContext.SaveChangesAsync();
|
||||
return;
|
||||
}
|
||||
|
||||
var previousTask = currentTask;
|
||||
var previousTaskId = previousTask.Id;
|
||||
var previousOutputPath = runtime.CurrentOutputFilePath;
|
||||
|
||||
var newTask = new RecordTask(
|
||||
session.LiveRoomId,
|
||||
session.Id,
|
||||
segmentIndex,
|
||||
session.PreferredQuality,
|
||||
session.OutputFormat,
|
||||
now);
|
||||
newTask.MarkStarting(runtime.StreamUrl, openedPath, now);
|
||||
newTask.AttachProcess(runtime.ProcessId, now);
|
||||
newTask.MarkRunning(now);
|
||||
await dbContext.RecordTasks.AddAsync(newTask);
|
||||
|
||||
previousTask.MarkCompleted(
|
||||
now,
|
||||
previousTask.StartedAt.HasValue ? Math.Max(0, (now - previousTask.StartedAt.Value).TotalSeconds) : null);
|
||||
previousTask.DetachProcess(now);
|
||||
|
||||
session.AttachProcess(runtime.ProcessId, now);
|
||||
session.MarkRunning(now);
|
||||
session.ActivateSegment(segmentIndex, now);
|
||||
|
||||
await dbContext.SaveChangesAsync();
|
||||
|
||||
if (runtime.DanmakuRecorder is not null)
|
||||
{
|
||||
await runtime.DanmakuRecorder.StartSegmentAsync(newTask.Id, segmentIndex, openedPath, now);
|
||||
}
|
||||
|
||||
var previousDanmakuSummary = runtime.DanmakuRecorder?.TakeSummary(previousTaskId);
|
||||
UpsertRecordResult(
|
||||
previousTask,
|
||||
dbContext,
|
||||
previousOutputPath,
|
||||
CalculateFileSize(previousOutputPath),
|
||||
previousTask.DurationSeconds,
|
||||
previousDanmakuSummary?.FilePath,
|
||||
previousDanmakuSummary?.MessageCount ?? 0,
|
||||
now);
|
||||
|
||||
await dbContext.SaveChangesAsync();
|
||||
|
||||
runtime.CurrentTaskId = newTask.Id;
|
||||
runtime.CurrentSegmentIndex = segmentIndex;
|
||||
runtime.CurrentOutputFilePath = openedPath;
|
||||
runtime.HasOpenedFirstSegment = true;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Handle segment open failed for session {RecordSessionId}", runtime.RecordSessionId);
|
||||
}
|
||||
finally
|
||||
{
|
||||
runtime.Gate.Release();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleProcessExitedAsync(SessionProcessRuntime runtime, Process process)
|
||||
{
|
||||
_processes.TryRemove(runtime.RecordSessionId, out _);
|
||||
|
||||
try
|
||||
{
|
||||
runtime.DanmakuCancellation.Cancel();
|
||||
if (runtime.DanmakuPumpTask is not null)
|
||||
{
|
||||
try
|
||||
{
|
||||
await runtime.DanmakuPumpTask;
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Danmaku pump ended with error for session {RecordSessionId}", runtime.RecordSessionId);
|
||||
}
|
||||
}
|
||||
|
||||
SessionDanmakuXmlRecorder.DanmakuSegmentSummary? activeDanmakuSummary = null;
|
||||
if (runtime.DanmakuRecorder is not null)
|
||||
{
|
||||
activeDanmakuSummary = await runtime.DanmakuRecorder.CompleteActiveSegmentAsync();
|
||||
await runtime.DanmakuRecorder.DisposeAsync();
|
||||
}
|
||||
|
||||
if (runtime.DanmakuConnection is not null)
|
||||
{
|
||||
await runtime.DanmakuConnection.DisposeAsync();
|
||||
}
|
||||
|
||||
if (await TryRecoverStartupFailureAsync(runtime))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await FinalizeExitedSessionAsync(runtime, process, activeDanmakuSummary);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Handle ffmpeg exit failed for session {RecordSessionId}", runtime.RecordSessionId);
|
||||
}
|
||||
finally
|
||||
{
|
||||
runtime.ExitCompletion.TrySetResult(true);
|
||||
runtime.Dispose();
|
||||
process.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<bool> TryRecoverStartupFailureAsync(SessionProcessRuntime runtime)
|
||||
{
|
||||
if (runtime.HasOpenedFirstSegment ||
|
||||
runtime.StartupFailureKind == StartupFailureKind.None ||
|
||||
string.IsNullOrWhiteSpace(runtime.LastStartupFailureLine))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
using var scope = _serviceScopeFactory.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
||||
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
||||
|
||||
var session = await dbContext.RecordSessions
|
||||
.Include(item => item.LiveRoom)
|
||||
.Include(item => item.RecordTasks)
|
||||
.ThenInclude(item => item.Result)
|
||||
.FirstOrDefaultAsync(item => item.Id == runtime.RecordSessionId);
|
||||
|
||||
if (session?.LiveRoom is null || !IsActiveSessionStatus(session.Status))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
var currentTask = session.RecordTasks
|
||||
.OrderBy(item => item.SegmentIndex)
|
||||
.ThenBy(item => item.CreatedAt)
|
||||
.FirstOrDefault(item => item.Id == runtime.CurrentTaskId);
|
||||
|
||||
if (currentTask is null || !IsActiveTaskStatus(currentTask.Status))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
var observedAt = DateTimeOffset.UtcNow;
|
||||
if (runtime.StartupFailureKind == StartupFailureKind.InputOptionCompatibility)
|
||||
{
|
||||
if (runtime.HasRetriedWithCompatibilityProfile)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Warning,
|
||||
"FFmpeg",
|
||||
"Initial ffmpeg input profile was rejected. Retrying once with minimal compatibility options.",
|
||||
runtime.LastStartupFailureLine,
|
||||
session.LiveRoomId,
|
||||
session.Id,
|
||||
currentTask.Id);
|
||||
|
||||
var retryStream = new StreamUrlResult(
|
||||
runtime.SelectedQuality,
|
||||
runtime.SelectedProtocol,
|
||||
runtime.StreamUrl,
|
||||
runtime.InputHeaders,
|
||||
Array.Empty<StreamQualityOption>());
|
||||
|
||||
session.MarkStarting(retryStream.SelectedUrl, session.OutputPathPattern ?? runtime.OutputPathPattern, observedAt);
|
||||
session.ActivateSegment(Math.Max(1, runtime.CurrentSegmentIndex), observedAt);
|
||||
currentTask.MarkStarting(retryStream.SelectedUrl, currentTask.OutputFilePath ?? runtime.CurrentOutputFilePath, observedAt);
|
||||
await dbContext.SaveChangesAsync();
|
||||
|
||||
await StartInternalAsync(
|
||||
session,
|
||||
currentTask,
|
||||
retryStream,
|
||||
FfmpegInputOptionProfile.Minimal,
|
||||
hasRetriedWithCompatibilityProfile: true,
|
||||
hasRetriedWithRefreshedStream: runtime.HasRetriedWithRefreshedStream,
|
||||
runtime.RetryAttemptCount + 1);
|
||||
|
||||
var restartedAt = DateTimeOffset.UtcNow;
|
||||
session.MarkRunning(restartedAt);
|
||||
currentTask.MarkRunning(restartedAt);
|
||||
await dbContext.SaveChangesAsync();
|
||||
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Info,
|
||||
"FFmpeg",
|
||||
"ffmpeg startup retry succeeded with minimal compatibility options.",
|
||||
liveRoomId: session.LiveRoomId,
|
||||
recordSessionId: session.Id,
|
||||
recordTaskId: currentTask.Id);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
if (runtime.StartupFailureKind != StartupFailureKind.StreamHandshake || runtime.HasRetriedWithRefreshedStream)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
var adapterFactory = scope.ServiceProvider.GetRequiredService<ILivePlatformAdapterFactory>();
|
||||
var liveRoomStatusService = scope.ServiceProvider.GetRequiredService<LiveRoomStatusService>();
|
||||
var adapter = adapterFactory.GetByPlatform(session.LiveRoom.Platform);
|
||||
var liveStatus = await adapter.GetLiveStatusAsync(session.LiveRoom.RoomId);
|
||||
await liveRoomStatusService.ApplySnapshotAsync(session.LiveRoom, liveStatus, observedAt);
|
||||
await dbContext.SaveChangesAsync();
|
||||
|
||||
if (!liveStatus.IsLive)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Warning,
|
||||
"FFmpeg",
|
||||
"Initial ffmpeg input handshake failed. Refreshing stream URL and retrying once.",
|
||||
runtime.LastStartupFailureLine,
|
||||
session.LiveRoomId,
|
||||
session.Id,
|
||||
currentTask.Id);
|
||||
|
||||
var refreshedStream = await adapter.GetStreamUrlAsync(session.LiveRoom.RoomId, session.PreferredQuality);
|
||||
session.MarkStarting(refreshedStream.SelectedUrl, session.OutputPathPattern ?? runtime.OutputPathPattern, observedAt);
|
||||
session.ActivateSegment(Math.Max(1, runtime.CurrentSegmentIndex), observedAt);
|
||||
currentTask.MarkStarting(refreshedStream.SelectedUrl, currentTask.OutputFilePath ?? runtime.CurrentOutputFilePath, observedAt);
|
||||
await dbContext.SaveChangesAsync();
|
||||
|
||||
await StartInternalAsync(
|
||||
session,
|
||||
currentTask,
|
||||
refreshedStream,
|
||||
runtime.InputOptionProfile,
|
||||
hasRetriedWithCompatibilityProfile: runtime.HasRetriedWithCompatibilityProfile,
|
||||
hasRetriedWithRefreshedStream: true,
|
||||
runtime.RetryAttemptCount + 1);
|
||||
|
||||
var refreshedRetryStartedAt = DateTimeOffset.UtcNow;
|
||||
session.MarkRunning(refreshedRetryStartedAt);
|
||||
currentTask.MarkRunning(refreshedRetryStartedAt);
|
||||
await dbContext.SaveChangesAsync();
|
||||
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Info,
|
||||
"FFmpeg",
|
||||
"ffmpeg startup retry succeeded with a refreshed stream URL.",
|
||||
liveRoomId: session.LiveRoomId,
|
||||
recordSessionId: session.Id,
|
||||
recordTaskId: currentTask.Id);
|
||||
|
||||
return true;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Recover ffmpeg startup failure failed for session {RecordSessionId}", runtime.RecordSessionId);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private async Task FinalizeExitedSessionAsync(
|
||||
SessionProcessRuntime runtime,
|
||||
Process process,
|
||||
SessionDanmakuXmlRecorder.DanmakuSegmentSummary? activeDanmakuSummary)
|
||||
{
|
||||
using var scope = _serviceScopeFactory.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
||||
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISystemSettingsService>();
|
||||
var emailNotificationService = scope.ServiceProvider.GetRequiredService<IEmailNotificationService>();
|
||||
var settings = await settingsService.GetAsync();
|
||||
|
||||
var session = await dbContext.RecordSessions
|
||||
.Include(item => item.LiveRoom)
|
||||
.Include(item => item.RecordTasks)
|
||||
.ThenInclude(item => item.Result)
|
||||
.FirstOrDefaultAsync(item => item.Id == runtime.RecordSessionId);
|
||||
|
||||
if (session is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var currentTask = session.RecordTasks.FirstOrDefault(item => item.Id == runtime.CurrentTaskId)
|
||||
?? session.RecordTasks.OrderByDescending(static item => item.SegmentIndex).FirstOrDefault();
|
||||
if (currentTask is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var endedAt = DateTimeOffset.UtcNow;
|
||||
string? finalizationError = null;
|
||||
var effectiveOutputPath = runtime.CurrentOutputFilePath;
|
||||
|
||||
if (session.SaveMode == RecordSaveMode.SingleFile &&
|
||||
session.OutputFormat == RecordOutputFormat.Mp4 &&
|
||||
!string.IsNullOrWhiteSpace(session.OutputPathPattern))
|
||||
{
|
||||
var finalOutputPath = Path.IsPathRooted(session.OutputPathPattern)
|
||||
? session.OutputPathPattern
|
||||
: Path.GetFullPath(session.OutputPathPattern, AppContext.BaseDirectory);
|
||||
var finalizationResult = await TryFinalizeMp4Async(settings.FfmpegPath, runtime.RecorderOutputPath, finalOutputPath);
|
||||
effectiveOutputPath = finalizationResult.OutputPath;
|
||||
finalizationError = finalizationResult.ErrorMessage;
|
||||
}
|
||||
|
||||
var fileSize = CalculateFileSize(effectiveOutputPath);
|
||||
var durationSeconds = currentTask.StartedAt.HasValue
|
||||
? (double?)Math.Max(0, (endedAt - currentTask.StartedAt.Value).TotalSeconds)
|
||||
: null;
|
||||
var danmakuPath = activeDanmakuSummary?.FilePath ?? currentTask.Result?.DanmakuFilePath ?? GuessDanmakuPath(currentTask.OutputFilePath);
|
||||
var danmakuMessageCount = activeDanmakuSummary?.MessageCount ?? CountDanmakuMessages(danmakuPath);
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(finalizationError))
|
||||
{
|
||||
currentTask.MarkFailed(finalizationError, endedAt);
|
||||
session.MarkFailed(finalizationError, endedAt);
|
||||
}
|
||||
else if (runtime.StopRequested)
|
||||
{
|
||||
currentTask.MarkStopped(endedAt, durationSeconds);
|
||||
session.MarkStopped(endedAt);
|
||||
}
|
||||
else if (process.ExitCode == 0 && (runtime.CompletionRequested || !runtime.StopRequested))
|
||||
{
|
||||
currentTask.MarkCompleted(endedAt, durationSeconds);
|
||||
session.MarkCompleted(endedAt);
|
||||
}
|
||||
else
|
||||
{
|
||||
var errorMessage = $"ffmpeg exit code: {process.ExitCode}";
|
||||
currentTask.MarkFailed(errorMessage, endedAt);
|
||||
session.MarkFailed(errorMessage, endedAt);
|
||||
}
|
||||
|
||||
currentTask.DetachProcess(endedAt);
|
||||
session.SyncSegmentCount(session.RecordTasks.Count, endedAt);
|
||||
UpsertRecordResult(currentTask, dbContext, effectiveOutputPath, fileSize, durationSeconds, danmakuPath, danmakuMessageCount, endedAt);
|
||||
|
||||
await dbContext.SaveChangesAsync();
|
||||
|
||||
await logService.WriteAsync(
|
||||
session.Status == RecordSessionStatus.Completed ? SystemLogLevel.Info :
|
||||
session.Status == RecordSessionStatus.Stopped ? SystemLogLevel.Warning :
|
||||
SystemLogLevel.Error,
|
||||
"FFmpeg",
|
||||
$"Recording session exited with status={session.Status}.",
|
||||
$"exitCode={process.ExitCode}; output={effectiveOutputPath}; recorderOutput={runtime.RecorderOutputPath}",
|
||||
session.LiveRoomId,
|
||||
session.Id,
|
||||
currentTask.Id);
|
||||
|
||||
if (session.Status == RecordSessionStatus.Failed && session.LiveRoom is not null)
|
||||
{
|
||||
await emailNotificationService.SendExceptionAsync(
|
||||
"FFmpeg",
|
||||
"Recording session exited abnormally.",
|
||||
$"exitCode={process.ExitCode}; output={effectiveOutputPath}",
|
||||
session.LiveRoom,
|
||||
currentTask);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task PersistProcessBindingAsync(Guid recordSessionId, Guid recordTaskId, int processId, CancellationToken cancellationToken)
|
||||
{
|
||||
using var scope = _serviceScopeFactory.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
||||
var session = await dbContext.RecordSessions.FirstOrDefaultAsync(item => item.Id == recordSessionId, cancellationToken);
|
||||
var task = await dbContext.RecordTasks.FirstOrDefaultAsync(item => item.Id == recordTaskId, cancellationToken);
|
||||
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
session?.AttachProcess(processId, now);
|
||||
task?.AttachProcess(processId, now);
|
||||
if (session is not null || task is not null)
|
||||
{
|
||||
await dbContext.SaveChangesAsync(cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task StartDanmakuAsync(SessionProcessRuntime runtime, RecordTask initialTask, CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
LiveRoom? liveRoom = null;
|
||||
var includeNonChatEvents = true;
|
||||
|
||||
using (var readScope = _serviceScopeFactory.CreateScope())
|
||||
{
|
||||
var dbContext = readScope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
||||
var adapterFactory = readScope.ServiceProvider.GetRequiredService<ILiveDanmakuAdapterFactory>();
|
||||
var settingsService = readScope.ServiceProvider.GetRequiredService<ISystemSettingsService>();
|
||||
var settings = await settingsService.GetAsync(cancellationToken);
|
||||
|
||||
if (!settings.EnableDanmakuRecording)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
liveRoom = await dbContext.LiveRooms
|
||||
.AsNoTracking()
|
||||
.FirstOrDefaultAsync(item => item.Id == runtime.LiveRoomId, cancellationToken);
|
||||
if (liveRoom is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
includeNonChatEvents = settings.DanmakuIncludeNonChatEvents;
|
||||
var adapter = adapterFactory.GetByPlatform(liveRoom.Platform);
|
||||
runtime.DanmakuConnection = await adapter.ConnectAsync(
|
||||
new DanmakuConnectionContext(
|
||||
liveRoom.Id,
|
||||
runtime.RecordSessionId,
|
||||
liveRoom.Platform,
|
||||
liveRoom.RoomId,
|
||||
liveRoom.AnchorName,
|
||||
liveRoom.Title,
|
||||
liveRoom.SourceUrl,
|
||||
settings.DanmakuMinPollIntervalMilliseconds,
|
||||
settings.DanmakuRetryDelayMaxSeconds),
|
||||
cancellationToken);
|
||||
}
|
||||
|
||||
runtime.DanmakuRecorder = new SessionDanmakuXmlRecorder(
|
||||
liveRoom.Platform,
|
||||
liveRoom.Id,
|
||||
runtime.RecordSessionId,
|
||||
liveRoom.RoomId);
|
||||
await runtime.DanmakuRecorder.StartSegmentAsync(
|
||||
initialTask.Id,
|
||||
initialTask.SegmentIndex,
|
||||
initialTask.OutputFilePath ?? runtime.OutputPathPattern,
|
||||
initialTask.StartedAt ?? DateTimeOffset.UtcNow);
|
||||
runtime.HasInitializedDanmaku = true;
|
||||
|
||||
runtime.DanmakuPumpTask = Task.Run(
|
||||
() => runtime.DanmakuConnection.StartAsync(
|
||||
danmakuEvent =>
|
||||
{
|
||||
if (runtime.DanmakuRecorder is null)
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
if (!includeNonChatEvents &&
|
||||
!string.Equals(danmakuEvent.Type, "chat", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
return runtime.DanmakuRecorder.AppendAsync(danmakuEvent);
|
||||
},
|
||||
runtime.DanmakuCancellation.Token),
|
||||
runtime.DanmakuCancellation.Token);
|
||||
|
||||
using var logScope = _serviceScopeFactory.CreateScope();
|
||||
var logService = logScope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Info,
|
||||
"Danmaku",
|
||||
"Danmaku capture started for the recording session.",
|
||||
liveRoomId: liveRoom.Id,
|
||||
recordSessionId: runtime.RecordSessionId,
|
||||
recordTaskId: initialTask.Id,
|
||||
cancellationToken: cancellationToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Start danmaku capture failed for session {RecordSessionId}", runtime.RecordSessionId);
|
||||
|
||||
try
|
||||
{
|
||||
using var scope = _serviceScopeFactory.CreateScope();
|
||||
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
||||
await logService.WriteAsync(
|
||||
SystemLogLevel.Warning,
|
||||
"Danmaku",
|
||||
"Danmaku capture startup failed. Recording will continue without live comments until retry succeeds.",
|
||||
ex.ToString(),
|
||||
runtime.LiveRoomId,
|
||||
runtime.RecordSessionId,
|
||||
runtime.CurrentTaskId,
|
||||
cancellationToken);
|
||||
}
|
||||
catch (Exception logEx)
|
||||
{
|
||||
_logger.LogWarning(logEx, "Persist danmaku startup failure log failed for session {RecordSessionId}", runtime.RecordSessionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static bool IsRetryableStartupFailureLine(string line) =>
|
||||
line.Contains("Error reading HTTP response", StringComparison.OrdinalIgnoreCase) ||
|
||||
line.Contains("unexpected EOF", StringComparison.OrdinalIgnoreCase) ||
|
||||
line.Contains("Connection reset", StringComparison.OrdinalIgnoreCase) ||
|
||||
line.Contains("Connection refused", StringComparison.OrdinalIgnoreCase) ||
|
||||
line.Contains("I/O error", StringComparison.OrdinalIgnoreCase) ||
|
||||
line.Contains("Invalid data found when processing input", StringComparison.OrdinalIgnoreCase) ||
|
||||
line.Contains("Server returned 4", StringComparison.OrdinalIgnoreCase) ||
|
||||
line.Contains("Server returned 5", StringComparison.OrdinalIgnoreCase) ||
|
||||
line.Contains("HTTP error 4", StringComparison.OrdinalIgnoreCase) ||
|
||||
line.Contains("HTTP error 5", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
private static bool IsOptionCompatibilityFailureLine(string line) =>
|
||||
line.Contains("Option not found", StringComparison.OrdinalIgnoreCase) ||
|
||||
line.Contains("Unrecognized option", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
private async Task EnsureDanmakuSegmentAsync(SessionProcessRuntime runtime, RecordTask recordTask, DateTimeOffset startedAt)
|
||||
{
|
||||
if (runtime.DanmakuRecorder is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await runtime.DanmakuRecorder.StartSegmentAsync(
|
||||
recordTask.Id,
|
||||
recordTask.SegmentIndex,
|
||||
recordTask.OutputFilePath ?? runtime.CurrentOutputFilePath,
|
||||
startedAt);
|
||||
}
|
||||
|
||||
private sealed class SessionProcessRuntime : IDisposable
|
||||
{
|
||||
public SessionProcessRuntime(
|
||||
Guid recordSessionId,
|
||||
Guid liveRoomId,
|
||||
string streamUrl,
|
||||
string outputPathPattern,
|
||||
string recorderOutputPath,
|
||||
RecordOutputFormat outputFormat,
|
||||
RecordSaveMode saveMode,
|
||||
Guid currentTaskId,
|
||||
int currentSegmentIndex,
|
||||
string currentOutputFilePath,
|
||||
string selectedQuality,
|
||||
string selectedProtocol,
|
||||
StreamInputHeaders? inputHeaders,
|
||||
FfmpegInputOptionProfile inputOptionProfile,
|
||||
bool hasRetriedWithCompatibilityProfile,
|
||||
bool hasRetriedWithRefreshedStream,
|
||||
int retryAttemptCount)
|
||||
{
|
||||
RecordSessionId = recordSessionId;
|
||||
LiveRoomId = liveRoomId;
|
||||
StreamUrl = streamUrl;
|
||||
OutputPathPattern = outputPathPattern;
|
||||
RecorderOutputPath = recorderOutputPath;
|
||||
OutputFormat = outputFormat;
|
||||
SaveMode = saveMode;
|
||||
CurrentTaskId = currentTaskId;
|
||||
CurrentSegmentIndex = currentSegmentIndex;
|
||||
CurrentOutputFilePath = currentOutputFilePath;
|
||||
SelectedQuality = selectedQuality;
|
||||
SelectedProtocol = selectedProtocol;
|
||||
InputHeaders = inputHeaders;
|
||||
InputOptionProfile = inputOptionProfile;
|
||||
HasRetriedWithCompatibilityProfile = hasRetriedWithCompatibilityProfile;
|
||||
HasRetriedWithRefreshedStream = hasRetriedWithRefreshedStream;
|
||||
RetryAttemptCount = Math.Max(0, retryAttemptCount);
|
||||
}
|
||||
|
||||
public Guid RecordSessionId { get; }
|
||||
public Guid LiveRoomId { get; }
|
||||
public string StreamUrl { get; }
|
||||
public string OutputPathPattern { get; }
|
||||
public string RecorderOutputPath { get; }
|
||||
public RecordOutputFormat OutputFormat { get; }
|
||||
public RecordSaveMode SaveMode { get; }
|
||||
public Guid CurrentTaskId { get; set; }
|
||||
public int CurrentSegmentIndex { get; set; }
|
||||
public string CurrentOutputFilePath { get; set; }
|
||||
public string SelectedQuality { get; }
|
||||
public string SelectedProtocol { get; }
|
||||
public StreamInputHeaders? InputHeaders { get; }
|
||||
public FfmpegInputOptionProfile InputOptionProfile { get; }
|
||||
public bool HasRetriedWithCompatibilityProfile { get; }
|
||||
public bool HasRetriedWithRefreshedStream { get; }
|
||||
public Process? Process { get; private set; }
|
||||
public int ProcessId => Process?.Id ?? 0;
|
||||
public bool CompletionRequested { get; private set; }
|
||||
public bool StopRequested { get; private set; }
|
||||
public bool HasInitializedDanmaku { get; set; }
|
||||
public bool HasOpenedFirstSegment { get; set; }
|
||||
public int RetryAttemptCount { get; }
|
||||
public StartupFailureKind StartupFailureKind { get; private set; }
|
||||
public string? LastStartupFailureLine { get; private set; }
|
||||
public SemaphoreSlim Gate { get; } = new(1, 1);
|
||||
public CancellationTokenSource DanmakuCancellation { get; } = new();
|
||||
public TaskCompletionSource<bool> ExitCompletion { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
public SessionDanmakuXmlRecorder? DanmakuRecorder { get; set; }
|
||||
public ILiveDanmakuConnection? DanmakuConnection { get; set; }
|
||||
public Task? DanmakuPumpTask { get; set; }
|
||||
|
||||
public void AttachProcess(Process process) => Process = process;
|
||||
|
||||
public void MarkStopRequested(bool markAsCompletedOnExit)
|
||||
{
|
||||
if (markAsCompletedOnExit)
|
||||
{
|
||||
CompletionRequested = true;
|
||||
StopRequested = false;
|
||||
}
|
||||
else
|
||||
{
|
||||
CompletionRequested = false;
|
||||
StopRequested = true;
|
||||
}
|
||||
}
|
||||
|
||||
public void MarkStartupFailure(StartupFailureKind kind, string line)
|
||||
{
|
||||
if (StartupFailureKind == StartupFailureKind.InputOptionCompatibility &&
|
||||
kind != StartupFailureKind.InputOptionCompatibility)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
StartupFailureKind = kind;
|
||||
LastStartupFailureLine = line;
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
DanmakuCancellation.Dispose();
|
||||
Gate.Dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user