rtcmsgcliprov.pas

Delphi快速开发Web Server

{
  "Message Client provider" - Copyright (c) Danijel Tkalcec
  @html(<br>)

  @exclude
}
unit rtcMsgCliProv;

{$INCLUDE rtcDefs.inc}

interface

uses
  rtcTrashcan,

  SysUtils,
  Windows,
  Classes,

  rtcSyncObjs,
  rtcThrPool,

  rtcLog,
  rtcInfo,
  rtcConn,
  rtcConnProv,
  rtcThrConnProv,

  rtcFastStrings,
  rtcTransports;

const
  LOG_MSGCLI_EXCEPTIONS:boolean=False;

type
  TRtcMsgClientProvider = class;

  RtcMsgCliException = class(Exception);

  TRtcMsgClientThread = class(TRtcThread)
  public
    RtcConn:TRtcMsgClientProvider;
    Releasing:boolean;

  public
    constructor Create; override;
    destructor Destroy; override;

    function Work(Job:TObject):boolean; override;

    procedure OpenConn;
    procedure CloseConn(_lost:boolean);
    end;

  TRtcMsgClientProvider = class(TRtcThrClientProvider)
  private
    Client_Thread:TRtcMsgClientThread;

    RequestStream, ResponseStream:TMemoryStream;

    FServer:IRTCMessageReceiver;

    Forc:boolean;

    FCS:TRtcCritSec;

    FOnInvalidResponse:TRtcEvent;

    FResponseBuffer:TRtcHugeString;

    FReadBuffer:string;

    FMaxHeaderSize:integer;
    FMaxResponseSize:integer;

    FHeaderOut:boolean;
    LenToWrite:int64;

    FRequest:TRtcClientRequest;
    FResponse:TRtcClientResponse;

    FDataWasSent:boolean;

  protected
    procedure Enter; override;
    procedure Leave; override;

    function GetClientThread:TRtcThread; override;

    procedure TriggerInvalidResponse; virtual;

    procedure AcceptResponse; virtual;

    function _Active:boolean;

    procedure OpenConnection;

  public
    constructor Create; override;
    destructor Destroy; override;

    procedure Connect(Force:boolean=False); override;
    procedure Disconnect; override;
    procedure Release; override;

    procedure InternalDisconnect; override;

    procedure LeavingEvent; virtual;

    procedure SetTriggerInvalidResponse(Event:TRtcEvent);

    procedure WriteHeader; overload; virtual;
    procedure WriteHeader(const Header_Text:string); overload; virtual;

    procedure Write(const s:string; SendNow:boolean=True); override;
    function Read:string; override;

    property Request:TRtcClientRequest read FRequest write FRequest;
    property Response:TRtcClientResponse read FResponse write FResponse;

    // Max. allowed size of the first (status) line in response header
    property MaxResponseSize:integer read FMaxResponseSize write FMaxResponseSize;
    // Max. allowed size of the complete response Header
    property MaxHeaderSize:integer read FMaxHeaderSize write FMaxHeaderSize;

    property Server:IRTCMessageReceiver read FServer write FServer;
    end;

implementation

const
  CRLF = #13#10;
  END_MARK = CRLF+CRLF;

type
  TRtcBaseMessage=class
    end;

var
  Message_WSStop,
  Message_WSRelease,
  Message_WSOpenConn,
  Message_WSCloseConn:TRtcBaseMessage;

{ TRtcMsgClientProvider }

constructor TRtcMsgClientProvider.Create;
  begin
  inherited;

  RequestStream:=TMemoryStream.Create;
  ResponseStream:=TMemoryStream.Create;

  FCS:=TRtcCritSec.Create;

  FResponseBuffer:=TRtcHugeString.Create;

  FDataWasSent:=False;
  SetLength(FReadBuffer,32000);
  end;

destructor TRtcMsgClientProvider.Destroy;
  begin
  Silent:=True;
  Closing:=True;

  InternalDisconnect;

  if assigned(Client_Thread) then
    TRtcThread.PostJob(Client_Thread, Message_WSStop, True);

  FResponseBuffer.Free;
  FResponseBuffer:=nil;

  RequestStream.Free;
  RequestStream:=nil;
  ResponseStream.Free;
  ResponseStream:=nil;

  FReadBuffer:='';
  FCS.Free;

  inherited;
  end;

procedure TRtcMsgClientProvider.Enter;
  begin
  FCS.Enter;
  end;

procedure TRtcMsgClientProvider.Leave;
  begin
  FCS.Leave;
  end;

procedure TRtcMsgClientProvider.SetTriggerInvalidResponse(Event: TRtcEvent);
  begin
  FOnInvalidResponse:=Event;
  end;

procedure TRtcMsgClientProvider.TriggerInvalidResponse;
  begin
  if assigned(FOnInvalidResponse) then
    FOnInvalidResponse;
  end;

function TRtcMsgClientProvider.GetClientThread: TRtcThread;
  begin
  Result:=Client_Thread;
  end;

procedure TRtcMsgClientProvider.Connect(Force: boolean);
  begin
  if assigned(Client_Thread) and not inThread then
    TRtcThread.PostJob(Client_Thread, Message_WSOpenConn)
  else
    begin
    if GetMultiThreaded then
      begin
      if not assigned(Client_Thread) then
        begin
        Client_Thread:=TRtcMsgClientThread.Create;
        Client_Thread.RtcConn:=self;
        end;
      Forc:=Force;
      TRtcThread.PostJob(Client_Thread, Message_WSOpenConn);
      end
    else
      OpenConnection;
    end;
  end;

