fpc/fcl/inc/asyncio.pp

{
    $Id$

    Async_IO: Mananging class for asynchronous input/output
    Copyright (C) 2000 by Sebastian Guenther ([email protected])

    See the file COPYING.FPC, included in this distribution,
    for details about the copyright.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
}

{$MODE objfpc}
{$H+}

unit AsyncIO;

interface

{$i asyncioh.inc}    

type
  TAsyncIOManager = class;

  TAsyncIONotify = procedure(UserData: TObject) of object;

  TAsyncIONotifyInfo = record
    Method: TAsyncIONotify;
    UserData: TObject;
  end;

// -------------------------------------------------------------------
//   TAsyncIOManager
// -------------------------------------------------------------------


  TAsyncIOManager = class
  protected
    DoBreak: Boolean;
    IOdata : TIOdata;
    ReadNotifies, WriteNotifies: array[0..MaxHandle] of TAsyncIONotifyInfo;
    HighestHandle: Integer;
    FTimeout: Integer;
    TimeoutNotify: TAsyncIONotifyInfo;
    procedure CalcHighestHandle(max: Integer);
    procedure ExecuteNotify(const Notify: TAsyncIONotifyInfo);
    function GetHandleAsync(AHandle: Integer): Boolean;
    procedure SetHandleAsync(AHandle: Integer; AValue: Boolean);
  public
    constructor Create;
    procedure Run;
    procedure BreakRun;
    procedure SetReadHandler(AHandle: Integer; AMethod: TAsyncIONotify; AUserData: TObject);
    procedure ClearReadHandler(AHandle: Integer);
    function GetReadHandler(AHandle: Integer): TAsyncIONotify;
    procedure SetWriteHandler(AHandle: Integer; AMethod: TAsyncIONotify; AUserData: TObject);
    procedure ClearWriteHandler(AHandle: Integer);
    function GetWriteHandler(AHandle: Integer): TAsyncIONotify;
    procedure SetTimeoutHandler(AMethod: TAsyncIONotify; AUserData: TObject);
    procedure ClearTimeoutHandler;
    function GetTimeoutHandler: TAsyncIONotify;
    property Timeout: Integer read FTimeout write FTimeout;
    property HandleAsync[AHandle: Integer]: Boolean read GetHandleAsync write SetHandleAsync;
  end;


// -------------------------------------------------------------------
//   Line reader classes
// -------------------------------------------------------------------

  TLineNotify = procedure(const line: String) of object;

  PBoolean = ^Boolean;

  TGenericLineReader = class
  protected
    FDestroyedFlag: PBoolean;
    RealBuffer, FBuffer: PChar;
    FBytesInBuffer: Integer;
    FOnLine: TLineNotify;

    function  Read(var ABuffer; count: Integer): Integer; virtual; abstract;
    procedure NoData; virtual; abstract;

  public
    destructor Destroy; override;
    procedure Run;		// Process as many lines as possible

    property Buffer: PChar read FBuffer;
    property BytesInBuffer: Integer read FBytesInBuffer;
    property OnLine: TLineNotify read FOnLine write FOnLine;
  end;


  TAsyncStreamLineReader = class(TGenericLineReader)
  protected
    FManager: TAsyncIOManager;
    FDataStream: TStream;
    FBlockingStream: THandleStream;
    FOnEOF: TNotifyEvent;

    function  Read(var ABuffer; count: Integer): Integer; override;
    procedure NoData; override;
    procedure StreamDataAvailable(UserData: TObject);
  public
    constructor Create(AManager: TAsyncIOManager; AStream: THandleStream);
    constructor Create(AManager: TAsyncIOManager;
      ADataStream: TStream; ABlockingStream: THandleStream);
    destructor Destroy; override;

    property DataStream: TStream read FDataStream;
    property BlockingStream: THandleStream read FBlockingStream;
    property OnEOF: TNotifyEvent read FOnEOF write FOnEOF;
  end;


