program ThreadedAsyncLogging;

// Atlas Log in depth. Four threads log through one logger. An asynchronous
// sink with a bounded queue writes their events to a JSON-lines file on a
// background thread; a callback sink watches for warnings on the logging
// threads; a custom sink counts events per level. A second part holds a slow
// target back to show the drop-newest and drop-oldest overflow policies and
// the Dropped and TargetErrors counters. Exit code 0 means every check passed.

{$APPTYPE CONSOLE}

uses
  System.SysUtils, System.Classes, System.SyncObjs, System.IOUtils, System.JSON,
  System.Generics.Collections, Atlas.Log, Atlas.LogSinks;

const
  Workers = 4;
  Steps = 500;

type
  // A custom sink. DoWrite runs under the sink's own lock, so it needs no locking.
  TLevelCounter = class(TAtlasLogSink)
  protected
    procedure DoWrite(const Event: TAtlasLogEvent); override;
  public
    Counts: array[TAtlasLogLevel] of Integer;
  end;

  // Handlers for the callback sinks.
  THandlers = class
  public
    Warnings: Integer;
    Gate: TEvent;
    Written: TList<string>;
    constructor Create;
    destructor Destroy; override;
    procedure WarningSeen(Sender: TObject; const Event: TAtlasLogEvent);
    procedure SlowWrite(Sender: TObject; const Event: TAtlasLogEvent);
  end;

procedure TLevelCounter.DoWrite(const Event: TAtlasLogEvent);
begin
  Inc(Counts[Event.Level]);
end;

constructor THandlers.Create;
begin
  inherited;
  Gate := TEvent.Create(nil, True, True, '');
  Written := TList<string>.Create;
end;

destructor THandlers.Destroy;
begin
  Written.Free;
  Gate.Free;
  inherited;
end;

procedure THandlers.WarningSeen(Sender: TObject; const Event: TAtlasLogEvent);
begin
  // Runs on the thread that logged; calls into one sink never overlap.
  Inc(Warnings);
end;

procedure THandlers.SlowWrite(Sender: TObject; const Event: TAtlasLogEvent);
begin
  // Runs on the asynchronous sink's background thread; waits while the gate is closed.
  Gate.WaitFor(INFINITE);
  Written.Add(Event.Message);
end;

procedure Check(Condition: Boolean; const What: string);
begin
  if not Condition then raise Exception.Create('Check failed: ' + What);
  Writeln('  ok: ', What);
end;

function ReadLines(const FileName: string): TArray<string>;
var Stream: TFileStream; Bytes: TBytes;
begin
  Stream := TFileStream.Create(FileName, fmOpenRead or fmShareDenyNone);
  try
    SetLength(Bytes, Stream.Size);
    if Length(Bytes) > 0 then Stream.ReadBuffer(Bytes[0], Length(Bytes));
  finally
    Stream.Free;
  end;
  Result := TEncoding.UTF8.GetString(Bytes).Split([sLineBreak], TStringSplitOptions.ExcludeEmpty);
end;

// A factory, so each thread captures its own number rather than the loop variable.
function Worker(Log: TAtlasLogger; Number: Integer): TProc;
begin
  Result :=
    procedure
    var Step: Integer; Scope: IInterface; Channel: TAtlasLogChannel;
    begin
      Channel := Log.Channel('worker');
      Scope := Log.BeginScope('worker ' + IntToStr(Number));
      for Step := 1 to Steps do
      begin
        Channel.Debug('Worker {worker} step {step}', [Number, Step]);
        if Step = Steps div 2 then Channel.Warning('Worker {worker} is half way', [Number]);
      end;
    end;
end;

procedure ThreadsIntoAsyncFile(const Folder: string; Handlers: THandlers);
var
  Log: TAtlasLogger; Json: TAtlasFileSink; Async: TAtlasAsyncSink; Alerts: TAtlasCallbackSink; Counter: TLevelCounter;
  Threads: TArray<TThread>; T, Step, Previous: Integer; Line: string; Lines: TArray<string>; Parsed: TJSONValue;
  LastStep: TDictionary<string, Integer>; Sequences: TDictionary<Int64, Boolean>; Ordered: Boolean;