procedure TRtcMsgClientProvider.OpenConnection;
  begin
  if (State=conActive) or (State=conActivating) then Exit; // already connected !!!

  if State<>conInactive then
    raise Exception.Create('Can not connect again, connection in use.');

  try
    Lost:=True;
    Closing:=False;
    Silent:=False;

    Request.Init;
    Response.Clear;

    State:=conActivating;

    TriggerConnectionOpening(Forc);

    if not assigned(FServer) then
      raise RtcMsgCliException.Create('Error connecting, Server component not assigned!');

    RequestStream.Clear;
    ResponseStream.Clear;

    State:=conActive;

    TriggerConnecting;
    TriggerConnect;
  except
    on E:Exception do
      begin
      TriggerConnectionClosing;
      TriggerConnectError(E);
      TriggerReadyToRelease;
      end;
    end;
  end;

procedure TRtcMsgClientProvider.Disconnect;
  begin
  Lost:=False;
  if assigned(Client_Thread) and not inThread then
    begin
    if TRtcThread.Lock(Client_Thread) then
      try
        TRtcThread.PostJob(Client_Thread, Message_WSCloseConn);
      finally
        TRtcThread.UnLock;
        end;
    end
  else
    InternalDisconnect;
  end;

procedure TRtcMsgClientProvider.InternalDisconnect;
  begin
  if Closing then Exit;

  Closing:=True;

  State:=conClosing;

  RequestStream.Clear;
  ResponseStream.Clear;

  if State=conClosing then
    begin
    TriggerDisconnecting;
    TriggerConnectionClosing;

    State:=conInactive;
    try
      TriggerDisconnect;
      if Lost then
        TriggerConnectLost;
    except
      end;

    FHeaderOut:=False;
    TriggerReadyToRelease;
    end;
  end;

function TRtcMsgClientProvider.Read: string;
  begin
  if not _Active then
    begin
    Result:='';
    Exit;
    end;

  if FResponseBuffer.Size>0 then
    begin
    Result:=FResponseBuffer.Get;
    FResponseBuffer.Clear;
    end
  else
    Result:='';
  end;

procedure TRtcMsgClientProvider.WriteHeader;
  var
    s:string;
  begin
  if not _Active then Exit;

  if FHeaderOut then
    raise Exception.Create('Last header intercepted with new header, before data sent out.');

  s:=Request.Method+' '+Request.URI+' HTTP/1.1'+CRLF+
     Request.HeaderText;

  if Request.Close then s:=s+'Connection: close'+CRLF;

  s:=s+CRLF;

  RequestStream.Write(s[1],length(s));

  FDataOut:=length(s);
  TriggerDataOut;

  Request.Started:=True;
  Request.Active:=True;

  LenToWrite:=Request.ContentLength;

  ResponseStream.Clear;

  FDataWasSent:=True;
  end;

procedure TRtcMsgClientProvider.WriteHeader(const Header_Text: string);
  begin
  if not _Active then Exit;

  Response.HeaderText:=Header_Text;
  WriteHeader;
  end;

procedure TRtcMsgClientProvider.Write(const s: string; SendNow:boolean=True);
  begin
  if not _Active then Exit;

  if s='' then Exit;

  if not Request.Active then
    raise Exception.Create('Sending data without header.');

  RequestStream.Write(s[1], length(s));

  FDataOut:=length(s);
  LenToWrite:=LenToWrite-FDataOut;
  TriggerDataOut;

  FDataWasSent:=True; // will call DataSent
  end;

procedure TRtcMsgClientProvider.LeavingEvent;
  begin
  If _Active and FDataWasSent then
    begin
    FDataWasSent:=False;

    if LenToWrite=0 then
      begin
      Request.Complete:=True;
      TriggerDataSent;
      if Request.Complete and not Response.Done then
        AcceptResponse;
      end
    else
      TriggerDataSent;
    end;
  TriggerReadyToRelease;
  end;

procedure TRtcMsgClientProvider.AcceptResponse;
  var
    s,
    StatusLine,
    HeadStr:string;

    len,len2,
    HeadLen,
    MyPos:integer;

    FChunked,
    FHaveResponse,
    FResponseLine:boolean;

    FChunkState:integer;

    LenToRead:int64;
    InBuffer:string;

  function HexToInt(s:string):integer;
    var
      i,len:integer;
      c:char;
    begin
    Result:=0;
    len:=length(s);
    i:=1;
    while len>0 do
      begin
      c:=s[len];
      if c in ['1'..'9'] then
        Result:=Result+i*(Ord(c)-Ord('0'))
      else if s[len] in ['A'..'F'] then
        Result:=Result+i*(Ord(c)-Ord('A')+10)
      else if s[len] in ['a'..'f'] then
        Result:=Result+i*(Ord(c)-Ord('a')+10);
      i:=i*16;Dec(len);
      end;
    end;

  procedure ResponseError;
    begin
    FResponseLine:=False;
    TriggerInvalidResponse;
    end;

  procedure ClearResponse;
    begin
    FResponseBuffer.Clear;

    FResponseLine:=False;
    FResponse.Clear;
    LenToRead:=-1;
    end;

  procedure ProcessData(const data:string);
    var
      s:string;
      FDone:boolean;
    begin
    FDone:=False;
    InBuffer := InBuffer + data;
    repeat
      if not FHaveResponse then // Don't have the header yet ...
        begin
        if not FResponseLine then
          begin
          // Accept streaming data as response
          if ((length(InBuffer)>=5) and (CompareText(Copy(InBuffer,1,5),'HTTP/')<>0)) or
             ((length(InBuffer)=1) and (CompareText(InBuffer,'H')<>0)) or
             ((length(InBuffer)=2) and (CompareText(InBuffer,'HT')<>0)) or
             ((length(InBuffer)=3) and (CompareText(InBuffer,'HTT')<>0)) or
             ((length(InBuffer)=4) and (CompareText(InBuffer,'HTTP')<>0)) then