psitcpconnection.pas

一个delphi的p2p控件的源代码

unit PsiTCPConnection;

//******************************************************************************
// The original software is under
// Copyright (c) 1993 - 2000, Chad Z. Hower (Kudzu)
//   and the Indy Pit Crew - http://www.nevrona.com/Indy/
//
// Amended : November 2000, by Michael M. Michalak MACS for use with
// MorphTek.com Inc Peer to Peer Open Source Components - http://www.morphtek.com
//
//******************************************************************************

interface

uses
  Classes, PsiException,
  PsiComponent, PsiGlobal, PsiSocketHandle, PsiIntercept;

type
  TPsiBuffer = class(TMemoryStream)
  public
    procedure RemoveXBytes(const AByteCount: integer);
  end;

  TPsiTCPConnection = class(TPsiComponent)
  protected
  	FASCIIFilter: boolean;
    FBinding: TPsiSocketHandle;
    FBuffer: TPsiBuffer;
    // TODO - Change the "move" functions to read write functinos. Get as much as possible down
    // to just TStream so we can replace it easily
  	FClosedGracefully: boolean;
    FCmdResultDetails: TStrings;
    FIntercept: TPsiConnectionIntercept;
    FInterceptEnabled: Boolean;
    FOnDisconnected: TNotifyEvent;
    FReadLnTimedOut: Boolean;
  	FRecvBuffer: TPsiBuffer; // To be used by ReadFromStack only
    FResultNo: SmallInt;
    FWriteBuffer: TPsiBuffer;
    FWriteBufferThreshhold: Integer;
    //
    procedure DoOnDisconnected; virtual;
    function GetCmdResult: string;
    function GetRecvBufferSize: Integer;
    procedure Notification(AComponent: TComponent; Operation: TOperation); override;
    procedure ResetConnection; virtual;
    procedure SetIntercept(AValue: TPsiConnectionIntercept);
    procedure SetInterceptEnabled(AValue: Boolean);
    procedure SetRecvBufferSize(const Value: Integer);
  public
    function AllData: string; virtual;
    procedure CancelWriteBuffer;
    procedure Capture(ADest: TObject; const ADelim: string = '.'; const AIsRFCMessage: Boolean = False);
    procedure CheckForDisconnect(const ARaiseExceptionIfDisconnected: boolean = true;
     const AIgnoreBuffer: boolean = false); virtual;
    procedure CheckForGracefulDisconnect(const ARaiseExceptionIfDisconnected: boolean = true); virtual;
    procedure ClearWriteBuffer;
    procedure CloseWriteBuffer;
    function Connected: boolean; virtual;
    constructor Create(AOwner: TComponent); override;
    function CurrentReadBuffer: string;
    function CurrentReadBufferSize: integer;
    destructor Destroy; override;
    procedure Disconnect; virtual;
    procedure DisconnectSocket; virtual;
    function ExtractXBytesFromBuffer(const AByteCount: Integer): string; virtual;
    procedure FlushWriteBuffer(const AByteCount: Integer = -1);
    function GetResponse: SmallInt; virtual;
    function InputLn(const AMask: string = ''): string;
    procedure OpenWriteBuffer(const AThreshhold: Integer = -1);
    // RaiseExceptionForCmdResult - Overload necesary as a exception as a default param doesnt work
    procedure RaiseExceptionForCmdResult; overload; virtual;
    procedure RaiseExceptionForCmdResult(axException: TClassPsiException); overload; virtual;
    procedure ReadBuffer(var ABuffer; const AByteCount: Longint);
    // ReadFromStack must be only call to Recv
    function ReadFromStack(const ARaiseExceptionIfDisconnected: boolean = true;
     const ATimeout: integer = PsiTimeoutInfinite; const AUseBuffer: boolean = true;
     ADestStream: TPsiBuffer = nil): integer;
     virtual;
    function ReadInteger(const AConvert: boolean = true): Integer;
    function ReadLn(const ATerminator: string = '';
     const ATimeout: integer = PsiTimeoutInfinite): string; virtual;
    function ReadLnWait: string;
    function ReadSmallInt: SmallInt;
    procedure ReadStream(AStream: TStream; AByteCount: LongInt = -1;
     const AReadUntilDisconnect: boolean = false);
    function ReadString(const ABytes: integer): string;
    procedure RemoveXBytesFromBuffer(const AByteCount: Integer); virtual;
    function SendCmd(const AOut: string; const AResponse: SmallInt = -1): SmallInt; overload; virtual;
    function SendCmd(const AOut: string; const AResponse: Array of SmallInt): SmallInt; overload; virtual;
    procedure Write(AOut: string); virtual;
    // WriteBuffer must be the ONLY call to SEND - all data goes thru this method
    procedure WriteBuffer(const ABuffer; AByteCount: Longint; const AWriteNow: boolean = false);
    procedure WriteHeader(axHeader: TStrings);
    procedure WriteInteger(AValue: Integer; const AConvert: boolean = true);
    procedure WriteLn(const AOut: string = ''); virtual;
    procedure WriteSmallInt(AValue: SmallInt);
    procedure WriteStream(AStream: TStream; const AAll: boolean = true;
     const AWriteByteCount: Boolean = false); virtual;
    procedure WriteStrings(AValue: TStrings);
    function WriteFile(AFile: String; const AEnableTransferFile: boolean = false): cardinal;
     virtual;
    //
    property Binding: TPsiSocketHandle read FBinding;
  	property ClosedGracefully: boolean read FClosedGracefully;
    property CmdResult: string read GetCmdResult;
    property CmdResultDetails: TStrings read FCmdResultDetails;
    property ReadLnTimedOut: Boolean read FReadLnTimedOut;
    property ResultNo: SmallInt read FResultNo;
  published
  	property ASCIIFilter: boolean read FASCIIFilter write FASCIIFilter;
    property Intercept: TPsiConnectionIntercept read FIntercept write SetIntercept;
    property InterceptEnabled: Boolean read FInterceptEnabled write SetInterceptEnabled;
    property OnDisconnected: TNotifyEvent read FOnDisconnected write FOnDisconnected;
    property OnWork;
    property OnWorkBegin;
    property OnWorkEnd;
    property RecvBufferSize: Integer read GetRecvBufferSize write SetRecvBufferSize;
  end;

