mmringbf.pas

一套及时通讯的原码

function TMMRingBuffer.AllocBuffer: PBufferItem;
begin
   if (BufferSize > 0) then
   begin
      Result := FAllocator.AllocBuffer(GHND,SizeOf(TBufferItem)+BufferSize);

      { Data occurs directly after the header }
      Result^.lpData         := PChar(Result) + sizeOf(TBufferItem);
      Result^.dwSize         := 0;
      Result^.dwUser         := 0;
      Result^.dwFlags        := 0;
   end
   else Result := nil;
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.CreateBuffers;
var
   i: integer;

begin
   { create the BufferPool itself }
   FBufferPool := FAllocator.AllocBuffer(GHND,FNumBuffers*sizeOf(PBufferItem));
   for i := 0 to FNumBuffers-1 do
   begin
      FBufferPool^[i] := AllocBuffer;
   end;
   if (FQueueMode <> qmNone) then
   begin
      { create the temp buffer }
      FTempBuffer := FAllocator.AllocBuffer(GHND,sizeOf(TMMWaveHdr)+Max(BufferSize,Max(Max(QUEUE_READ_SIZE,QUEUE_WRITE_SIZE),BufferSize)));
      FTempBuffer^.lpData := PChar(FTempBuffer) + sizeOf(TMMWaveHdr);
   end;
   { reset the queue }
   Clear;
   {$IFDEF TRIAL}
   {$DEFINE _HACK2}
   {$I MMHACK.INC}
   {$ENDIF}
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.FreeBuffers;
var
   i: integer;
begin
   if (FBufferPool <> nil) then
   begin
      for i := 0 to FNumBuffers-1 do
      begin
         { unlock and free memory for Buffer }
         if FBufferPool^[i] <> nil then
         begin
            FAllocator.FreeBuffer(Pointer(FBufferPool^[i]));
            FBufferPool^[i] := nil;
         end;
      end;
      FAllocator.FreeBuffer(Pointer(FBufferPool));
      FBufferPool := nil;
   end;
   FAllocator.FreeBuffer(Pointer(FTempBuffer));
   { reset the queue }
   Clear;
end;

{-- TMMRingBuffer --------------------------------------------------------}
Procedure TMMRingBuffer.SetNumBuffers(aValue: integer);
begin
   if (aValue <> FNumBuffers) AND (aValue > 1) then
   begin
      if FOpen then
          raise EMMBufferError.Create(LoadResStr(IDS_PROPERTYOPEN));

      FNumBuffers := aValue;
   end;
   {$IFDEF TRIAL}
   {$DEFINE _HACK3}
   {$I MMHACK.INC}
   {$ENDIF}
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.SetQueueMode(aValue: TMMQueueMode);
begin
   if (aValue <> FQueueMode) then
   begin
      if FOpen then
          raise EMMBufferError.Create(LoadResStr(IDS_PROPERTYOPEN));

      {$IFNDEF WIN32}
      if aValue in [qmRead,qmWrite] then
      begin
         if (csDesigning in ComponentState) then
            Application.MessageBox('"qmRead" and "qmWrite" is only supported under Win32',
                                   'TMMRingBuffer', MB_OK);
      end;
      {$ENDIF}
      FQueueMode := aValue;
   end;
end;

{-- TMMRingBuffer --------------------------------------------------------}
Procedure TMMRingBuffer.SetBufferSize(aValue: Longint);
begin
   if (aValue <> inherited GetBufferSize) then
   begin
      if FOpen then
          raise EMMBufferError.Create(LoadResStr(IDS_PROPERTYOPEN));

      if assigned(FAllocator) then
         FAllocator.Discard;

      inherited SetBufferSize(Max(aValue,MINBUFFERSIZE));
   end;
end;

