program ReactiveX;

{$APPTYPE CONSOLE}

uses
  Spring,
  Spring.Collections,
  SysUtils;

type
  IDisposable = interface
    procedure Dispose;
  end;

  IObserver<T> = interface
    procedure OnNext(const value: T);
    procedure OnError(const error: Exception);
    procedure OnCompleted;
  end;

  IObservable<T> = interface
    function Subscribe(const observer: IObserver<T>): IDisposable;
  end;

  TMyConsoleObserver<T> = class(TInterfacedObject, IObserver<T>)
    procedure OnNext(const value: T);
    procedure OnError(const error: Exception);
    procedure OnCompleted;
  end;

{ TMyConsoleObserver<T> }

procedure TMyConsoleObserver<T>.OnCompleted;
begin
  Writeln('Sequence terminated');
end;

procedure TMyConsoleObserver<T>.OnError(const error: Exception);
begin
  Writeln('Sequence faulted with ', error.Message);
end;

procedure TMyConsoleObserver<T>.OnNext(const value: T);
begin
  Writeln('Received value ', TValue.From<T>(value).ToString);
end;

type
  TMySequenceOfNumbers = class(TInterfacedObject, IObservable<Integer>)
    function Subscribe(const observer: IObserver<Integer>): IDisposable;
  end;

{ TMySequenceOfNumbers }

function TMySequenceOfNumbers.Subscribe(
  const observer: IObserver<Integer>): IDisposable;
begin
  observer.OnNext(1);
  observer.OnNext(2);
  observer.OnNext(3);
  observer.OnCompleted();
end;

type
  ISubject<T> = interface(IObservable<T>)
    procedure OnNext(const value: T);
    procedure OnError(const error: Exception);
    procedure OnCompleted;
  end;

  TSubject<T> = class(TInterfacedObject, ISubject<T>, IObserver<T>, IObservable<T>)
  private
    fObservers: IList<IObserver<T>>;
    procedure Unsubscribe(const observer: IObserver<T>);
  public
    constructor Create;

    procedure OnNext(const value: T);
    procedure OnError(const error: Exception);
    procedure OnCompleted;

    function Subscribe(const observer: IObserver<T>): IDisposable;

    type
      TSubscription = class(TInterfacedObject, IDisposable)
      private
        fSubject: TSubject<T>;//unsafe
        fObserver: IObserver<T>;
      public
        constructor Create(const subject: TSubject<T>; const observer: IObserver<T>);

        procedure Dispose;
      end;
  end;

procedure Main;
var
  numbers: IObservable<Integer>;
  observer: IObserver<Integer>;
begin
  numbers := TMySequenceOfNumbers.Create;
  observer := TMyConsoleObserver<Integer>.Create;
  numbers.Subscribe(observer);
end;

{ TSubject<T> }

constructor TSubject<T>.Create;
begin
  fObservers := TCollections.CreateInterfaceList<IObserver<T>>;
end;

procedure TSubject<T>.OnCompleted;
begin

end;

procedure TSubject<T>.OnError(const error: Exception);
begin

end;

procedure TSubject<T>.OnNext(const value: T);
var
  observer: IObserver<T>;
begin
  for observer in fObservers.ToArray do
    observer.OnNext(value);
end;

function TSubject<T>.Subscribe(const observer: IObserver<T>): IDisposable;
begin
  fObservers.Add(observer);
  Result := TSubscription.Create(Self, observer);
end;

procedure TSubject<T>.Unsubscribe(const observer: IObserver<T>);
begin
  fObservers.Remove(observer);
end;

{ TSubject<T>.TSubscription }

constructor TSubject<T>.TSubscription.Create(const subject: TSubject<T>;
  const observer: IObserver<T>);
begin
  inherited Create;
  fSubject := subject;
  fObserver := observer;
end;

procedure TSubject<T>.TSubscription.Dispose;
begin
  if Assigned(fSubject) then
    fSubject.Unsubscribe(fObserver);
