iggclientpool.pas

通信控件

unit IGgClientPool;

interface

uses
  Windows, SysUtils, Classes, SyncObjs, Controls, RTLConsts, IGgCommonType;

const
  MAX_BLOCK_POOL      = 120;
  SUB_BLOCK_SIZE      = 1024;
  BLOCK_LIFE_SEC      = 16;
  BLOCK_INTE_SEC      = 3;

type

  TBlockStatus = (bsIdle, bsWait, bsOut, bsIn, bsComplete, bsOvertime, bsEmpty);
  TRouteEventV = (rvSans, rvRoute1, rvRoute2, rvRouteOvertime, rvRouteEmpty);

  //块信息描述
  PBlock = ^TBlock;
  TBlock = record
    OutID,InID: Integer;
    Packet: PChar;
    PacketSize: Integer;
    Data: PChar;
    DataSize: Integer;
    DataBit32: Integer;
    SubDataSize: Word;
    SubDataAmount: Word;
    CompleteSize: Word;
    FreeData: Byte;
    Reserved: Byte;
  end;
  //Out块信息描述
  POutItem = ^TOutItem;
  TOutItem = record
    InAttachID: Integer;
    Block: TBlock;
    Status: TBlockStatus;
    OutTime: TDateTime;
    LifeSecond: Integer;
  end;
  TOutItemList = array[0..MAX_BLOCK_POOL-1] of POutItem;
  //In块信息描述
  PInItem = ^TInItem;
  TInItem = record
    OutAttachID: Integer;
    Block: TBlock;
    Status: TBlockStatus;
    InTime: TDateTime;
    LifeSecond: Integer;
  end;
  TInItemList = array[0..MAX_BLOCK_POOL-1] of PInItem;

   { TGGOutPool }

  TGGOutPool = class
  private
    FList: TOutItemList;
    FCount: Integer;
    FBlockID: Integer;
    FEnable : Boolean;

  protected
    procedure Init;
    procedure Clear(FreeData: Boolean=FALSE);
    procedure Reset(var Item: TOutItem);
    function GetIdle: Integer;
    function Rescue(Index: Integer): POutItem;
    
  public
    FLock: TCriticalSection;
    constructor Create;
    destructor Destroy; override;
    property Enable: Boolean read FEnable default FALSE;
    property Count: Integer read FCount default 0;
    property BlockID: Integer read FBlockID default 0;

    function IndexValid(Index: Integer): POutItem;
    function CheckValid(BlockID: Integer): POutItem; overload;
    function CheckValid(OutID: Integer; InID: Integer): POutItem; overload;
    function Exist(BlockID: Integer): Boolean;
    procedure Cancel(BlockID: Integer);
    procedure FormatBlock(var Block: TBlock);
    function Put(InAttachID: Integer; Packet: PChar; PacketSize: Integer; Data: PChar; DataSize: Integer; FreeData: Byte=0): Integer; overload;
    function Put(var Item: TOutItem): Integer; overload;
    function Find(const BlockID: Integer): POutItem;
    function Get(var Item: TOutItem; const BlockID: Integer): Boolean;
    function RequestBlock(OutID: Integer; InID: Integer; InAttachID: Integer; CompleteBit32: Integer; First: Boolean=TRUE): POutItem;
    function SetStatus(const BlockID: Integer; const Status: TBlockStatus=bsIdle): Boolean;
    function GetStatus(const BlockID: Integer): TBlockStatus;
    //function IndexByDynamic(var Index: Integer; var EventV: TBlockEventV): POutItem;

  end;

 { TGGInPool }

  TGGInPool = class
  private
    FList: TInItemList;
    FCount: Integer;
    FBlockID: Integer;
    FEnable : Boolean;

  protected
    procedure Init;
    procedure Clear();
    procedure Reset(var Item: TInItem);
    function GetIdle: Integer;
    function Rescue(Index: Integer): PInItem;

  public
    FLock: TCriticalSection;
    constructor Create;
    destructor Destroy; override;
    property Enable: Boolean read FEnable default FALSE;
    property Count: Integer read FCount default 0;
    property BlockID: Integer read FBlockID default 0;

    function IndexValid(Index: Integer): PInItem;
    function CheckValid(BlockID: Integer): PInItem; overload;
    function CheckValid(InID: Integer; OutID: Integer): PInItem; overload;
    function Exist(BlockID: Integer): Boolean;
    procedure Cancel(BlockID: Integer; FreeData: Boolean=TRUE);
    //procedure FormatBlock(var Block: TGGInBlock);
    function Put(OutAttachID: Integer; OutID: Integer; Packet: PChar; PacketSize: Word; DataSize: Word; DataBit32: Integer;
      SubDataSize: Word; SubDataAmount: Word): Integer; overload;
    function Put(var Item: TInItem): Integer; overload;
    function Find(const BlockID: Integer): PInItem;
    function Get(var Item: TInItem; const BlockID: Integer): Boolean;
    function SetStatus(const BlockID: Integer; const Status: TBlockStatus=bsIdle): Boolean;
    function GetStatus(const BlockID: Integer): TBlockStatus;
    function IndexByDynamic(var Index: Integer): PInItem;
  end;

  { TGGMessagePool }
  