// -------------------------------------------------------------------
//   TWriteBuffer
// -------------------------------------------------------------------

  TWriteBuffer = class(TStream)
  protected
    FBuffer: PChar;
    FBytesInBuffer: Integer;
    FOnBufferEmpty: TNotifyEvent;

    function  Seek(Offset: LongInt; Origin: Word): LongInt; override;
    function  Write(const ABuffer; Count: LongInt): LongInt; override;
    function  DoRealWrite(const ABuffer; Count: Integer): Integer; virtual; abstract;
    procedure WritingFailed; virtual; abstract;
    procedure WantWrite; virtual; abstract;
    procedure BufferEmpty; virtual;

    constructor Create;

  public
    EndOfLineMarker: String;

    destructor Destroy; override;
    procedure WriteLine(const line: String);
    procedure Run;		// Write as many data as possible

    property BytesInBuffer: Integer read FBytesInBuffer;
    property OnBufferEmpty: TNotifyEvent read FOnBufferEmpty write FOnBufferEmpty;
  end;


  TAsyncWriteStream = class(TWriteBuffer)
  protected
    FManager: TAsyncIOManager;
    FDataStream: TStream;
    FBlockingStream: THandleStream;

    function  DoRealWrite(const ABuffer; Count: Integer): Integer; override;
    procedure WritingFailed; override;
    procedure WantWrite; override;
    procedure BufferEmpty; override;
    procedure CanWrite(UserData: TObject);
  public
    constructor Create(AManager: TAsyncIOManager; AStream: THandleStream);
    constructor Create(AManager: TAsyncIOManager;
      ADataStream: TStream; ABlockingStream: THandleStream);
    destructor Destroy; override;

    property DataStream: TStream read FDataStream;
    property BlockingStream: THandleStream read FBlockingStream;
  end;



// ===================================================================
// ===================================================================

implementation

uses SysUtils;

{$i asyncio.inc}

// -------------------------------------------------------------------
//   TAsyncIOManager
// -------------------------------------------------------------------

procedure TAsyncIOManager.ExecuteNotify(const Notify: TAsyncIONotifyInfo);
begin
  if Assigned(Notify.Method) then
    Notify.Method(Notify.UserData);
end;

procedure TAsyncIOManager.SetTimeoutHandler(AMethod: TAsyncIONotify; AUserData: TObject);
begin
  TimeoutNotify.Method := AMethod;
  TimeoutNotify.UserData := AUserData;
end;

procedure TAsyncIOManager.ClearTimeoutHandler;
begin
  TimeoutNotify.Method := nil;
end;

function TAsyncIOManager.GetTimeoutHandler: TAsyncIONotify;
begin
  Result := TimeoutNotify.Method;
end;


// -------------------------------------------------------------------
//   TGenericLineReader
// -------------------------------------------------------------------

destructor TGenericLineReader.Destroy;
begin
  if Assigned(FDestroyedFlag) then
    FDestroyedFlag^ := True;
  if Assigned(RealBuffer) then
  begin
    FreeMem(RealBuffer);
    RealBuffer := nil;
  end;
  inherited Destroy;
end;

procedure TGenericLineReader.Run;
var
  NewData: array[0..1023] of Byte;
  p: PChar;
  BytesRead, OldBufSize, CurBytesInBuffer, LastEndOfLine, i, LineLength: Integer;
  line: String;
  FirstRun, DestroyedFlag: Boolean;