end;

type
  Console = record
    class procedure WriteLine(const s: string); static;
  end;

  TAnonymousObserver<T> = class sealed(TInterfacedObject, IObserver<T>)
  private
    fOnNext: TAction<T>;
    fOnError: TAction<Exception>;
    fOnComplected: TProc;
  public
    constructor Create(const onNext: TAction<T>;
      const onError: TAction<Exception>; const onCompleted: TProc); overload;
    constructor Create(const onNext: TAction<T>); overload;

    procedure OnNext(const value: T);
    procedure OnError(const error: Exception);
    procedure OnCompleted;
  end;

procedure WriteSequenceToConsole(const sequence: IObservable<string>);
begin
  sequence.Subscribe(TAnonymousObserver<string>.Create(Console.WriteLine));
end;

procedure Main2;
var
  subject: ISubject<string>;
begin
  subject := TSubject<string>.Create;
  WriteSequenceToConsole(subject);
  subject.OnNext('a');
  subject.OnNext('b');
  subject.OnNext('c');
end;

{ Console }

class procedure Console.WriteLine(const s: string);
begin
  Writeln(s);
end;

{ TAnonymousObserver<T> }

constructor TAnonymousObserver<T>.Create(const onNext: TAction<T>;
  const onError: TAction<Exception>; const onCompleted: TProc);
begin
  inherited Create;
  fOnNext := onNext;
  fOnError := onError;
  fOnComplected := onCompleted;
end;

constructor TAnonymousObserver<T>.Create(const onNext: TAction<T>);
begin
  // TODO: take from global stubs
  Create(onNext,
    procedure(const e: Exception) begin raise e; end,
    procedure begin end);
//    Stubs.Throw, Stubs.Nop);
end;

procedure TAnonymousObserver<T>.OnCompleted;
begin
  fOnComplected();
end;

procedure TAnonymousObserver<T>.OnError(const error: Exception);
begin
  fOnError(error);
end;

procedure TAnonymousObserver<T>.OnNext(const value: T);
begin
  fOnNext(value);
end;

type
  TTimeInterval<T> = record
  private
    fInterval: TTimeSpan;
    fValue: T;
  public
    constructor Create(const value: T; const interval: TTimeSpan);
    property Interval: TTimeSpan read fInterval;
    property Value: T read fValue;
  end;

  TReplaySubject<T> = class(TInterfacedObject, ISubject<T>)
  private const
    InfiniteBufferSize = MaxInt;
  private
    fQueue: IQueue<TTimeInterval<T>>;
    fObservers: IList<IObserver<T>>;
    fBufferSize: Integer;
    fWindow: TTimeSpan;
    fStopwatch: TStopwatch;
    procedure Unsubscribe(const observer: IObserver<T>);
  protected
    procedure Next(const value: T);
    function Replay(const observer: IObserver<T>): Integer;
    procedure Trim;
  public
    constructor Create(bufferSize: Integer; const window: TTimeSpan); overload;
    constructor Create(bufferSize: Integer = InfiniteBufferSize); overload;
    constructor Create(const window: TTimeSpan); overload;

    procedure OnNext(const value: T);
    procedure OnError(const error: Exception);
    procedure OnCompleted;

    function Subscribe(const observer: IObserver<T>): IDisposable;

    type
      TSubscription = class(TInterfacedObject, IDisposable)
      private
        fSubject: TReplaySubject<T>;//unsafe
        fObserver: IObserver<T>;
      public
        constructor Create(const subject: TReplaySubject<T>; const observer: IObserver<T>);

        procedure Dispose;
      end;
  end;

{ TReplaySubject<T> }

constructor TReplaySubject<T>.Create(bufferSize: Integer;
  const window: TTimeSpan);