const
  MAX_PACKET_SIZE   = 1280;
  MAX_DESCRIBE_SIZE = 1248;

  MAX_MESSAGE_POOL = 200;
  MESSAGE_LIFE_SEC = 20;
  MESSAGE_SEND_RTY = 4;
  BLOCK_SEND_RTY   = 3;

  mgOrd00 = $00;
  mgOrd01 = $01;
  mgOrd10 = $10;
  mgOrd11 = $11;
  mgOrd30 = $30;
  mgOrd31 = $31;
  mgOrd70 = $70;
  mgOrd71 = $71;
  mgOrdF0 = $F0;
  mgOrdF1 = $F1;
  
  mgOnEv0 = $000;
  mgOnEv1 = $100;

type

  TMessage = record
    RecvID:       Integer;
    Rec:          PChar;
    RecSize:      Integer;
    Data:         PChar;
    DataSize:     Integer;
    CommandV:     Word;
    RespondV:     Word;
    Rule:         Word;
    Reserved:     Word;
    OutID:        Integer;
  end;
  PMessage = ^TMessage;

  TMsgStatus = (msEmpty, msIdle, msSend, msWaitBlock, msSendBlock, msSendSucc, msRouteFail, msOvertime);
  TMsgEventV = (mvSans, mvSend, mvCheckBlock1, mvDelBlock2, mvRouteFail, mvOvertime);
  
  TMessageItem = record
    Msg:          TMessage;
    MsgSequ:      Word;
    Retry:        SmallInt;
    LifeSecond:   SmallInt;
    Reserved:     Word;
    Status:       TMsgStatus;
  end;
  PMessageItem = ^TMessageItem;
  TMessageList = array[0..MAX_MESSAGE_POOL-1] of PMessageItem;

  //消息池类
  TGGMessagePool = class
  private
    FList:   TMessageList;
    FCount:  Integer;
    FLock:   TCriticalSection;
    FSequ:   Word;
    FMsgs:   Word;
    FEnable: Boolean;

  protected
    procedure Init;
    procedure Clear;
    procedure FreeMsg(var Msg: TMessage);
    procedure ResetItem(var PItem: PMessageItem);
  public
    constructor Create;
    destructor Destroy; overload;
    property Count: Integer read FCount default 0;
    property Enable: Boolean read FEnable default FALSE;
    property Sequ: Word read FSequ default 0;
    function Remove(Sequ: Integer): Boolean;
    function Put(var Item: TMessageItem): Integer; overload;
    function Put(var Msg: TMessage): Integer; overload;
    function Put(RecvID: Integer; CommandV: Word; RespondV: Word; var Rec; RecSize: Integer;
      Data: PChar; DataSize: Integer; Rule: Integer): Integer; overload;
    function IndexVaild(var Index: Integer; ExcIndex: Integer): PMessageItem;
    function Find(Sequ: Word): PMessageItem;
    function SendCheck(const PItem: PMessageItem): Boolean;
    function SendOrd(Sequ: Integer; RecvID: Integer; CommandV: Word; Rule: Word = mgOrd10): Integer;
    function SendGet(var PItem: PMessageItem; Sequ: Word; CutRetry: Word=1): Boolean;

    function SetStatus(Sequ: Word; Status: TMsgStatus): Boolean;
    function GetStatus(Sequ: Word): TMsgStatus;
  end;

    { TGGPacketSequ }
  { Comment: the class TGGPacketSequ is remember history receive packet sequence, refuse receive existed packet sequence. }

