communication.pas

《delphi深度编程及其项目开发》

unit Communication;

interface

uses
  Windows, Messages, SysUtils, Variants, Classes, Graphics, Controls, Forms,
  Dialogs, ScktComp, Menus, ComCtrls, StdCtrls, ExtCtrls,
  IdBaseComponent, IdComponent, IdTCPConnection, IdTCPClient, IdTCPServer, DiskList;

//每次传送的最大字节数
const MaxBufferSize = 100;

//返回确认标记
const returnconfirm = '!@#$';
type
  //用于接收数据
  TIdcommserver = class(TIdTCPServer)
    procedure commExecute(AThread: TIdPeerThread);
  end;

type
  //用于发送数据的线程
  TsendThread = class(TThread)
  protected
    procedure execute; override;
  end;

//初始化客户端
function initClientConnect(const host: string; const port: integer; const AFileName: string; const AFilesize: longint): boolean; export;
//关闭客户端
procedure desClientConnect; export;
//客户端向磁盘文件写数据
function ClientWriteBuffer(var Abuffer; const aBuffersize: integer): integer; export;
//初始化服务器端
function initServerConnect(const port: integer; const AFileName: string; const AFilesize: longint): boolean; export;
//断开服务器端连接
procedure desserverconnect; export;
//服务器端从磁盘队列读出数据
function serverreadbuffer(var buffer; var buffersize: longint): boolean; export;

var
  IdCommClient: TIdTcpClient; //客户端Socket组件
  IdCommServer: TIdCommServer; //服务器端socket组件
  clientcommqueue: TcommQueue; //客户端磁盘队列
  servercommqueue: TcommQueue; //服务器端磁盘队列
  recvstream: TMemorystream; //接收的内存流
  sendthread: TsendThread; //发送线程
implementation

//监视发送队列，发现磁盘队列不为空，就进行发送;

procedure TSendThread.execute;
var
  buffersize: longint;
  frontposition, buffsize: longint;
  sendbuffer: pchar;
  i: longint;
  divbuffer: array[1..maxbuffersize + 5] of char;
  recvaccount: integer;
  recvbuffer: array[0..3] of char;
  ss: string;
  connectAgain: boolean;
begin
  connectAgain := false;
  while not terminated do
  begin
    sleep(10);
    //判断磁盘队列是否为空
    if clientCommqueue.empty = 1 then
      continue;
    //判断是否重新连接
    if connectAgain = true then
    begin
      try
        IdCommClient.Disconnect;
        idcommclient.Connect;
      except
        continue;
      end;
    end;
    //得到头指针
    frontposition := clientcommqueue.frontpointer;
    //得到读出队列的大小
    clientcommqueue.readbuffersize(frontposition, buffersize);
    //分配sendBuffer缓冲区
    getmem(sendbuffer, buffersize);
    try
      //读出队列
      clientcommqueue.readQueue(sendbuffer^, buffersize);
      //判断是否段开发送
      if Buffersize > Maxbuffersize then
      begin
        for I := 1 to Buffersize div Maxbuffersize do
        begin
          buffsize := maxbuffersize + 1;
          move(buffsize, divbuffer[1], 4);
          move(sendbuffer[(I - 1) * maxbuffersize], divbuffer[5], maxbuffersize);
          //最后一位代表还未发完，'Y'代表一条队列内容已发完
          divbuffer[maxbuffersize + 5] := 'N';
          try
            IdCommClient.WriteBuffer(divbuffer, maxbuffersize + 5, true);
          except
            connectagain := true;
            continue;
          end;
          sleep(10);
        end;
        buffsize := (buffersize mod maxbuffersize) + 1;
        move(buffsize, divbuffer[1], 4);
        move(sendbuffer[(I - 1) * maxbuffersize], divbuffer[5], Buffersize mod maxbuffersize);
        divbuffer[(buffersize mod maxbuffersize) + 5] := 'Y';
        try
          IdCommClient.WriteBuffer(divbuffer, (Buffersize mod maxbuffersize) + 5, true);
        except
          connectagain := true;
          continue;
        end;
      end
      else
      begin
        buffsize := buffersize + 1;
        move(buffsize, divbuffer[1], 4);
        move(sendbuffer[0], divbuffer[5], buffersize);
        divbuffer[buffersize + 5] := 'Y';
        try
          IdCommClient.WriteBuffer(divbuffer, Buffersize + 5, true);
        except
          connectagain := true;
          continue;
        end;
      end;
      if idcommclient.ReadFromStack(true, 1000, true, nil) <> 4 then
        continue;
      ss := idcommclient.ReadString(4);
      //如果未发送成功，则回写
      if ss <> returnconfirm then
        clientcommqueue.writeQueue(sendbuffer^, buffersize);
    finally
      freemem(sendbuffer);
    end;
  end;