begin
  inherited Create;
  fQueue := TCollections.CreateQueue<TTimeInterval<T>>;
  fObservers := TCollections.CreateInterfaceList<IObserver<T>>;
  fBufferSize := bufferSize;
  fWindow := window;
  fStopwatch.Start;
end;

constructor TReplaySubject<T>.Create(bufferSize: Integer);
begin
  Create(bufferSize, TTimeSpan.MaxValue);
end;

constructor TReplaySubject<T>.Create(const window: TTimeSpan);
begin
  Create(InfiniteBufferSize, window);
end;

procedure TReplaySubject<T>.Next(const value: T);
begin
  fQueue.Enqueue(TTimeInterval<T>.Create(value, fStopwatch.Elapsed));
end;

procedure TReplaySubject<T>.OnCompleted;
var
  observer: IObserver<T>;
begin
  for observer in fObservers.ToArray do
    observer.OnCompleted;
end;

procedure TReplaySubject<T>.OnError(const error: Exception);
var
  observer: IObserver<T>;
begin
  for observer in fObservers.ToArray do
    observer.OnError(error);
end;

procedure TReplaySubject<T>.OnNext(const value: T);
var
  observer: IObserver<T>;
begin
  Trim;
  Next(value);
  for observer in fObservers.ToArray do
    observer.OnNext(value);
end;

function TReplaySubject<T>.Replay(const observer: IObserver<T>): Integer;
var
  item: TTimeInterval<T>;
begin
  Result := fQueue.Count;
  for item in fQueue do
    observer.OnNext(item.Value);
end;

function TReplaySubject<T>.Subscribe(const observer: IObserver<T>): IDisposable;
begin
  Trim;
  Replay(observer);
  fObservers.Add(observer);
  Result := TSubscription.Create(Self, observer);
end;

procedure TReplaySubject<T>.Trim;
var
  now: TTimeSpan;
begin
  now := fStopwatch.Elapsed;
  while fQueue.Count > fBufferSize do
    fQueue.Dequeue;
  while (fQueue.Count > 0) and (now.Subtract(fQueue.Peek.Interval) > fWindow) do
    fQueue.Dequeue;
end;

procedure TReplaySubject<T>.Unsubscribe(const observer: IObserver<T>);
begin
  fObservers.Remove(observer);
end;

{ TReplaySubject<T>.TSubscription }

constructor TReplaySubject<T>.TSubscription.Create(
  const subject: TReplaySubject<T>; const observer: IObserver<T>);
begin
  inherited Create;
  fSubject := subject;
  fObserver := observer;
end;

procedure TReplaySubject<T>.TSubscription.Dispose;
begin
  if Assigned(fSubject) then
    fSubject.Unsubscribe(fObserver);
end;

procedure Main3;
var
  subject: ISubject<string>;
begin
  subject := TReplaySubject<string>.Create;
  subject.OnNext('a');
  WriteSequenceToConsole(subject);
  subject.OnNext('b');
  subject.OnNext('c');
end;

procedure Main4;
var
  subject: ISubject<string>;
begin
  subject := TReplaySubject<string>.Create(2);
  subject.OnNext('a');
  subject.OnNext('b');
  subject.OnNext('c');
  WriteSequenceToConsole(subject);
  subject.OnNext('d');
end;

procedure Main5;
var
  subject: ISubject<string>;
begin
  subject := TReplaySubject<string>.Create(TTimeSpan.FromMilliseconds(150));
  subject.OnNext('w');
  Sleep(100);
  subject.OnNext('x');
  Sleep(100);
  subject.OnNext('y');
  WriteSequenceToConsole(subject);
  subject.OnNext('z');
end;


{ TTimeInterval<T> }

constructor TTimeInterval<T>.Create(const value: T; const interval: TTimeSpan);
begin
  fInterval := interval;
  fValue := value;
end;

begin
  try
    Main5;
  except
    on E: Exception do
      Writeln(E.ClassName, ': ', E.Message);
  end;
  Readln;
end.