const
  MAX_RECV_SEQU = 255;

type
  TSequTag = record
    SendID: DWord;
    SendSequ: Word;
    LifeHeart: Word;
  end;
  PSequTag = ^TSequTag;
  TSequTags = array[0..MAX_RECV_SEQU-1] of TSequTag;

  TGGRecvSEQU = class
  private
    FTags: TSequTags;
    FFirst: Integer;
    FLast: Integer;
  protected
    procedure Init;
  public
    constructor Create;
    destructor Destroy; overload;
    procedure Put(SendID: DWord; SendSequ: Word; LifeHeart: Word=4);
    function IsExist(SendID: DWord; SendSequ: Word; CutHeart: Word=1): Boolean;
    procedure Check(CutHeart: Word=1);
  end;


  { TIRoutePool }

const
  MAX_ROUTE_POOL      = 13;
  ROUTE_LIFE_SEC      = 180;  //(60*2)
  ROUTE_INTE_SEC      = 2;
  ROUTE_HEART_SEC     = 47;
  ROUTE_WAIT_SEC      = 12;

type
  TRouteStatus = (rsIdle, rsWait, rsHeart, rsOvertime);

  PRouteItem = ^TRouteItem;
  TRouteItem = record
    AttathIDmod: Integer;
    Route: TRoute;
    LifeSec: Integer;
  end;
  PRouteItemX = ^TRouteItemX;
  TRouteItemX = record
    Item: TRouteItem;
    Next: PRouteItemX;
    Status: TRouteStatus;
  end;
  TRouteItemList = array[0..MAX_ROUTE_POOL-1] of PRouteItemX;

  TIRoutePool = class
  private
    FLock: TCriticalSection;
    FList: TRouteItemList;
    FCount: Integer;
    FCurPItem: PRouteItem;
    FActiveRoute: Boolean;
    FIntervalSec: Integer;
    FEnable: Boolean;

  protected
    procedure Init;
    procedure Clear;
    function RescueX(Index: Integer): PRouteItemX;
    procedure Reset(var Item: TRouteItem);
    procedure ResetX(var PItemX: PRouteItemX);
    function  FreeX(var PItemX: PRouteItemX): PRouteItemX;
    function NextX(var PItemX: PRouteItemX): PRouteItemX;
    function PrevX(var PBeginX: PRouteItemX; const AttachID: Integer): PRouteItemX;
    function NewX(const AttachID: Integer): PRouteItemX;
    function IndexValidBeginX(Index: Integer): PRouteItemX;
    function IndexValidEndX(Index: Integer): PRouteItemX;
    function FindX(const AttachID: Integer): PRouteItemX;
    function GetIntervalSec: Integer;
  public
    constructor Create;
    destructor Destroy; override;

    property ActiveRoute: Boolean read FActiveRoute default FALSE;
    property Count: Integer read FCount default 0;
    property CurPItem: PRouteItem read FCurPItem default nil;
    property IntervalSec: Integer read GetIntervalSec default 0;
    property Enable: Boolean read FEnable;
    
    function Exist(const AttachID: Integer): Boolean;
    procedure Cancel(const AttachID: Integer);
    function Find(const AttachID: Integer): PRouteItem;
    function Get(var Item: TRouteItem; const AttachID: Integer): Boolean; overload;
    function Get(var Route: TRoute; const AttachID: Integer): TRouteStatus; overload;
    function UpdateRoute(AttachID: Integer; FromIP: Integer; FromPort: Word): Boolean;
    function Put(var Route: TRoute): PRouteItemX;overload;
    function Put(const AttachID: Integer): PRouteItemX; overload;
    function IndexByState(var Index: Integer; State: TRouteStatus): PRouteItemX;
    function IndexByNextX(var Index: Integer; PItemX: PRouteItemX=nil): PRouteItemX;
    function IndexByDynamic(var Index: Integer; var EventV: TRouteEventV): PRouteItemX;
    function SetStatus(const AttachID: Integer; const Status: TRouteStatus=rsIdle): Boolean;
    function GetStatus(const AttachID: Integer): TRouteStatus;
    function Put(const AttachID: Integer; var Route: TRoute): Integer; overload;
    procedure AdjustRoute(var Route: TRoute);
    function Route(const AttachID: Integer): TRoute;
    function CurRoute(const AttachID: Integer): Integer;
  end;

