uMessageQueue.pas

unit uMessageQueue;

interface
uses  Classes, SyncObjs,Windows;
type
   TMessageQueue = class(TObject)
  private
  CS :TCriticalSection;
  event : Cardinal;
  queue :TStringList;
  public
    constructor Create; virtual;
    destructor Destroy; override;
    function Size:integer;
    function Extract:string;
    procedure Put(packet : string);
    procedure FreeTread;
 end;
implementation

constructor TMessageQueue.Create;
begin
  inherited;
  queue := TStringList.Create;
  CS := TCriticalSection.Create;
 // mutex := CreateMutex(nil, true, nil); //true - ñðàçó çàíèìàåì ñåìàôîð
   event := CreateEvent(nil, false,false, nil);              ///false
end;

destructor TMessageQueue.Destroy;
begin
  queue.Free;
  CloseHandle(event);
  CS.Free;
  inherited;
end;

function TMessageQueue.Size:integer;
begin
  CS.Enter;
  result := queue.Count;
  CS.Leave;
end;
//!!!! îáÿçàòåëüíî ñèíõðîíèçàöèþ è íåäåëèìîñòü  ìåòîäîâ Extract è Put
function TMessageQueue.Extract:string;
begin
result :='';
  if Size=0 then
    begin
      WaitForSingleObject(event, INFINITE);//(1, mutex, true,INFINITE);
     // ReleaseMutex(mutex);
    end;
 if Size>0 then
  begin
  CS.Enter;
 //åñòü åùå ñîîáùåíèÿ â î÷åðåäè
  result :=  queue.Strings[0];
  queue.Delete(0);
  CS.Leave;
   end;
  //  ReleaseMutex(mutex);
end;

procedure TMessageQueue.Put(packet : string);
begin
  CS.Enter;
  queue.Add(packet);
  if queue.Count = 1 then
     begin
     SetEvent(event);//ReleaseMutex(mutex);//ðàçáëîêèðîâàòü
    // PulseEvent(event);
    // ResetEvent(event);
     end;
  CS.Leave;
end;

procedure TMessageQueue.FreeTread;
begin
//åñëè íèêòî íå âèñèò, òî íå íàäî îñâîáîæäàòü
  SetEvent(event);
end;
end.