end;

//初始化客户端连接，并且指定磁盘队列；

function initclientconnect(const host: string; const port: integer; const AFileName: string; const AFilesize: longint): boolean;
begin
  //创建客户端的indy组件
  IdCommClient := TIdTCPClient.Create(nil);
  result := true;
  if IdCommClient.Connected = true then
    exit;
  //打开socket，并连接服务器
  IdCommClient.Host := host;
  IdCommClient.Port := port;
  try
    IdCommClient.Connect;
  except
    result := false;
  end;
  //创建客户端磁盘队列
  clientCommQueue := TcommQueue.Create(nil);
  //打开磁盘队列
  if clientcommqueue.open(AFilename, AFilesize) <> 1 then
    result := false;
  //创建发送监送线程
  sendthread := Tsendthread.create(false);
end;

//客户端发送缓冲区，并进行队列处理；

function ClientWriteBuffer(var Abuffer; const aBuffersize: integer): integer;
var
  I: integer;
begin
  result := 1;
  //写队列
  if clientcommqueue.writequeue(abuffer, aBuffersize) <> 1 then
  begin
    result := -1; //写队列出错
    exit;
  end;
end;

//断开客户端连接，并关闭磁盘队列；

procedure desclientconnect;
begin
  //发送监视线程终止
  sendThread.Terminate;
  //断开连接
  if IdCommClient.Connected = true then
    IdCommClient.Disconnect;
  IdCommClient.Free;
  //关闭磁盘队列
  clientcommqueue.close;
end;

//初始化服务器端监听，并且指定接收磁盘队列；

function initserverconnect(const port: integer; const AFileName: string; const AFilesize: longint): boolean;
begin
  //创建服务器端socket
  IdCommServer := TIdcommServer.Create(nil);
  result := true;
  IdCommServer.OnExecute := IdCommServer.commexecute;
  if IdCommServer.active = true then
    exit;
  //打开服务器端socket，并开始监听连接
  IdCommServer.DefaultPort := port;
  try
    IdCommServer.active := true;
  except
    result := false;
  end;
  recvstream := TMemoryStream.create;
  //创建服务器端磁盘队列
  servercommqueue := Tcommqueue.Create(nil);
  //打开磁盘队列
  if servercommqueue.open(AFileName, AFilesize) <> 1 then
    result := false;
end;

//服务器端接收事件

procedure TIdcommserver.commexecute(AThread: TIdPeerThread);
var
  recvsize: integer;
  buffer: pchar;
  recvbuffer: pchar;
begin
  //接收数据，一开始接收发送的数据区的大小
  Athread.Connection.ReadBuffer(recvsize, 4);
  getmem(buffer, recvsize);
  try
    Athread.Connection.Readbuffer(buffer^, recvsize);
    //判断最后一位是'Y'还是'N'，以此代表是否发送完
    if (recvStream.Size = 0) and (buffer[recvsize - 1] = 'Y') then
    begin
      if servercommqueue.writequeue(buffer^, recvsize - 1) = 1 then
        Athread.Connection.Write(returnconfirm);
    end
    else if not assigned(recvstream) and (buffer[recvsize - 1] = 'N') then
    begin
      recvstream.WriteBuffer(buffer^, recvsize - 1);
    end
    else if assigned(recvstream) and (buffer[recvsize - 1] = 'N') then
    begin
      recvstream.WriteBuffer(buffer^, recvsize - 1);
    end
    else if (assigned(recvstream)) and (buffer[recvsize - 1] = 'Y') then
    begin
      recvstream.Write(buffer^, recvsize - 1);
      try
        getmem(recvbuffer, recvstream.size);
        recvstream.Seek(0, soFromBeginning);
        recvstream.ReadBuffer(recvbuffer^, recvstream.Size);
        if servercommqueue.writequeue(recvbuffer^, recvstream.Size) = 1 then
          //回发收到信息
          Athread.Connection.Write(returnconfirm);
        freemem(recvbuffer, recvstream.size);
      finally
        recvstream.Clear;
      end;
    end;
  finally
    freemem(buffer);
  end;
end;

//服务器端读队列

function serverReadBuffer(var buffer; var buffersize: longint): boolean;
begin
  result := true;
  if not servercommqueue.readQueue(buffer, buffersize) = 1 then
    result := false;
end;

//服务器端断开连接

procedure desServerConnect;
begin
  if IdCommServer.Active = true then
    IdCommServer.Active := false;
  IdCommServer.Free;
  recvstream.Free;
    //关闭磁盘队列
  servercommqueue.Free;
end;

end.