begin
  FirstRun := True;
  DestroyedFlag := False;
  while True do
  begin
    BytesRead := Read(NewData, SizeOf(NewData));
    //WriteLn('Linereader: ', BytesRead, ' bytes read');
    if BytesRead <= 0 then begin
      if FirstRun then
        NoData;
      break;
    end;
    FirstRun := False;
    OldBufSize := FBytesInBuffer;

    // Append the new received data to the read buffer
    Inc(FBytesInBuffer, BytesRead);
    ReallocMem(RealBuffer, FBytesInBuffer);
    Move(NewData, RealBuffer[OldBufSize], BytesRead);

    {Process all potential lines in the current buffer. Attention: FBuffer and
     FBytesInBuffer MUST be updated for each line, as they can be accessed from
     within the FOnLine handler!}
    LastEndOfLine := 0;
    if OldBufSize > 0 then
      i := OldBufSize - 1
    else
      i := 0;

    CurBytesInBuffer := FBytesInBuffer;

    while i <= CurBytesInBuffer - 2 do
    begin
      if (RealBuffer[i] = #13) or (RealBuffer[i] = #10) then
      begin
        LineLength := i - LastEndOfLine;
	SetLength(line, LineLength);
	if LineLength > 0 then
	  Move(RealBuffer[LastEndOfLine], line[1], LineLength);

	if (RealBuffer[i] = #13) and (RealBuffer[i + 1] = #10) then
	  Inc(i);
	LastEndOfLine := i + 1;

	if Assigned(FOnLine) then
	begin
	  FBuffer := RealBuffer + LastEndOfLine;
	  FBytesInBuffer := CurBytesInBuffer - LastEndOfLine;
	  FDestroyedFlag := @DestroyedFlag;
	  FOnLine(line);
	  FDestroyedFlag := nil;
	  if DestroyedFlag then
	    exit;
	end;
      end;
      Inc(i);
    end;

    FBytesInBuffer := CurBytesInBuffer;

    if LastEndOfLine > 0 then
    begin
      // Remove all processed lines from the buffer
      Dec(FBytesInBuffer, LastEndOfLine);
      GetMem(p, FBytesInBuffer);
      Move(RealBuffer[LastEndOfLine], p^, FBytesInBuffer);
      FreeMem(RealBuffer);
      RealBuffer := p;
    end;
    FBuffer := RealBuffer;
  end;
end;


// -------------------------------------------------------------------
//   TAsyncStreamLineReader
// -------------------------------------------------------------------

function TAsyncStreamLineReader.Read(var ABuffer; count: Integer): Integer;
begin
  Result := FDataStream.Read(ABuffer, count);
end;

procedure TAsyncStreamLineReader.NoData;
var
  s: String;
begin
  if (FDataStream = FBlockingStream) or (FDataStream.Position = FDataStream.Size) then
  begin

    if (FBytesInBuffer > 0) and Assigned(FOnLine) then
    begin
      if FBuffer[FBytesInBuffer - 1] in [#13, #10] then
        Dec(FBytesInBuffer);
      SetLength(s, FBytesInBuffer);
      Move(FBuffer^, s[1], FBytesInBuffer);
      FOnLine(s);
    end;

    FManager.ClearReadHandler(FBlockingStream.Handle);
    if Assigned(FOnEOF) then
      FOnEOF(Self);
  end;
end;

procedure TAsyncStreamLineReader.StreamDataAvailable(UserData: TObject);
begin
  Run;
end;

constructor TAsyncStreamLineReader.Create(AManager: TAsyncIOManager; AStream: THandleStream);
begin
  Self.Create(AManager, AStream, AStream);
end;

constructor TAsyncStreamLineReader.Create(AManager: TAsyncIOManager;
  ADataStream: TStream; ABlockingStream: THandleStream);
begin
  ASSERT(Assigned(ADataStream) and Assigned(ABlockingStream));

  inherited Create;
  FManager := AManager;
  FDataStream := ADataStream;
  FBlockingStream := ABlockingStream;
  AManager.SetReadHandler(FBlockingStream.Handle, @StreamDataAvailable, nil);
  AManager.HandleAsync[FBlockingStream.Handle] := True;
end;

destructor TAsyncStreamLineReader.Destroy;
var
  Handler: TMethod;
begin
  Handler := TMethod(FManager.GetReadHandler(FBlockingStream.Handle));
  if (Handler.Code = Pointer(@StreamDataAvailable)) and
    (Handler.Data = Pointer(Self)) then
    FManager.ClearReadHandler(FBlockingStream.Handle);
  inherited Destroy;
end;


// -------------------------------------------------------------------
//   TWriteBuffer
// -------------------------------------------------------------------

procedure TWriteBuffer.BufferEmpty;
begin
  if Assigned(FOnBufferEmpty) then
    FOnBufferEmpty(Self);
end;

constructor TWriteBuffer.Create;
begin
  inherited Create;

  FBuffer := nil;
  FBytesInBuffer := 0;
  EndOfLineMarker := #10;
end;

destructor TWriteBuffer.Destroy;
begin
  if Assigned(FBuffer) then
    FreeMem(FBuffer);
  inherited Destroy;
end;

function TWriteBuffer.Seek(Offset: LongInt; Origin: Word): LongInt;
begin
  if ((Offset = 0) and ((Origin = soFromCurrent) or (Origin = soFromEnd))) or
     ((Offset = FBytesInBuffer) and (Origin = soFromBeginning)) then
    Result := FBytesInBuffer
  else
    raise EStreamError.Create('Invalid stream operation');
end;

function TWriteBuffer.Write(const ABuffer; Count: LongInt): LongInt;
begin
  ReallocMem(FBuffer, FBytesInBuffer + Count);
  Move(ABuffer, FBuffer[FBytesInBuffer], Count);
  Inc(FBytesInBuffer, Count);
  WantWrite;
  Result := Count;
end;

procedure TWriteBuffer.WriteLine(const line: String);
var
  s: String;
begin
  s := line + EndOfLineMarker;
  WriteBuffer(s[1], Length(s));
end;

procedure TWriteBuffer.Run;
var
  CurStart, written: Integer;
  NewBuf: PChar;
  failed: Boolean;
begin
  CurStart := 0;
  failed := True;
  repeat
    if FBytesInBuffer = 0 then begin
      BufferEmpty;
      exit;
    end;

    written := DoRealWrite(FBuffer[CurStart], FBytesInBuffer - CurStart);
    if written > 0 then begin
      Inc(CurStart, written);
      failed := False;
      GetMem(NewBuf, FBytesInBuffer - CurStart);
      Move(FBuffer[CurStart], NewBuf[0], FBytesInBuffer - CurStart);
      FreeMem(FBuffer);
      FBuffer := NewBuf;
      Dec(FBytesInBuffer, CurStart);
    end;
  until written <= 0;

  if failed then
    WritingFailed;
end;


// -------------------------------------------------------------------
//   TAsyncWriteStream
// -------------------------------------------------------------------

function TAsyncWriteStream.DoRealWrite(const ABuffer; Count: Integer): Integer;
begin
  Result := FDataStream.Write(ABuffer, count);
end;

procedure TAsyncWriteStream.WritingFailed;
begin
  if FDataStream <> FBlockingStream then
    FManager.ClearWriteHandler(FBlockingStream.Handle);
end;

procedure TAsyncWriteStream.WantWrite;
begin
  FManager.SetWriteHandler(FBlockingStream.Handle, @CanWrite, nil);
end;

procedure TAsyncWriteStream.BufferEmpty;
begin
  FManager.ClearWriteHandler(FBlockingStream.Handle);
  inherited BufferEmpty;
end;

procedure TAsyncWriteStream.CanWrite(UserData: TObject);
begin
  Run;
end;

constructor TAsyncWriteStream.Create(AManager: TAsyncIOManager; AStream: THandleStream);
begin
  Self.Create(AManager, AStream, AStream);
end;

constructor TAsyncWriteStream.Create(AManager: TAsyncIOManager;
  ADataStream: TStream; ABlockingStream: THandleStream);
begin
  ASSERT(Assigned(ADataStream) and Assigned(ABlockingStream));

  inherited Create;
  FManager := AManager;
  FDataStream := ADataStream;
  FBlockingStream := ABlockingStream;
  AManager.HandleAsync[FBlockingStream.Handle] := True;
end;

destructor TAsyncWriteStream.Destroy;
begin
  FManager.ClearWriteHandler(FBlockingStream.Handle);
  inherited Destroy;
end;


end.


{
  $Log$
  Revision 1.1  2000-07-13 06:31:29  michael
  + Initial import

  Revision 1.2  2000/07/09 11:49:05  sg
  * Added methods for reading event handlers in TAsyncIOManager
  * Fixed problems when LineReader gets destroyed while it still was
    parsing input data.

  Revision 1.1  2000/02/18 23:14:48  michael
  + Initial implementation

  Revision 1.1  2000/02/17 22:40:05  sg
  * First version. This unit should go into FCL soon...

}