{-- TMMRingBuffer --------------------------------------------------------}
function TMMRingBuffer.GetBufferSize: Longint;
begin
   Result := inherited GetBufferSize;
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.DoOverflow;
begin
   if assigned(FOnOverflow) then FOnOverflow(Self);
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.DoEmpty;
begin
   if assigned(FOnEmpty) then FOnEmpty(Self);
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.DoFillQueue;
begin
   if assigned(FOnFillQueue) then FOnFillQueue(Self);
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.DoFillQueueEnd;
begin
   if assigned(FOnFillQueueEnd) then FOnFillQueueEnd(Self);
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.DoFlushQueue;
begin
   if assigned(FOnFlushQueue) then FOnFlushQueue(Self);
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.DoFlushQueueEnd;
begin
   if assigned(FOnFlushQueueEnd) then FOnFlushQueueEnd(Self);
end;

{-- TMMRingBuffer --------------------------------------------------------}
function TMMRingBuffer.Empty: Boolean;
begin
   EnterCritical;
   Result := (FQueueCount = 0);
   LeaveCritical;
end;

{-- TMMRingBuffer -------------------------------------------------------}
function TMMRingBuffer.Full: Boolean;
begin
   EnterCritical;
   Result := (FQueueCount = FNumBuffers);
   LeaveCritical;
end;

{-- TMMRingBuffer --------------------------------------------------------}
function TMMRingBuffer.GetQueueCount: integer;
begin
   { returns the number of bytes in the queue }
   EnterCritical;
   Result := FQueueCount;
   LeaveCritical;
end;

{-- TMMRingBuffer --------------------------------------------------------}
function TMMRingBuffer.GetBytesQueued: Longint;
begin
   { returns the number of bytes in the queue }
   EnterCritical;
   Result := FBytesQueued;
   LeaveCritical;
end;

{-- TMMRingBuffer --------------------------------------------------------}
function TMMRingBuffer.GetBytesFree: Longint;
begin
   { returns the number of bytes which can put in the queue }
   EnterCritical;
   Result := (FNumBuffers-FQueueCount)*BufferSize;
   LeaveCritical;
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.Open;
begin
   if not FOpen then
   begin
      {$IFDEF _MMDEBUG}
      DebugStr(0,'Try to open queue...');
      {$ENDIF}

      InitCritical;
      InitThread;
      { round read and write size }
      FReadSize := Max((QUEUE_READ_SIZE div BufferSize)*BufferSize,BufferSize);
      FWriteSize := Max((QUEUE_WRITE_SIZE div BufferSize)*BufferSize,BufferSize);
      CreateBuffers;

      FLoopRec.dwLoop := False;
      FOpen := True;
      FStopping := False;

      {$IFDEF _MMDEBUG}
      DebugStr(0,'Queue is now opened...');
      {$ENDIF}
   end;
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.Close;
begin
   if FOpen then
   begin
      {$IFDEF _MMDEBUG}
      DebugStr(0,'Try to close queue...');
      {$ENDIF}

      FStopping := True;

      { Flush the queue }
      Flush;

      { reset the open flag }
      FOpen := False;

      { shot down the thread }
      DoneThread;

      { free the critical section object }
      DoneCritical;

      { release all buffers }
      FreeBuffers;

      {$IFDEF _MMDEBUG}
      DebugStr(0,'Queue is now closed...');
      {$ENDIF}
   end;
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.Clear;
begin
   EnterCritical;

   FQueueCount := 0;
   FBytesQueued := 0;
   FInIndex := 0;
   FOutIndex := 0;
   FLastIndex := 0;
   FStarted := False;

   LeaveCritical;
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.Flush;
begin
   if FOpen then
   begin
      {$IFDEF _MMDEBUG}
      DebugStr(0,'Try to flush queue...')
      {$ENDIF};

      if (FQueueMode in [qmRead,qmWrite]) then
      begin
         { wait until the thread is ready with the work }
         WaitForDoneEvent(False);

         if (FQueueMode = qmWrite) then
         begin
            { flush the queue }
            FlushQueue;

            { and wait again }
            WaitForDoneEvent(True);
         end;
      end;

      {$IFDEF _MMDEBUG}
      DebugStr(0,'Queue flushed...')
      {$ENDIF};
   end;
   Clear;
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.PutBuffer(lpData: PChar; dwSize,dwUser,dwFlags: DWORD);
var
   Size: Longint;
