clsocket.pas

Clever_Internet_Suite_6.2的代码 Clever_Internet_Suite_6.2的代码 Clever_Internet_Suite_6.2的代码

begin
  case NetworkStream.NextAction of
    saRead: ReadData(nil);
    saWrite: WriteData(nil);
  end;
end;

{ TclAsyncConnection }

procedure TclAsyncConnection.AcceptConnection;
begin
  NetworkStream.Accept();
  FActive := True;
end;

constructor TclAsyncConnection.Create;
begin
  inherited Create();
  FRefCount := 0;
end;

function TclAsyncConnection._AddRef: Integer;
begin
  Result := InterlockedIncrement(FRefCount);
(*{$IFDEF LOGGER}clPutLogMessage(Self, edInside, '_AddRef, %d', nil, [FRefCount]);{$ENDIF}*)
end;

function TclAsyncConnection._Release: Integer;
begin
  if (Self <> nil) then
  begin
    Result := InterlockedDecrement(FRefCount);
    if (Result = 0) then
    begin
(*      {$IFDEF LOGGER}clPutLogMessage(Self, edInside, '_Release.Destroy');{$ENDIF}*)
      Destroy();
    end;
  end else
  begin
    Result := 0;
  end;
(*{$IFDEF LOGGER}clPutLogMessage(Self, edInside, '_Release, %d', nil, [FRefCount]);{$ENDIF}*)
end;

function TclAsyncConnection.GetPeerIP: string;
begin
  Result := NetworkStream.PeerIP;
end;

function TclAsyncConnection.GetPeerName: string;
begin
  Result := NetworkStream.PeerName;
end;

procedure TclAsyncConnection.ReadData(AData: TStream);
begin
  NetworkStream.Read(AData);
end;

procedure TclAsyncConnection.WriteData(AData: TStream);
begin
  NetworkStream.Write(AData);
end;

procedure TclAsyncConnection.DispatchNextAction;
begin
end;

procedure TclAsyncConnection.AcceptConnectionDone;
begin
  NetworkStream.StreamReady();
end;

procedure TclAsyncConnection.OpenSession;
begin
  NetworkStream.OpenServerSession();
end;

{ TclUdpClientConnection }

function TclUdpClientConnection.GetIP: string;
begin
  Result := NetworkStream.IP;
end;

function TclUdpClientConnection.GetPort: Integer;
begin
  Result := NetworkStream.Port;
end;

procedure TclUdpClientConnection.Open(const AIP: string; APort: Integer);
var
  isReadIntilCloseOld: Boolean;
begin
  isReadIntilCloseOld := IsReadUntilClose;
  IsReadUntilClose := False;
  
  CreateSocket(SOCK_DGRAM, IPPROTO_UDP);
  NetworkStream.Connect(AIP, APort);
  SelectSocketEvent(FD_READ or FD_CLOSE or FD_WRITE);
  FActive := True;
  DispatchNextAction();
  NetworkStream.StreamReady();

  IsReadUntilClose := isReadIntilCloseOld;
end;

{ TclNetworkStream }

procedure TclNetworkStream.Accept;
var
  client_addr: TSockAddrIn;
  hostInfo: PHostEnt;
  addrList: PChar;
  len: Integer;
begin
  ClearNextAction();

  len := SizeOf(client_addr);
  Connection.Socket := winsock.accept(Connection.Socket, PSOCKADDR(@client_addr), @len);
  if (Connection.Socket = INVALID_SOCKET) then
  begin
    RaiseSocketError(WSAGetLastError());
  end;

  hostInfo := gethostbyaddr(@client_addr.sin_addr.S_addr, SizeOf(TSockAddr), AF_INET);
  if (hostInfo <> nil) then
  begin
    addrList := hostInfo^.h_addr_list^;
    FPeerIP := Format('%d.%d.%d.%d',
      [Ord(addrList[0]), Ord(addrList[1]), Ord(addrList[2]), Ord(addrList[3])]);
    FPeerName := Trim(string(hostInfo^.h_name));
    if (PeerName = '') then
    begin
      FPeerName := PeerIP;
    end;
  end;
end;

procedure TclNetworkStream.Close(ANotifyPeer: Boolean);
begin
  ClearNextAction();
end;

constructor TclNetworkStream.Create;
begin
  inherited Create();
  FSleepEvent := CreateEvent(nil, False, False, nil);
end;