const
  DEF_I_STREAM_COUNT  = 37;
  DEF_BUF_POOL_LENGTH = 10;
  BUFFER_HASH_SIZE    = 40;
  MAX_BLOCK_SIZE  = 1024*32;
  MIN_BLOCK_SIZE  = 1024*16;
  DEF_BLOCK_SIZE  = 1024*16;

  //Status Value
  BUF_INIT     = 0;
  BUF_RECORD   = 1;
  BUF_DATA     = 2;
  BUF_TIMEOUT  = 3;
  BUF_FAIL     = 4;
  BUF_STOP     = 5;
  BUF_COMPLETE = 6;
  BUF_DESTROY  = 7;

type

  TIBufferAttr = record
    ID: DWORD;
    SelfID  : DWORD;
    AttachID: DWORD;
    DataSize: Integer;
    NameID: array[0..MAX_I_NAMEID-1] of Char;
    Hash: array[0..BUFFER_HASH_SIZE-1] of Char;
  end;
  PIBufferAttr = ^TIBufferAttr;

  TIBufferDataREQ = record
    ID      : DWORD;
    SelfID  : DWORD;
    Position: Integer;
    ReadSize: WORD;
    RateSize: WORD;
  end;
  PIBufferDataREQ = ^TIBufferDataREQ;

  TIBufferDataSEQU = record
    ID      : DWORD;
    SelfID  : DWORD;
    Position: Integer;
    ReadSize: WORD;
    Reserved: WORD;
    //Data ....
  end;
  PIBufferDataSEQU = ^TIBufferDataSEQU;

  TIBufferCmd = record
    ID     : DWORD;
    SelfID : DWORD;
    Command: WORD;
    Para1  : WORD;
    Para2  : Integer;
    //Describe ...
  end;
  PIBufferCmd = ^TIBufferCmd;

  TBufferErr = (beZero, beParaErr, beSizeErr, beMemoryErr, beBlockPosErr, beRouteErr);

  TIBufferCallback = procedure(AttachID: DWORD; var Rec; RecSize: Integer; PData: Pointer; DataSize, Cmd: WORD) of object;

  TIBuffer = class
  private
    FID: DWORD;
    FSelfID  : DWORD;
    FAttachID: DWORD;
    FDataSize: Integer;
    FData: PChar;
    FPosition: Integer;
    FNameID: string;
    FHash: string;

    FBlock    : PChar;
    FBlockSize: Integer;

    FEnable  : Boolean;
    FWrite   : Boolean;
    FErrCode : TBufferErr;
    FStatus  : Integer;
    FRetry   : Integer;
    FSecTime : Integer;
    FLock: TCriticalSection;
    FCallback: TIBufferCallback;

  protected
    procedure Init;
    procedure Clear;
    procedure SetEnable(const Value: Boolean);
    function CheckWork(): Integer;
  public
    constructor Create(); overload;
    constructor Create(AID, ASelfID, AAttachID: DWORD; AData: PChar; ADataSize: Integer; AWrite: Boolean=TRUE; ABlockSize: Integer=DEF_BLOCK_SIZE); overload;
    destructor Destroy(); override;

    procedure SetPara(AID, ASelfID, AAttachID: DWORD; AData: PChar; ADataSize: Integer; AWrite: Boolean=TRUE; ABlockSize: Integer=DEF_BLOCK_SIZE);
    function CheckProcess(Position: Integer; SubSize: Integer): Integer;
    procedure SetStatus(Status: WORD; Para1: WORD=0; Para2: Integer=0; Send: Boolean=FALSE);
    procedure Command(Cmd: Word; Para1: Word; Para2: Integer);
    function ReadREQ(): Integer;
    function Read(var DataREQ: PIBufferDataREQ): Integer; overload;
    function WriteREQ(): Integer;
    function Write(var DataSEQU: PIBufferDataSEQU; PData: Pointer; DataSize: Integer): Integer;
    procedure AutoAdjust(Reserved: Integer=0);
    function Query(var Res: Integer; PRec: Pointer=nil; Cmd: DWORD=0): Integer;

    property Enable: Boolean read FEnable write SetEnable;
    property ID: DWORD read FID;
    property AttachID: DWORD read FAttachID;
    property Status: Integer read FStatus;
    property Data: PChar read FData;
    property DataSize: Integer read FDataSize;
    property NameID: string read FNameID write FNameID;
    property Callback: TIBufferCallback write FCallback;
    property Work: Integer read CheckWork;
  end;

  TIBufferList = array of TIBuffer;
  
  TIBufferPool = class
  private
    FList: TIBufferList;
    FCount: Integer;
    FCapacity: Integer;
    FLock: TCriticalSection;

  protected
    procedure Init;
    procedure Clear;
    procedure SetCapacity(const Value: Integer);
  public
    constructor Create(ACapacity: Integer=DEF_BUF_POOL_LENGTH);
    destructor Destroy(); override;

    function IdleSlot(): Integer;
    function IndexOf(const IBuffer: TIBuffer): Integer; overload;
    function IndexOf(const ID: DWORD): Integer; overload;
    procedure Put(Index: Integer; IBuffer: TIBuffer);
    function Add(IBuffer: TIBuffer): Integer;
    procedure Delete(const Index: Integer); overload;
    class procedure Delete(var IBuffer: TIBuffer); overload;
    procedure Remove(const ID: DWORD);
    function Get(const ID: DWORD): TIBuffer;
    function SearchIn(const Index: Integer): TIBuffer;

    property Count: Integer read FCount;
    property Capacity: Integer read FCapacity write SetCapacity;
  end;

  { TIStreamPool }

  TIStream = record
    AttachID: DWORD;
    NameID  : array[0..MAX_I_NAMEID-1] of Char;
    ID      : DWORD;
    Memory  : TMemoryStream;
  end;
  PIStream = ^TIStream;

  TIStreamItem = record
    Stream : TIStream;
    Status : Integer;
    LifeSec: Integer;
  end;
  PIStreamItem = ^TIStreamItem;

  TIStreamList = array of TIStreamItem;

  TIStreamPool = class
  private
    FList  : TIStreamList;
    FItem  : TIStreamItem;
    FCount : Integer;
    FCapacity: Integer;
    FEnable: Boolean;

  protected
    procedure Init;
    function Get(Index: Integer): TIStreamItem;
    procedure Grow; virtual;
    procedure Put(Index: Integer; Item: TIStreamItem);
    procedure SetCapacity(NewCapacity: Integer);
    procedure SetCount(NewCount: Integer);
  public
    constructor Create(ACapacity: Integer=DEF_I_STREAM_COUNT);
    destructor Destroy(); override;

    function Add(var Item: TIStreamItem): Integer;
    procedure Clear; virtual;
    procedure Delete(Index: Integer);
    class procedure Error(const Msg: string; Data: Integer); overload; virtual;
    class procedure Error(Msg: PResStringRec; Data: Integer); overload;