MycLib/Src/Myc.Signals.FMX.pas

unit Myc.Signals.FMX;

interface

uses
    System.Classes,
    System.SysUtils,
    System.Generics.Collections,
    System.Diagnostics,
    System.Messaging,
    Myc.Signals;

type
    TSignalComponentHelper = class helper for TComponent
        type
            TMsgProc = reference to procedure(out IsDone: Boolean);

            TSignalSubscription = class(TComponent, TSignal.ISubscriber)
            private
                FSignal: TSignal;
                FSigSubscr: TSignal.TSubscription;
                FProc: TMsgProc;
                FNotified: Integer;
                FIdleSubscrId: TMessageSubscriptionId;
                class var
                    FQueued: Integer;
                    FCount: Integer;
                    FItems: TList<TSignalSubscription>;
                    FIdx: Integer;
                class procedure HandleSignals(Timeout: Int64);
                function Notify: Boolean;
            public
                constructor Create(AOwner: TComponent; const ASignal: TSignal; const AProc: TMsgProc); reintroduce;
                destructor Destroy; override;
            end;
    public
        procedure ProcessSignal(const Signal: TSignal; const Proc: TMsgProc); overload;
        procedure ProcessSignal(const Signal: TSignal; const Proc: TProc); overload;
    end;

implementation

uses
    FMX.Types;

{ TSignalComponentHelper.TSignalSubscription }

constructor TSignalComponentHelper.TSignalSubscription.Create(AOwner: TComponent; const ASignal: TSignal; const AProc: TMsgProc);
begin
    inherited Create(AOwner);
    FSignal := ASignal;
    FProc := AProc;
    FIdx := 0;

    if FItems = nil then
    begin
        FItems := TList<TSignalSubscription>.Create;
        FIdleSubscrId :=
            TMessageManager
                .DefaultManager
                .SubscribeToMessage(TIdleMessage, procedure(const Sender: TObject; const M: TMessage) begin HandleSignals(50); end);
    end;
    FItems.Add(Self);

    FSigSubscr := FSignal.Subscribe(Self);
end;

destructor TSignalComponentHelper.TSignalSubscription.Destroy;
begin
    FSigSubscr.Unsubscribe;

    FItems.Remove(Self);
    if FItems.Count = 0 then
    begin
        TMessageManager.DefaultManager.Unsubscribe(TIdleMessage, FIdleSubscrId);
        FreeAndNil(FItems);
    end;

    inherited;
end;

class procedure TSignalComponentHelper.TSignalSubscription.HandleSignals(Timeout: Int64);
begin
    if AtomicExchange(FQueued, 0) = 0 then
        exit;

    var Stopwatch := TStopwatch.StartNew;

    if (FIdx = 0) or (FIdx > FItems.Count) then
        FIdx := FItems.Count;

    while FIdx > 0 do
    begin
        dec(FIdx);
        var sub := FItems[FIdx];

        if AtomicExchange(sub.FNotified, 0) = 1 then
        begin
            var done := false;
            try
                sub.FProc(done);
            finally
                if done then
                    sub.Free;
            end;

            if AtomicDecrement(FCount) = 1 then
                break;
        end;

        if Stopwatch.ElapsedMilliseconds > Timeout then
        begin
            AtomicExchange(FQueued, 1);
            break;
        end;

        if FIdx = 0 then
            FIdx := FItems.Count;
    end;
end;

function TSignalComponentHelper.TSignalSubscription.Notify: Boolean;
begin
    if AtomicExchange(FNotified, 1) = 0 then
        AtomicIncrement(FCount);
    AtomicExchange(FQueued, 1);

    //    if AtomicExchange(FQueued, 1) = 0 then
    //        TThread.Queue( nil,
    //            procedure
    //            begin
    //                if AtomicExchange(FQueued, 0) = 1 then
    //                    HandleSignals( 40 );
    //            end);
    Result := true;
end;

procedure TSignalComponentHelper.ProcessSignal(const Signal: TSignal; const Proc: TProc);
begin
    var cProc: TProc := Proc;
    TSignalSubscription.Create(Self, Signal, procedure(out IsDone: Boolean) begin cProc(); end);
end;

procedure TSignalComponentHelper.ProcessSignal(const Signal: TSignal; const Proc: TMsgProc);
begin
    TSignalSubscription.Create(Self, Signal, Proc);
end;

end.