using System.Collections.Concurrent; using Inspectron.Epson.PrintServer.Telemetry; using Microsoft.Extensions.Logging; namespace Inspectron.Epson.Queue; public class PrinterQueue { public string PrinterIp { get; } private readonly ConcurrentQueue _queue; private readonly ConcurrentStack _priorityQueue; // For failed jobs private readonly SemaphoreSlim _signal; private readonly CancellationTokenSource _cancellationTokenSource; private Task _processingTask; private readonly IPrintService _printService; private readonly ILogger _logger; private readonly PrintServer _printServer; private readonly IJobStatusReporter _statusReporter; private readonly IInsinTelemetry _telemetry; public bool IsProcessing { get; private set; } public int QueueLength => _queue.Count + _priorityQueue.Count; public PrinterQueue(string printerIp, IPrintService printService, ILogger logger, PrintServer printServer, IJobStatusReporter statusReporter, IInsinTelemetry telemetry) { PrinterIp = printerIp; _queue = new ConcurrentQueue(); _priorityQueue = new ConcurrentStack(); _signal = new SemaphoreSlim(0); _cancellationTokenSource = new CancellationTokenSource(); _printService = printService; _logger = logger; _printServer = printServer; _statusReporter = statusReporter; _telemetry = telemetry; } public void Enqueue(PrintJob job) { _queue.Enqueue(job); _signal.Release(); // Signal that there's work to do _statusReporter.ReportStatusAsync(job, PrintJobStatus.Received).GetAwaiter().GetResult(); _logger.LogInformation("Job queued for printer {PrinterId}. Queue length: {QueueLength}", PrinterIp, QueueLength); } public void Start() { if (_processingTask != null) return; IsProcessing = true; _processingTask = Task.Run(() => ProcessQueueAsync(_cancellationTokenSource.Token)); _logger.LogInformation("Printer queue {PrinterId} started", PrinterIp); } private void EnqueuePriority(PrintJob job) { _priorityQueue.Push(job); _signal.Release(); _logger.LogInformation("Job priority queued for printer {PrinterId}. Queue length: {QueueLength}", PrinterIp, QueueLength); } private async Task ProcessQueueAsync(CancellationToken cancellationToken) { while (!cancellationToken.IsCancellationRequested) { try { // Wait for signal that there's work or cancellation await _signal.WaitAsync(cancellationToken); PrintJob job = null; if (!_priorityQueue.TryPop(out job)) { _queue.TryDequeue(out job); } if (job!=null) { _logger.LogInformation("Processing job on printer {PrinterId}", PrinterIp); _printServer.EnterPrintLock(); var stopwatch = System.Diagnostics.Stopwatch.StartNew(); try { // Call the actual print function var result = await _printService.PrintAsync(PrinterIp, job); stopwatch.Stop(); if (!result.Success) { var failKind = InsinEventKinds.MapErrorTypeToKind(result.ErrorType); SafeEmit(failKind, InsinMessageFormatter.Format( ("printer", PrinterIp), ("receiptType", job.Document?.ReceiptType.ToString()), ("retry", job.RetryCount.ToString()), ("error", result.ErrorType.ToString()))); if (result.ErrorType == PrintErrorType.ConversionError) { _logger.LogWarning("Job failed due to conversion error, not re-queuing"); await _statusReporter.ReportStatusAsync(job, PrintJobStatus.Failed); } else if (result.ErrorType == PrintErrorType.Offline) { _logger.LogWarning("Printer is offline, job failed"); await _statusReporter.ReportStatusAsync(job, PrintJobStatus.Failed); } else { // Re-queue with retry logic job.RetryCount++; if (job.RetryCount == 1) { await _statusReporter.ReportStatusAsync(job, PrintJobStatus.OutOfPaper); } _logger.LogWarning("Job failed, retrying ({RetryCount}/3)", job.RetryCount); await Task.Delay(5000, cancellationToken); // Wait before retry EnqueuePriority(job); } } else { _logger.LogInformation("Job completed successfully on printer {PrinterId}", PrinterIp); await _statusReporter.ReportStatusAsync(job, PrintJobStatus.Completed); SafeEmit(InsinEventKinds.JobPrinted, InsinMessageFormatter.Format( ("printer", PrinterIp), ("receiptType", job.Document?.ReceiptType.ToString()), ("durationMs", stopwatch.ElapsedMilliseconds.ToString()))); } } catch (Exception ex) { _logger.LogError(ex, "Error processing job"); await _statusReporter.ReportStatusAsync(job, PrintJobStatus.Failed); } finally { _printServer.ExitPrintLock(); } } } catch (OperationCanceledException) { break; } } IsProcessing = false; _logger.LogInformation("Printer queue {PrinterId} stopped", PrinterIp); } public async Task StopAsync() { _cancellationTokenSource.Cancel(); _signal.Release(); // Release to unblock the wait if (_processingTask != null) { await _processingTask; } } public List GetPendingJobs() { return new List(_queue); } private void SafeEmit(string kind, string message) { try { _telemetry.Emit(kind, message); } catch (Exception ex) { _logger.LogWarning(ex, "insin telemetry emit failed"); } } }