Print_server/Inspectron.Epson/Queue/PrinterQueue.cs

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<PrintJob> _queue;
    private readonly ConcurrentStack<PrintJob> _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<PrintJob>();
        _priorityQueue = new ConcurrentStack<PrintJob>();
        _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()),
                                ("durationMs",  stopwatch.ElapsedMilliseconds.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)
                    {
                        stopwatch.Stop();
                        _logger.LogError(ex, "Error processing job");
                        await _statusReporter.ReportStatusAsync(job, PrintJobStatus.Failed);
                        SafeEmit(InsinEventKinds.MapErrorTypeToKind(PrintErrorType.Other), InsinMessageFormatter.Format(
                            ("printer",     PrinterIp),
                            ("receiptType", job.Document?.ReceiptType.ToString()),
                            ("retry",       job.RetryCount.ToString()),
                            ("durationMs",  stopwatch.ElapsedMilliseconds.ToString()),
                            ("error",       ex.GetType().Name)));
                    }
                    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<PrintJob> GetPendingJobs()
    {
        return new List<PrintJob>(_queue);
    }

    private void SafeEmit(string kind, string message)
    {
        try { _telemetry.Emit(kind, message); }
        catch (Exception ex) { _logger.LogWarning(ex, "insin telemetry emit failed"); }
    }
}