implementation

uses
  PsiAntiFreezeBase,
  PsiStack, PsiStackConsts, PsiResourceStrings,
  SysUtils;

function TPsiTCPConnection.AllData: string;
begin
  BeginWork(wmRead); try
    result := '';
    while Connected do begin
      Result := Result + CurrentReadBuffer;
    end;
  finally EndWork(wmRead); end;
end;

procedure TPsiTCPConnection.Capture(ADest: TObject; const ADelim: string = '.';
  const AIsRFCMessage: Boolean = False);
var
  s: string;
begin
  BeginWork(wmRead); try
    repeat
      s := ReadLn;
      if s = ADelim then begin
        exit;
      end;

      // For RFC 822 retrieves
      if AIsRFCMessage and (Copy(s, 1, 2) = '..') then begin
        Delete(s, 1, 1);
      end;

      // Write to output
      if ADest is TStrings then begin
        TStrings(ADest).Add(s);
      end else if ADest is TStream then begin
        TStream(ADest).WriteBuffer(s[1], Length(s));
        s := EOL;
        TStream(ADest).WriteBuffer(s[1], Length(s));
      end else begin
        raise EPsiException.Create(RSObjectTypeNotSupported);
      end;
    until false;
  finally EndWork(wmRead); end;
end;

procedure TPsiTCPConnection.CheckForDisconnect(const ARaiseExceptionIfDisconnected: boolean = true;
 const AIgnoreBuffer: boolean = false);