destructor TclNetworkStream.Destroy;
begin
  if (FSleepEvent <> INVALID_HANDLE_VALUE) then
  begin
    CloseHandle(FSleepEvent);
    FSleepEvent := INVALID_HANDLE_VALUE;
  end;
  inherited Destroy();
end;

function TclNetworkStream.DoRecv(s: TSocket; var Buf; len, flags: Integer): Integer;
var
  err: Integer;
begin
  Result := winsock.recv(s, Buf, len, flags);
  if Connection.BitsPerSec > 0 then
  begin
    err := WSAGetLastError();
    try
      DoSleep((len * 8 * 1000) div Connection.BitsPerSec);
    finally
      SetLastError(err);
    end;
  end;
end;

function TclNetworkStream.DoSend(s: TSocket; var Buf; len, flags: Integer): Integer;
var
  err: Integer;
begin
  Result := winsock.send(s, Buf, len, flags);
  if Connection.BitsPerSec > 0 then
  begin
    err := WSAGetLastError();
    try
      DoSleep((len * 8 * 1000) div Connection.BitsPerSec);
    finally
      SetLastError(err);
    end;
  end;
end;

procedure TclNetworkStream.DoSleep(AMilliseconds: Integer);
var
  Msg: TMsg;
  res: DWORD;
  events: array[0..0] of THandle;
  sleepTicks: DWORD;
begin
  events[0] := FSleepEvent;
  sleepTicks := GetTickCount();
  repeat
    res := MsgWaitForMultipleObjects(1, events, FALSE, DWORD(AMilliseconds), QS_ALLEVENTS);
    case res of
      WAIT_TIMEOUT,
      WAIT_OBJECT_0:
        begin
          Break;
        end;
      WAIT_OBJECT_0 + 1:
        begin
          while PeekMessage(Msg, 0, 0, 0, PM_REMOVE) do
          begin
            TranslateMessage(Msg);
            DispatchMessage(Msg);
            if Integer(GetTickCount() - sleepTicks) > AMilliseconds then
            begin
              Break;
            end;
          end;
          if Integer(GetTickCount() - sleepTicks) > AMilliseconds then
          begin
            Break;
          end;
        end;
    end;
  until False;
end;

function TclNetworkStream.GetConnection: TclConnection;
begin
  Assert(FConnection <> nil);
  Result := FConnection;
end;

function TclNetworkStream.GetBatchSize: Integer;
begin
  if (Connection.BatchSize < 1) then
  begin
    RaiseSocketError(cBatchSizeInvalid, -1);
  end;

  Result := Connection.BatchSize;
  if (Connection.BytesToProceed > -1) and ((Connection.BytesToProceed - Connection.FBytesProceed) < Result) then
  begin
    Result := (Connection.BytesToProceed - Connection.FBytesProceed);
  end;
end;

procedure TclNetworkStream.Listen(APort: Integer);
var
  sa: TSockAddr;
  srv_address: TSockAddrIn;
  res, useNonblock, saLen: Integer;
begin
  ClearNextAction();

  srv_address.sin_family := AF_INET;
  srv_address.sin_addr.s_addr := INADDR_ANY;
  srv_address.sin_port := htons(APort);

  res := winsock.bind(Connection.Socket, TSockAddr(srv_address), SizeOf(TSockAddrIn));
  if (res = SOCKET_ERROR) then
  begin
    RaiseSocketError(WSAGetLastError());
  end;

  useNonblock := 1;
  winsock.ioctlsocket(Connection.Socket, FIONBIO, useNonblock);

  res := winsock.listen(Connection.Socket, 1);
  if (res = SOCKET_ERROR) then
  begin
    RaiseSocketError(WSAGetLastError());
  end;

  saLen := SizeOf(sa);
  res := winsock.getsockname(Connection.Socket, sa, saLen);
  if (res = SOCKET_ERROR) then
  begin
    RaiseSocketError(WSAGetLastError());
  end;

  FListenPort := ntohs(sa.sin_port);
  FIP := '';
  FPort := APort;
end;

function TclNetworkStream.NeedStop: Boolean;
begin
  Result := Connection.IsAborted or Connection.IsProceedLimit()
end;

function TclNetworkStream.Connect(const AIP: string; APort: Integer): Boolean;
var
  sock_address: TSockAddrIn;
  res, addr: Integer;