begin
   if FOpen and FEnabled then
   begin
      if not Full or FWaitIfFull then
      begin
         { insert a new buffer in the queue }
         if (lpData = nil) or (dwSize = 0) then exit;

         EnterCritical;
         try
            Size := Min(dwSize,BufferSize);

            { it's time to write the data and is no other write working ? }
            if (FQueueMode = qmWrite) then
            begin
               if (FQueueCount+1 >= FNumBuffers div 2) and
                  (WaitForSingleObject(FQueueEvent,0) <> WAIT_OBJECT_0) then
               begin
                  { reset the done event }
                  ResetEvent(FDoneEvent);
                  SetEvent(FQueueEvent);
               end;

               LeaveCritical;

               if FWaitIfFull then
               begin
                  while Full do Sleep(SLEEPTIME);
               end;

               EnterCritical;
            end;

            GlobalMoveMem(lpData^,FBufferPool^[FInIndex]^.lpData^,Size);
            FBufferPool^[FInIndex]^.dwSize := Size;
            FBufferPool^[FInIndex]^.dwUser := dwUser;
            FBufferPool^[FInIndex]^.dwFlags:= dwFlags;

            FInIndex := (FInIndex+1)mod FNumBuffers;

            inc(FQueueCount);
            inc(FBytesQueued,Size);

         finally
            LeaveCritical;
         end;

         {$IFDEF _MMDEBUG}
         DebugStr(2,'Buffer '+IntToStr(dwUser)+' added to queue')
         {$ENDIF};
      end
      else
      begin
         {$IFDEF _MMDEBUG}
         DebugStr(2,'Overflow, queue is full...')
         {$ENDIF};

         { queue is full, buffer lost }
         DoOverflow;
      end;
   end;
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.GetBuffer(lpData: PChar; var dwSize,dwUser,dwFlags: DWORD);
var
   Size,Time: Longint;
   bTimeOut: Boolean;

begin
   if FOpen and (lpData <> nil) and (dwSize > 0) then
   begin
      EnterCritical;
      { it's time to load new data ? }
      if FEnabled and
         (FQueueMode = qmRead) and
         (FQueueCount < FNumBuffers div 2) and
         (WaitForSingleObject(FQueueEvent,0) <> WAIT_OBJECT_0) then
      begin
         { reset the done event }
         ResetEvent(FDoneEvent);
         { set the queue event }
         SetEvent(FQueueEvent);
         LeaveCritical;

         if not FStarted then
         begin
            { wait until the queue is filled with some data }

            Time := TimeGetTime;
            bTimeOut := False;

            if (GetCurrentThreadID = MainThreadID) then FMainWaiting := True;

            while (WaitForSingleObject(FDoneEvent,0) <> WAIT_OBJECT_0) do
            try
               { simulated event synchronization }
               if FMainWaiting and assigned(FSyncProc) then
               begin
                  FSyncProc;
                  FSyncProc := nil;
               end;

               if (TimeGetTime - Time >= TimeOut) then
               begin
                  bTimeOut := True;
                  break;
               end;

               Sleep(SLEEPTIME);

            finally
               FMainWaiting := False;
            end;

            if not bTimeOut and (QueueCount > 0) then
               FStarted := True
            else
            begin
               dwSize := 0;
               exit;
            end;
         end;
      end
      else if (FQueueMode = qmDuplex) and not FStarted then
      begin
         {$IFDEF _MMDEBUG}
         DebugStr(0,'Duplex mode waiting for start...'+IntToStr(FQueueCount))
         {$ENDIF};

         GlobalMoveMem(FTempBuffer^.lpData^,lpData^,dwSize);
         dwUser := 0;
         dwFlags := 0;
         if (FQueueCount > FNumBuffers div 2) then