begin
  if ClosedGracefully then begin
    if Binding.HandleAllocated then begin
    	DisconnectSocket;
    end;
    // Do not raise unless all data has been read by the user
	  if ((CurrentReadBufferSize = 0) or AIgnoreBuffer) and ARaiseExceptionIfDisconnected then begin
      // ************************************************************* //
      // An exception may occur here. This only happens in the IDE and this is normal.
      // Winshoes will handle this. This does not happen in the EXE.
      // ************************************************************* //
      raise EPsiConnClosedGraceful.Create(RSConnectionClosedGracefully);
    end;
  end;
end;

function TPsiTCPConnection.Connected: boolean;
begin
  CheckForDisconnect(False);
  result := Binding.HandleAllocated;
end;

constructor TPsiTCPConnection.Create(AOwner: TComponent);
begin
  inherited;
  FBinding := TPsiSocketHandle.Create(nil);
  FCmdResultDetails := TStringList.Create;
 	FRecvBuffer := TPsiBuffer.Create;
  RecvBufferSize := 8192;
  FBuffer := TPsiBuffer.Create;
end;

function TPsiTCPConnection.CurrentReadBuffer: string;
begin
  result := '';
  if Connected then begin
    ReadFromStack(False);
    result := ExtractXBytesFromBuffer(FBuffer.Size);
  end;
end;

function TPsiTCPConnection.CurrentReadBufferSize: integer;
begin
  result := FBuffer.Size;
end;

destructor TPsiTCPConnection.Destroy;
begin
  FreeAndNil(FBuffer);
 	FreeAndNil(FRecvBuffer);
  FreeAndNil(FCmdResultDetails);
  FreeAndNil(FBinding);
  inherited;
end;

procedure TPsiTCPConnection.Disconnect;
begin
  DisconnectSocket;
end;

procedure TPsiTCPConnection.DoOnDisconnected;
begin
  if assigned(OnDisconnected) then begin
    OnDisconnected(Self);
  end;
end;

function TPsiTCPConnection.ExtractXBytesFromBuffer(const AByteCount: Integer): string;
begin
  if AByteCount > FBuffer.Size then begin
    raise EPsiException.Create(RSNotEnoughDataInBuffer);
  end;
  SetString(result, PChar(FBuffer.Memory), AByteCount);
  RemoveXBytesFromBuffer(AByteCount);
  DoWork(wmRead, AByteCount);
end;

function TPsiTCPConnection.GetCmdResult: string;
begin
	result := '';
	if CmdResultDetails.Count > 0 then begin
		result := CmdResultDetails[CmdResultDetails.Count - 1];
  end;
end;

function TPsiTCPConnection.GetRecvBufferSize: Integer;
begin
	result := FRecvBuffer.Size;
end;

function TPsiTCPConnection.GetResponse: SmallInt;
var
  sLine, sTerm: string;
begin
  CmdResultDetails.Clear;
  sLine := ReadLnWait;
  CmdResultDetails.Add(sLine);
  if length(sLine) > 3 then begin
    if sLine[4] = '-' then begin // Multi line response coming
      sTerm := Copy(sLine, 1, 3) + ' ';
      {We keep reading lines until we encounter either a line such as "250"
       or "250 Read"}
      repeat
        sLine := ReadLnWait;
        CmdResultDetails.Add(sLine);
      until (Length(sLine) < 4) or (AnsiSameText(Copy(sLine, 1, 4), sTerm));
    end;
  end;

  if AnsiSameText(Copy(CmdResult, 1, 3), '+OK') then begin
    FResultNo := wsOK;
  end else if AnsiSameText(Copy(CmdResult, 1, 4), '-ERR') then begin
    FResultNo := wsErr;
  end else begin
    FResultNo := StrToIntDef(Copy(CmdResult, 1, 3), 0);
  end;

  Result := ResultNo;
end;

procedure TPsiTCPConnection.RaiseExceptionForCmdResult(axException: TClassPsiException);
begin
  raise axException.Create(CmdResult);
end;

procedure TPsiTCPConnection.RaiseExceptionForCmdResult;
begin
  EPsiProtocolReplyError.CreateError(ResultNo, CmdResult);
end;