begin
  Log := TAtlasLogger.Create(nil);
  Json := TAtlasFileSink.Create(nil);
  Async := TAtlasAsyncSink.Create(nil);
  Alerts := TAtlasCallbackSink.Create(nil);
  Counter := TLevelCounter.Create(nil);
  LastStep := TDictionary<string, Integer>.Create;
  Sequences := TDictionary<Int64, Boolean>.Create;
  try
    Log.MinLevel := llDebug;
    // The file sink is the async sink's target; it is not attached to the logger itself.
    Json.FileName := TPath.Combine(Folder, 'workers.jsonl');
    Json.Layout := loJson;
    Async.Target := Json;
    Async.Capacity := 256;
    Async.Overflow := ofBlock;
    Async.Logger := Log;
    Alerts.MinLevel := llWarning;
    Alerts.OnEvent := Handlers.WarningSeen;
    Alerts.Logger := Log;
    Counter.Logger := Log;

    SetLength(Threads, Workers);
    for T := 0 to High(Threads) do
    begin
      Threads[T] := TThread.CreateAnonymousThread(Worker(Log, T + 1));
      Threads[T].FreeOnTerminate := False;
      Threads[T].Start;
    end;
    for T := 0 to High(Threads) do
    begin
      Threads[T].WaitFor;
      FreeAndNil(Threads[T]);
    end;
    // Waits until the background thread has written the whole queue, then flushes the file.
    Log.Flush;

    Check((Async.Pending = 0) and (Async.Dropped = 0) and (Async.TargetErrors = 0), 'Flush emptied the queue and ofBlock dropped nothing');
    Check(Handlers.Warnings = Workers, 'the callback sink saw one warning per worker');
    Check((Counter.Counts[llDebug] = Workers * Steps) and (Counter.Counts[llWarning] = Workers), 'the custom sink counted every event by level');
    Lines := ReadLines(Json.ActiveFileName);
    Check(Length(Lines) = Workers * (Steps + 1), Format('%d JSON lines in %s', [Length(Lines), ExtractFileName(Json.ActiveFileName)]));
    Ordered := True;
    for Line in Lines do
    begin
      Parsed := TJSONObject.ParseJSONValue(Line);
      try
        if not (Parsed is TJSONObject) then Check(False, 'every line parses as a JSON object');
        if Sequences.ContainsKey(Parsed.GetValue<Int64>('seq')) then Check(False, 'sequence numbers are unique');
        Sequences.Add(Parsed.GetValue<Int64>('seq'), True);
        if Parsed.GetValue<string>('level') = 'debug' then
        begin
          Step := StrToInt(Parsed.GetValue<string>('props.step'));
          if LastStep.TryGetValue(Parsed.GetValue<string>('scope'), Previous) and (Step <> Previous + 1) then Ordered := False;
          LastStep.AddOrSetValue(Parsed.GetValue<string>('scope'), Step);
        end;
      finally
        Parsed.Free;
      end;
    end;
    Check(Ordered and (LastStep.Count = Workers), 'each thread''s events stay in order and carry that thread''s scope only');
    Writeln('    ', Lines[High(Lines)]);
  finally
    Sequences.Free;
    LastStep.Free;
    // The async sink first: destroying it writes anything still queued to its target.
    Async.Free;
    Json.Free;
    Alerts.Free;
    Counter.Free;
    Log.Free;
  end;
end;

procedure OverflowPolicies(Handlers: THandlers);
var Log: TAtlasLogger; Slow: TAtlasCallbackSink; Async: TAtlasAsyncSink; Broken: TAtlasFileSink; I: Integer; Before: Int64;
begin
  Log := TAtlasLogger.Create(nil);
  Slow := TAtlasCallbackSink.Create(nil);
  Async := TAtlasAsyncSink.Create(nil);
  Broken := TAtlasFileSink.Create(nil);
  try
    Slow.OnEvent := Handlers.SlowWrite;
    Async.Target := Slow;
    Async.Capacity := 10;
    Async.Overflow := ofDropNewest;
    Async.Logger := Log;

    // Hold the target back: the queue fills and the overflow policy decides.
    Handlers.Gate.ResetEvent;
    for I := 1 to 30 do Log.Info('n{i}', [I]);
    Check((Async.Dropped >= 19) and (Async.Dropped <= 20), Format('ofDropNewest: %d of 30 events dropped while the target was stuck', [Async.Dropped]));
    Handlers.Gate.SetEvent;
    Log.Flush;
    Check((Handlers.Written.Count + Async.Dropped = 30) and (Handlers.Written[0] = 'n1'), 'the rest were written, oldest first');

    Handlers.Written.Clear;
    Handlers.Gate.ResetEvent;
    Async.Overflow := ofDropOldest;
    Before := Async.Dropped;
    for I := 1 to 30 do Log.Info('o{i}', [I]);
    Handlers.Gate.SetEvent;
    Log.Flush;
    Check((Async.Dropped > Before) and (Handlers.Written.Count + (Async.Dropped - Before) = 30) and (Handlers.Written.Last = 'o30'),
      Format('ofDropOldest: %d dropped from the front, the newest event kept', [Async.Dropped - Before]));

    // A target that fails: a file sink without a FileName raises in DoWrite.
    Async.Target := Broken;
    Log.Info('Nowhere to go');
    Log.Flush;
    Check((Async.TargetErrors = 1) and (Log.SinkErrors = 0), 'a failing target is counted in TargetErrors and never raised');
  finally
    // Never leave the background thread waiting on a closed gate.
    Handlers.Gate.SetEvent;
    Async.Free;
    Broken.Free;
    Slow.Free;
    Log.Free;
  end;
end;

var
  Folder: string;
  Handlers: THandlers;
begin
  Folder := TPath.Combine(TPath.GetTempPath, 'AtlasLogThreadedAsync');
  Handlers := THandlers.Create;
  try
    try
      if TDirectory.Exists(Folder) then TDirectory.Delete(Folder, True);
      Writeln('Part 1: four threads, one asynchronous JSON file');
      ThreadsIntoAsyncFile(Folder, Handlers);
      Writeln('Part 2: a stuck target and the overflow policies');
      OverflowPolicies(Handlers);
      Writeln('All checks passed.');
    finally
      if TDirectory.Exists(Folder) then TDirectory.Delete(Folder, True);
      Handlers.Free;
    end;
  except
    on E: Exception do
    begin
      Writeln(E.ClassName, ': ', E.Message);
      ExitCode := 1;
    end;
  end;
end.