begin
  ClearNextAction();

  addr := inet_addr(PChar(AIP));
  if (addr = Integer(INADDR_NONE)) then
  begin
    RaiseSocketError(cInvalidAddress, -1);
  end;

  sock_address.sin_family := AF_INET;

  if (APort <= 0) then
  begin
    RaiseSocketError(cInvalidPort, -1);
  end;
  
  sock_address.sin_port := htons(APort);
  sock_address.sin_addr.S_addr := addr;

  res := winsock.connect(Connection.Socket, TSockAddr(sock_address), sizeof(sock_address));

  Result := (res <> SOCKET_ERROR);
  if not Result and (WSAGetLastError() <> WSAEWOULDBLOCK) then
  begin
    RaiseSocketError(WSAGetLastError());
  end;

  FIP := AIP;
  FPort := APort;
end;

function TclNetworkStream.Read(AData: TStream): Boolean;
var
  bytesRead, toRead: Integer;
  buf: PChar;
begin
  ClearNextAction();
  Result := True;

  toRead := GetBatchSize();
  if (toRead <= 0) then Exit;

  GetMem(buf, toRead);
  if (buf = nil) then
  begin
    RaiseSocketError(GetLastError());
  end;
  try
    repeat
      bytesRead := DoRecv(Connection.Socket, buf^, GetBatchSize(), 0);
      Result := (bytesRead <> SOCKET_ERROR);
      if not Result and (WSAGetLastError() <> WSAEWOULDBLOCK) then
      begin
        RaiseSocketError(WSAGetLastError());
      end;
      
      if Result and (bytesRead > 0) then
      begin
        AData.Write(buf^, bytesRead);
        UpdateProgress(bytesRead);
      end;

      if (bytesRead = 0) then
      begin
        FNeedClose := True;
      end;
    until (not Result) or (not (bytesRead > 0)) or NeedStop();

  finally
    FreeMem(buf)
  end;
end;

procedure TclNetworkStream.UpdateProgress(ABytesProceed: Int64);
begin
  Connection.FBytesProceed := Connection.FBytesProceed + ABytesProceed;
  Connection.FTotalBytesProceed := Connection.FTotalBytesProceed + ABytesProceed;
  Connection.DoProgress(Connection.FTotalBytesProceed, Connection.FTotalBytes);
end;

function TclNetworkStream.Write(AData: TStream): Boolean;
var
  buf: PChar;
  written, toWrite: Integer;
  total: Int64;
begin
  ClearNextAction();
  Result := True;

  if (AData.Size = 0) then Exit;

  toWrite := GetBatchSize();
  if (toWrite <= 0) then Exit;

  GetMem(buf, toWrite);
  if (buf = nil) then
  begin
    RaiseSocketError(GetLastError());
  end;
  try
    total := AData.Position;
    repeat
      toWrite := GetBatchSize();
      if (toWrite > (AData.Size - total)) then
      begin
        toWrite := (AData.Size - total);
      end;
      AData.Read(buf^, toWrite);

{$IFDEF LOGGER}clPutLogMessage(Self, edInside, 'TclNetworkStream.Write');{$ENDIF}
      written := DoSend(Connection.Socket, buf^, toWrite, 0);
      Result := (written <> SOCKET_ERROR);
      if not Result and (WSAGetLastError() <> WSAEWOULDBLOCK) then
      begin
        RaiseSocketError(WSAGetLastError());
      end;

      if Result then
      begin
        total := total + written;
        if (written < toWrite) then
        begin
          AData.Position := AData.Position - toWrite + written;
        end;
        UpdateProgress(written);
      end else
      begin
        AData.Position := AData.Position - toWrite;
        total := AData.Size;
      end;
    until (not Result) or (not (total < AData.Size)) or NeedStop();
  finally
    FreeMem(buf)
  end;
end;

procedure TclNetworkStream.ClearNextAction;
begin
  FNeedClose := False;
  FNextAction := saNone;
end;

procedure TclNetworkStream.SetNextAction(Action: TclNetworkStreamAction);
begin
  if (FNextAction = saNone) then
  begin
    FNextAction := Action;
  end;
end;

procedure TclNetworkStream.StreamReady;
begin
  Connection.DoReady();
end;

procedure TclNetworkStream.OpenClientSession;
begin
end;

procedure TclNetworkStream.OpenServerSession;
begin
end;

procedure TclNetworkStream.Assign(ASource: TclNetworkStream);
begin
  FListenPort := ASource.ListenPort;
  FPeerName := ASource.PeerName;
  FPeerIP := ASource.PeerIP;
  FIP := ASource.IP;
  FPort := ASource.Port;
end;

end.