procedure TPsiTCPConnection.ReadBuffer(var ABuffer; const AByteCount: Integer);
begin
	if (AByteCount > 0) and (@ABuffer <> nil) then begin
    // Read from stack until we have enough data
    while CurrentReadBufferSize < AByteCount do begin
      ReadFromStack;
    end;
    // Copy it to the callers buffer
    Move(PChar(FBuffer.Memory)[0], ABuffer, AByteCount);
    // Remove used data from buffer
    RemoveXBytesFromBuffer(AByteCount);
  end;
end;

function TPsiTCPConnection.ReadFromStack(const ARaiseExceptionIfDisconnected: boolean = true;
 const ATimeout: integer = PsiTimeoutInfinite; const AUseBuffer: boolean = true;
 ADestStream: TPsiBuffer = nil): integer;
// Reads any data in tcp/ip buffer and puts it into Winshoe buffer
// This must be the ONLY raw read from Winsock routine
// This must be the ONLY call to RECV - all data goes thru this method
var
  nByteCount, j: Integer;

  procedure DefaultRecv;
  begin
    // No need to call AntiFreeze, the Readable does that.
    nByteCount := Binding.Recv(ADestStream.Memory^, ADestStream.Size, 0);
  end;

begin
  result := 0;
  // Check here as this side may have closed the socket
  CheckForDisconnect(ARaiseExceptionIfDisconnected);
  if Connected then begin
    if ADestStream = nil then begin
      ADestStream := FRecvBuffer;
    end;
    if Binding.Readable(ATimeout) then begin
      if InterceptEnabled then begin
        if Intercept.RecvHandling then begin
          nByteCount := Intercept.Recv(ADestStream.Memory^, ADestStream.Size);
        end else begin
          DefaultRecv;
        end;
      end else begin
        DefaultRecv;
      end;

      FClosedGracefully := nByteCount = 0;
      if not ClosedGracefully then begin
        if GStack.CheckForSocketError(nByteCount, [Psi_WSAESHUTDOWN]) then begin
          nByteCount := 0;
          if Binding.HandleAllocated then begin
            DisconnectSocket;
          end;
          // Do not raise unless all data has been read by the user
          if CurrentReadBufferSize = 0 then begin
            GStack.RaiseSocketError(Psi_WSAESHUTDOWN);
          end;
        end;
        if ASCIIFilter then begin
          for j := 1 to nByteCount do begin
            PChar(ADestStream.Memory)[j] := Chr(Ord(PChar(ADestStream.Memory)[j]) and $7F);
          end;
        end;
      end;
      if AUseBuffer then begin
        FBuffer.Position := FBuffer.Size;
        FBuffer.WriteBuffer(ADestStream.Memory^, nByteCount);
      end else begin
        // If buffered, DoWork is called when bytes are removed from the buffer
        DoWork(wmRead, nByteCount);
      end;
      if InterceptEnabled then begin
        Intercept.DataReceived(ADestStream.Memory^, nByteCount);
      end;
      // Check here as other side may have closed connection
      CheckForDisconnect(ARaiseExceptionIfDisconnected);
      result := nByteCount;
    end;
  end;
end;

function TPsiTCPConnection.ReadInteger(const AConvert: boolean = true): Integer;
begin
	ReadBuffer(Result, SizeOf(Result));
  if AConvert then begin
    Result := Integer(GStack.WSNToHL(LongWord(Result)));
  end;
end;

function TPsiTCPConnection.ReadLn(const ATerminator: string = '';
 const ATimeout: integer = PsiTimeoutInfinite): string;
var
  i: Integer;
  s: string;
  LTerminator: string;
begin
  if Length(ATerminator) = 0 then begin
    LTerminator := LF;
  end else begin
    LTerminator := ATerminator;
  end;
  FReadLnTimedOut := False;
 	i := 0;
  repeat
    if CurrentReadBufferSize > 0 then begin
	    {TODO make searches more efficient, do not copy to a string, search mem directly}
      SetString(s, PChar(FBuffer.Memory), FBuffer.Size);
      i := AnsiPos(LTerminator, s);
    end;
    // ReadFromStack blocks - do not call unless we need to