fpsockets.pas

unit fpsockets;

{$mode ObjFPC}{$H+}
{$TypedAddress on}

interface

uses
  SysUtils, Sockets, Nullable;

type

  { Basic Socket Types }

  TFPSocketType = (stIPv4, stIPv6, stIPDualStack, stUnixSocket);
  TFPSocketProto = (spStream, spDatagram);
  TFPSocket = record
    FD: TSocket;
    Protocol: TFPSocketProto;
    SocketType: TFPSocketType;
  end;

  TAddressType = (atIN4, atIN6, atUnixSock);
  TNetworkAddress = record
    Address: String;
    AddressType: TAddressType;
  end;

  TFPSocketConnection = record
    ClientAddress: TNetworkAddress;
    ClientPort: Word;
    Socket: TFPSocket;
  end;

  { ReceiveFrom Return Types }

  TReceiveFromResult = record
    FromAddr: TNetworkAddress;
    FromPort: Word;
    DataSize: SizeInt;
  end;

  generic TReceiveFromMessage<T> = record
    FromAddr: TNetworkAddress;
    FromPort: Word;
    Data: T;
  end;

  TReceiveFromStringMessage = specialize TReceiveFromMessage<String>;

  { State Management }

  TConnectionState = (csError, csNotConnected, csRefused, csPending, csConnected);

  { Exceptions }

  EDualStackNotSupported = class(Exception);
  EUnsupportedAddress = class(Exception);

  { ESocketError }

  ESocketError = class(Exception)
  private
    FCode: Integer;
  public
    constructor Create(ACode: Integer; const FunName: String);

    property Code: Integer read FCode;
  end;

  EConnectionClosedException = class(Exception);

  { EFragmentedData }

  EFragmentedData = class(Exception)
  private
    FFragment: TBytes;
    FExpectedSize: SizeInt;
  public
    constructor Create(const AFragment: TBytes; AExpected: SizeInt; const AMessage: String);

    property Fragment: TBytes read FFragment;
    property ExpectedSize: SizeInt read FExpectedSize;
  end;

const
  MaxUDPPackageSize = 512;

  { Address Management }

function isIPv4Address(const Address: String): Boolean; inline;
function isIPv6Address(const Address: String): Boolean; inline;

function IN4Address(const Address: String): TNetworkAddress; inline;
function IN6Address(const Address: String): TNetworkAddress; inline;
function IN4MappedIN6Address(const In4Address: String): TNetworkAddress; inline;
function UnixAddr(const Address: String): TNetworkAddress; inline;
function NetAddr(const Address: String): TNetworkAddress; inline;

function isINAddr(const AAddr: TNetworkAddress): Boolean; inline;
function IsIPv4Mapped(const IPv6Addr: TNetworkAddress): Boolean; inline;
function ExtractIPv4Address(const IPv6Addr: TNetworkAddress): TNetworkAddress; inline;

function IN6Equal(const A, B: String): Boolean;
operator =(const A, B: TNetworkAddress): Boolean; inline;
operator <>(const A, B: TNetworkAddress): Boolean; inline;
operator :=(const AStr: String): TNetworkAddress; inline;

  { Socket Functions }

function TCPSocket(AType: TFPSocketType): TFPSocket; inline;
function UDPSocket(AType: TFPSocketType): TFPSocket; inline;

procedure CloseSocket(const ASocket: TFPSocket); inline;

procedure Bind(const ASocket: TFPSocket; const AAddress: TNetworkAddress; APort: Word; ReuseAddr: Boolean = True);

procedure Listen(const ASocket: TFPSocket; Backlog: Integer); inline;
function AcceptConnection(const ASocket: TFPSocket): TFPSocketConnection; inline;
function AcceptNonBlocking(const ASocket: TFPSocket): specialize TNullable<TFPSocketConnection>; inline;

function Connect(const ASocket: TFPSocket; const AAddress: TNetworkAddress; APort: Word): TConnectionState; inline;

function Receive(const ASocket: TFPSocket; ABuffer: Pointer; MaxSize: SizeInt; AFlags: Integer = 0): SizeInt; inline;
function ReceiveFrom(const ASocket: TFPSocket; ABuffer: Pointer; MaxSize: SizeInt; AFlags: Integer = 0): TReceiveFromResult;
function ReceiveFromNonBlocking(const ASocket: TFPSocket; ABuffer: Pointer; MaxSize: SizeInt; AFlags: Integer = 0): specialize TNullable<TReceiveFromResult>; inline;
function Send(const ASocket: TFPSocket; ABuffer: Pointer; ASize: SizeInt; AFlags: Integer = 0): SizeInt; inline;
function SendTo(const ASocket: TFPSocket; const ReceiverAddr: TNetworkAddress;
                  ReceiverPort: Word; ABuffer: Pointer; ASize: SizeInt; AFlags: Integer = 0): SizeInt; inline;

function ReceiveStr(const ASocket: TFPSocket; MaxLength: SizeInt = -1; AFlags: Integer = 0): String;
function ReceiveStrFrom(const ASocket: TFPSocket; MaxLength: SizeInt = MaxUDPPackageSize; AFlags: Integer = 0): TReceiveFromStringMessage; inline;
function ReceiveStrFromNonBlocking(const ASocket: TFPSocket; MaxLength: SizeInt = MaxUDPPackageSize; AFlags: Integer = 0): specialize TNullable<TReceiveFromStringMessage>; inline;
function SendStr(const ASocket: TFPSocket; const AData: String; AFlags: Integer = 0): SizeInt; inline;
function SendStrTo(const ASocket: TFPSocket; const ReceiverAddr: TNetworkAddress; ReceiverPort: Word; const AData: String; AFlags: Integer = 0): SizeInt; inline;

generic function Receive<T>(const ASocket: TFPSocket; AFlags: Integer = 0): T;
generic function ReceiveNonBlocking<T>(const ASocket: TFPSocket; AFlags: Integer = 0): specialize TNullable<T>;
generic function ReceiveFrom<T>(const ASocket: TFPSocket; AFlags: Integer = 0): specialize TReceiveFromMessage<T>;
generic function ReceiveFromNonBlocking<T>(const ASocket: TFPSocket; AFlags: Integer = 0): specialize TNullable<specialize TReceiveFromMessage<T>>;
generic function Send<T>(const ASocket: TFPSocket; constref AData: T; AFlags: Integer = 0): SizeInt; inline;
generic function SendTo<T>(const ASocket: TFPSocket; constref ReceiverAddr: TNetworkAddress; ReceiverPort: Word; const AData: T; AFlags: Integer = 0): SizeInt; inline;

generic function ReceiveArray<T>(const ASocket: TFPSocket; MaxCount: SizeInt = -1; AFlags: Integer = 0): specialize TArray<T>;
generic function ReceiveArrayFrom<T>(const ASocket: TFPSocket; MaxCount: SizeInt = -1; AFlags: Integer = 0): specialize TReceiveFromMessage<specialize TArray<T>>; inline;
generic function ReceiveArrayFromNonBlocking<T>(const ASocket: TFPSocket; MaxCount: SizeInt = -1; AFlags: Integer = 0): specialize TNullable<specialize TReceiveFromMessage<specialize TArray<T>>>; inline;
generic function SendArray<T>(const ASocket: TFPSocket; const AData: specialize TArray<T>; AFlags: Integer = 0): SizeInt; inline;
generic function SendArrayTo<T>(const ASocket: TFPSocket; const ReceiverAddr: TNetworkAddress; ReceiverPort: Word; const AData: specialize TArray<T>; AFlags: Integer = 0): SizeInt; inline;

  { Socket/Connection State Management }

procedure SetNonBlocking(const ASocket: TFPSocket; AValue: Boolean);

// Timeout in MS
function DataAvailable(const SocketArray: specialize TArray<TFPSocket>; TimeOut: Integer = 0): specialize TArray<TFPSocket>; overload;
function DataAvailable(const ASocket: TFPSocket; TimeOut: Integer = 0): Boolean; overload; //inline;
function DataAvailable(const SocketArray: array of TFPSocket; TimeOut: Integer = 0): specialize TArray<TFPSocket>; overload; inline;

function BytesAvailable(const ASocket: TFPSocket): SizeInt;

function StreamClosed(const ASocket: TFPSocket): Boolean; inline;

// For non blocking connections, connect will return a pending connection that needs to be checked
// Note: csConnected means that connection was establised at least once
// If it has been closed by the other side, it is still csConnected, use StreamClosed to figure out
// if the stream is actually open
function ConnectionState(const ASocket: TFPSocket): TConnectionState;

implementation

uses
  {$IfDef WINDOWS}WinSock2{$Else}BaseUnix, termio{$EndIf}, Math;

{ Helper }

type
  _PAddressUnion = ^_TAddressUnion;
  _TAddressUnion = record
  case TFPSocketType of
  stIPv4: (In4Addr: sockets.sockaddr_in);
  stIPv6: (In6Addr: sockets.sockaddr_in6);
  stUnixSocket: (UnixAddr: sockets.sockaddr_un);
  end;

const
  IPPROTO_IPV6 = {$IfDef WINDOWS}41{$Else}41{$EndIf};
  IPV6_V6ONLY = {$IfDef WINDOWS}27{$Else}26{$EndIf};

function WouldBlock(SockErr: Integer): Boolean; inline;
begin
  Result := (SockErr = EsockEWOULDBLOCK)
            {$IfDef Unix} or (SockErr = ESysEAGAIN) {$EndIf}
end;

function CreateAddr(AAddress: TNetworkAddress; APort: Word; DualStack: Boolean): _TAddressUnion;
begin
  if (AAddress.AddressType = atIN4) and DualStack then
    AAddress := IN4MappedIN6Address(AAddress.Address);
  if AAddress.AddressType = atIN4 then
  begin
    Result.In4Addr.sin_family := AF_INET;
    Result.In4Addr.sin_port := HToNS(APort);
    Result.In4Addr.sin_addr.s_addr := LongWord(StrToNetAddr(AAddress.Address));
  end
  else if AAddress.AddressType = atIN6 then
  begin
    Result.In6Addr.sin6_family := AF_INET6;
    Result.In6Addr.sin6_port := HToNS(APort);
    Result.In6Addr.sin6_addr := StrToHostAddr6(AAddress.Address);
    Result.In6Addr.sin6_flowinfo := 0;
    Result.In6Addr.sin6_scope_id := 0;
  end
  else if AAddress.AddressType = atUnixSock then
  begin
    if Length(AAddress.Address) > SizeOf(Result.UnixAddr.sun_path)-1 then
      raise EUnsupportedAddress.Create('Unix address should be at most 108 characters');
    Result.UnixAddr.sun_family := AF_UNIX;
    FillChar(Result.UnixAddr, SizeOf(Result.UnixAddr), #00);
    Move(AAddress.Address[1], Result.UnixAddr.sun_path, Length(AAddress.Address));
  end
  else
    raise EUnsupportedAddress.Create('Address type ' + ord(AAddress.AddressType).ToString + ' not supported');
end;

procedure ReadAddr(constref Addr: _TAddressUnion; DualStack: Boolean; out
  AAddress: TNetworkAddress; out APort: Word);
var
  i:Integer;
begin
  if Addr.In4Addr.sin_family = AF_INET then
  begin
    AAddress := IN4Address(NetAddrToStr(Addr.In4Addr.sin_addr));
    APort := NToHs(Addr.In4Addr.sin_port);
  end
  else if Addr.In6Addr.sin6_family = AF_INET6 then
  begin
    AAddress := IN6Address(HostAddrToStr6(Addr.In6Addr.sin6_addr));
    if DualStack and IsIPv4Mapped(AAddress.Address) then
      AAddress := ExtractIPv4Address(AAddress);
    APort := NToHs(Addr.In6Addr.sin6_port);
  end
  else if Addr.In6Addr.sin6_family = AF_INET6 then
  begin
    AAddress.AddressType := atUnixSock;
    SetLength(AAddress.Address, SizeOf(Addr.UnixAddr.sun_path));
    i:=0;
    while i < Length(Addr.UnixAddr.sun_path) do
      if Addr.UnixAddr.sun_path[i+low(Addr.UnixAddr.sun_path)] = #00 then
        break
      else
        AAddress.Address[i+1] := Addr.UnixAddr.sun_path[i+low(Addr.UnixAddr.sun_path)];
    SetLength(AAddress.Address, i);
    APort := 0;
  end
  else
    raise EUnsupportedAddress.Create('Address Family ' + Addr.In4Addr.sin_family.ToString + ' not supported');
end;

function SocketInvalid(ASocket: TSocket): Boolean; inline;
begin
  {$IfDef Windows}
  Result := ASocket = TSocket(INVALID_SOCKET);
  {$Else}
  Result := ASocket < 0;
  {$EndIf}
end;

function CreateRawSocket(ADomain: TFPSocketType; ASockProto: TFPSocketProto; AProto: Integer): TSocket;
var
  AFam, AType, v6Only: Integer;
begin
  case ADomain of
  stIPv4: AFam := AF_INET;
  stIPv6,
  stIPDualStack: AFam := AF_INET6;
  stUnixSocket: AFam := AF_UNIX;
  end;
  case ASockProto of
  spStream: AType := SOCK_STREAM;
  spDatagram: AType := SOCK_DGRAM;
  end;
  Result := fpsocket(AFam, AType, AProto);
  if SocketInvalid(Result) then
    raise ESocketError.Create(socketerror, 'socket');

  if ADomain = stIPDualStack then
  begin
    v6Only := 0;
    if fpsetsockopt(Result, IPPROTO_IPV6, IPV6_V6ONLY, @v6Only, SizeOf(v6Only)) <> 0 then
    begin
      Sockets.CloseSocket(Result);
      raise EDualStackNotSupported.Create('Dualstack option not supported on this system: ' + socketerror.ToString);
    end;
  end;
end;

function isIPv4Address(const Address:String):Boolean;
var
  dummy:sockets.in_addr;
begin
  Result := TryStrToHostAddr(Address, dummy);
end;

function isIPv6Address(const Address:String):Boolean;
var
  dummy:in6_addr;
begin
  Result := TryStrToHostAddr6(Address, dummy);
end;

function IN4Address(const Address: String): TNetworkAddress;
begin
 Result := Default(TNetworkAddress);
 Result.Address := Address;
 Result.AddressType := atIN4;
end;

function IN6Address(const Address: String): TNetworkAddress;
begin
 Result := Default(TNetworkAddress);
 Result.Address := Address;
 Result.AddressType := atIN6;
end;

function IN4MappedIN6Address(const In4Address: String): TNetworkAddress;
var
  InAddr: TIn_addr;
begin
  InAddr := StrToNetAddr(In4Address);
  Result := IN6Address('::FFFF:%x:%x'.Format([(InAddr.s_bytes[1] shl 8) or InAddr.s_bytes[2],
                                              (InAddr.s_bytes[3] shl 8) or InAddr.s_bytes[4]]));
end;

function UnixAddr(const Address: String):TNetworkAddress;
begin
 Result := Default(TNetworkAddress);
 Result.Address := Address;
 Result.AddressType := atUnixSock;
end;

function NetAddr(const Address: String): TNetworkAddress;
begin
 Result := Default(TNetworkAddress);
  if isIPv4Address(Address) then
    Result.AddressType := atIN4
  else if isIPv6Address(Address) then
    Result.AddressType := atIN6
  else // Filenames can be pretty much anything
    Result.AddressType := atUnixSock;
  Result.Address := Address;
end;

function IsIPv4Mapped(const IPv6Addr: TNetworkAddress): Boolean;
var
  In6Addr: Sockets.TIn6Addr;
begin
  if IPv6Addr.AddressType = atIN4 then
    Exit(True);
  if IPv6Addr.AddressType  <> atIN6 then
    raise EUnsupportedAddress.Create('Can only check IPv4 mapping for IPv6 addresses');
  IN6Addr := StrToHostAddr6(IPv6Addr.Address);
  Result := (IN6Addr.u6_addr16[0] = 0) and
            (IN6Addr.u6_addr16[1] = 0) and
            (IN6Addr.u6_addr16[2] = 0) and
            (IN6Addr.u6_addr16[3] = 0) and
            (IN6Addr.u6_addr16[4] = 0) and
            (IN6Addr.u6_addr16[5] = $FFFF);
end;

function isINAddr(const AAddr:TNetworkAddress):Boolean;
begin
  Result := AAddr.AddressType in [atIN4, atIN6];
end;

function ExtractIPv4Address(const IPv6Addr: TNetworkAddress): TNetworkAddress;
var
  In6Addr: Sockets.TIn6Addr;
begin
  if IPv6Addr.AddressType = atIN4 then
    Exit(IPv6Addr);
  if IPv6Addr.AddressType  <> atIN6 then
    raise EUnsupportedAddress.Create('Can only extract IPv4 mapping from IPv6 addresses');
  IN6Addr := StrToHostAddr6(IPv6Addr.Address);
  Result := IN4Address('%d.%d.%d.%d'.Format([IN6Addr.s6_addr8[12],
                                             IN6Addr.s6_addr8[13],
                                             IN6Addr.s6_addr8[14],
                                             IN6Addr.s6_addr8[15]]));
end;

function IN6Equal(const A, B: String): Boolean;
var
  AAddr, BAddr: Sockets.Tin6_addr;
begin
  AAddr := StrToHostAddr6(A);
  BAddr := StrToHostAddr6(B);
  Result := (AAddr.s6_addr32[0] = BAddr.s6_addr32[0]) and
            (AAddr.s6_addr32[1] = BAddr.s6_addr32[1]) and
            (AAddr.s6_addr32[2] = BAddr.s6_addr32[2]) and
            (AAddr.s6_addr32[3] = BAddr.s6_addr32[3]);
end;

operator=(const A, B: TNetworkAddress): Boolean;
begin
  Result := (A.AddressType = B.AddressType) and (
              ((A.AddressType = atIN4) and (A.Address = B.Address)) or // IPv4: simple string equality
              ((A.AddressType = atIN6) and IN6Equal(A.Address, B.Address)) or // IPv6 check binary equality
              ((A.AddressType = atUnixSock) and SameFileName(A.Address, B.Address)) // UnixSock check if filename equals
            );
end;

operator<>(const A, B: TNetworkAddress): Boolean;
begin
  Result := (A.AddressType <> B.AddressType) or not (
              ((A.AddressType = atIN4) and (A.Address = B.Address)) or // IPv4: simple string equality
              ((A.AddressType = atIN6) and IN6Equal(A.Address, B.Address)) or // IPv6 check binary equality
              ((A.AddressType = atUnixSock) and SameFileName(A.Address, B.Address)) // UnixSock check if filename equals
            );
end;

operator:=(const AStr: String): TNetworkAddress;
begin
  Result := NetAddr(AStr);
end;

function TCPSocket(AType: TFPSocketType): TFPSocket;
begin
  Result.SocketType := AType;
  Result.Protocol := spStream;
  Result.FD := CreateRawSocket(Result.SocketType, Result.Protocol, 0);
end;

function UDPSocket(AType: TFPSocketType): TFPSocket;
begin
  Result.SocketType := AType;
  Result.Protocol := spDatagram;
  Result.FD := CreateRawSocket(Result.SocketType, Result.Protocol, 0);
end;

procedure CloseSocket(const ASocket: TFPSocket);
begin
  Sockets.CloseSocket(ASocket.FD);
end;

procedure Bind(const ASocket: TFPSocket; const AAddress: TNetworkAddress;
  APort: Word; ReuseAddr: Boolean);
var
  enableReuse: Integer = 1;
  addr: _TAddressUnion;
begin
  if ReuseAddr then
    fpsetsockopt(ASocket.FD, SOL_SOCKET, SO_REUSEADDR, @enableReuse, SizeOf(enableReuse));
  addr := CreateAddr(AAddress, APort, ASocket.SocketType = stIPDualStack);
  if fpbind(ASocket.FD, Sockets.PSockAddr(@addr), SizeOf(addr)) <> 0 then raise
    ESocketError.Create(socketerror, 'bind (%s:%d)'.Format([AAddress.Address, APort]));
end;

procedure Listen(const ASocket: TFPSocket; Backlog: Integer);
begin
  if fplisten(ASocket.FD, Backlog) <> 0 then raise
    ESocketError.Create(socketerror, 'listen');
end;

function AcceptConnection(const ASocket: TFPSocket): TFPSocketConnection;
var
  addr: _TAddressUnion;
  addrLen: TSocklen = SizeOf(addr);
begin
  Result.Socket.FD := fpaccept(ASocket.FD, Sockets.psockaddr(@addr), @addrLen);
  if SocketInvalid(Result.Socket.FD) then
    raise ESocketError.Create(socketerror, 'accept');
  Result.Socket.SocketType := ASocket.SocketType;
  Result.Socket.Protocol := ASocket.Protocol;
  ReadAddr(addr, ASocket.SocketType = stIPDualStack, Result.ClientAddress, Result.ClientPort);
end;

function AcceptNonBlocking(const ASocket: TFPSocket): specialize TNullable<
  TFPSocketConnection>;
var
  addr: _TAddressUnion;
  addrLen: TSocklen = SizeOf(addr);
begin
  Result.Ptr^.Socket.FD := fpaccept(ASocket.FD, Sockets.psockaddr(@addr), @addrLen);
  if SocketInvalid(Result.Ptr^.Socket.FD) then
    if WouldBlock(socketerror) then
      Exit(null)
    else
      raise ESocketError.Create(socketerror, 'accept');
  Result.Ptr^.Socket.SocketType := ASocket.SocketType;
  Result.Ptr^.Socket.Protocol := ASocket.Protocol;
  ReadAddr(addr, ASocket.SocketType = stIPDualStack, Result.Ptr^.ClientAddress, Result.Ptr^.ClientPort);
end;

function Connect(const ASocket: TFPSocket; const AAddress: TNetworkAddress;
  APort: Word): TConnectionState;
var
  addr: _TAddressUnion;
const
  EALREADY = {$IfDef Windows}WSAEALREADY{$Else}ESysEALREADY{$EndIf};
  EINPROGRESS = {$IfDef Windows}WSAEINPROGRESS{$Else}ESysEINPROGRESS{$EndIf};
  ECONNREFUSED = {$IfDef Windows}WSAECONNREFUSED{$Else}ESysECONNREFUSED{$EndIf};
begin
  addr := CreateAddr(AAddress, APort, ASocket.SocketType = stIPDualStack);
  if fpconnect(ASocket.FD, Sockets.psockaddr(@addr), SizeOf(addr)) <> 0 then
    case socketerror of
    EALREADY,
    EINPROGRESS,
    EsockEWOULDBLOCK:
      Exit(csPending);
    ECONNREFUSED:
      Exit(csRefused);
    else
      raise ESocketError.Create(socketerror, 'connect');
    end;
  if ASocket.Protocol<>spStream then
    Result := csNotConnected
  else
    Result := csConnected;
end;

function Receive(const ASocket: TFPSocket; ABuffer: Pointer; MaxSize: SizeInt;
  AFlags: Integer): SizeInt;
begin
  Result := fprecv(ASocket.FD, ABuffer, MaxSize, AFlags);
  if Result = 0 then
    raise EConnectionClosedException.Create('The connection closed')
  else if Result < 0 then
    if WouldBlock(socketerror) then
      Result := 0
    else
      raise ESocketError.Create(socketerror, 'recv');
end;

function ReceiveFrom(const ASocket: TFPSocket; ABuffer: Pointer; MaxSize: SizeInt;
  AFlags: Integer): TReceiveFromResult;
var
  addr: _TAddressUnion;
  addrLen: TSocklen;
begin
  Result := Default(TReceiveFromResult);
  addrLen := SizeOf(_TAddressUnion);
  Result.DataSize := fprecvfrom(ASocket.FD, ABuffer, MaxSize, AFlags, Sockets.PSockAddr(@addr), @addrLen);
  if Result.DataSize < 0 then
    if WouldBlock(socketerror) then
      Exit(Default(TReceiveFromResult)) // Will set the DataSize of return to 0
    else
      raise ESocketError.Create(socketerror, 'recvfrom');
  ReadAddr(addr, ASocket.SocketType = stIPDualStack, Result.FromAddr, Result.FromPort);
end;

function ReceiveFromNonBlocking(const ASocket:TFPSocket;ABuffer:Pointer;MaxSize:
  SizeInt;AFlags:Integer):specialize TNullable<TReceiveFromResult>;
begin
  Result := ReceiveFromNonBlocking(ASocket, ABuffer, MaxSize, AFlags);
  if Result.Value.DataSize = 0 then
    Result := null;
end;

function Send(const ASocket: TFPSocket; ABuffer: Pointer; ASize: SizeInt;
  AFlags: Integer): SizeInt;
begin
  Result := fpsend(ASocket.FD, ABuffer, ASize, AFlags);
  if Result < 0 then
    if WouldBlock(socketerror) then
      Result := 0
    else
      raise ESocketError.Create(socketerror, 'send');
end;

function SendTo(const ASocket: TFPSocket; const ReceiverAddr: TNetworkAddress;
  ReceiverPort: Word; ABuffer: Pointer; ASize: SizeInt; AFlags: Integer
  ): SizeInt;
var
  addr: _TAddressUnion;
begin
  addr := CreateAddr(ReceiverAddr, ReceiverPort, ASocket.SocketType = stIPDualStack);
  Result := fpsendto(ASocket.FD, ABuffer, ASize, AFlags, Sockets.psockaddr(@addr), SizeOf(addr));
  if Result < 0 then
    if WouldBlock(socketerror) then
      Result := 0
    else
      raise ESocketError.Create(socketerror, 'sendto');
end;

function ReceiveStr(const ASocket: TFPSocket; MaxLength: SizeInt;
  AFlags: Integer): String;
const
  ReadSize = 1024;
var
  Len, ReadLen: SizeInt;
begin
  Result := '';
  if (MaxLength < 0) and (ASocket.Protocol = spDatagram) then
    MaxLength := MaxUDPPackageSize;
  // If maxlength read as much
  if MaxLength > 0 then
  begin
    SetLength(Result, MaxLength);
    Len := Receive(ASocket, @Result[1], MaxLength, AFlags);
    SetLength(Result, Len);
    Exit;
  end;
  // If no maxlength do a blocking read (required to figure if stream was closed)
  Len := 0;
  MaxLength := BytesAvailable(ASocket);
  if MaxLength = 0 then
    MaxLength := ReadSize;
  repeat
    SetLength(Result, Len + MaxLength);
    ReadLen := Receive(ASocket, @Result[1+Len], MaxLength, AFlags);
    if ReadLen = 0 then // non blocking
      break;
    Len += ReadLen;
    // Check if more was received while reading
    MaxLength:=BytesAvailable(ASocket);
  until (Len < Length(Result)) or (MaxLength <= 0);
  SetLength(Result, Len);
end;

function ReceiveStrFrom(const ASocket: TFPSocket; MaxLength: SizeInt;
  AFlags: Integer): TReceiveFromStringMessage;
var
  UdpMessage: TReceiveFromResult;
begin
  Result := Default(TReceiveFromStringMessage);
  SetLength(Result.Data, MaxLength);
  UdpMessage := ReceiveFrom(ASocket, @Result.Data[1], MaxLength, AFlags);
  SetLength(Result.Data, UdpMessage.DataSize);
  Result.FromAddr := UdpMessage.FromAddr;
  Result.FromPort := UdpMessage.FromPort;
end;

function ReceiveStrFromNonBlocking(const ASocket: TFPSocket;
  MaxLength: SizeInt; AFlags: Integer): specialize TNullable<
  TReceiveFromStringMessage>;
var
  UdpMessage: TReceiveFromResult;
begin
  SetLength(Result.Ptr^.Data, MaxLength);
  UdpMessage := ReceiveFrom(ASocket, @Result.Ptr^.Data[1], MaxLength, AFlags);
  if UdpMessage.DataSize = 0 then
    Exit(null);
  SetLength(Result.Ptr^.Data, UdpMessage.DataSize);
  Result.Ptr^.FromAddr := UdpMessage.FromAddr;
  Result.Ptr^.FromPort := UdpMessage.FromPort;
end;

function SendStr(const ASocket: TFPSocket; const AData: String; AFlags: Integer
  ): SizeInt;
begin
  if Length(AData) = 0 then Exit(0);
  Result := Send(ASocket, @AData[1], Length(AData), AFlags);
end;

function SendStrTo(const ASocket: TFPSocket;
  const ReceiverAddr: TNetworkAddress; ReceiverPort: Word; const AData: String; AFlags: Integer
  ): SizeInt;
begin
  if Length(AData) = 0 then Exit(0);
  Result := SendTo(ASocket, ReceiverAddr, ReceiverPort, @AData[1], Length(AData), AFlags);
end;

generic function Receive<T>(const ASocket: TFPSocket; AFlags: Integer = 0): T;
var
  Frag: TBytes;
  Len, ReadLen: SizeInt;
begin
  Result := Default(T);
  Len := 0;
  while Len < SizeOf(Result) do
  begin
    ReadLen := Receive(ASocket, @PByte(@Result)[Len], SizeOf(Result) - Len, AFlags);
    if ReadLen = 0 then
      if Len = 0 then
        raise ESocketError.Create(EsockEWOULDBLOCK, 'recv')
      else // Fragment received but non blocking afterwards
      begin
        SetLength(Frag, Len);
        Move(Result, Frag[0], Len);
        raise EFragmentedData.Create(Frag, SizeOf(T), 'Only fragment received in non blocking read');
      end;
    Len += ReadLen;
  end;
end;

generic function ReceiveNonBlocking<T>(const ASocket: TFPSocket; AFlags: Integer = 0): specialize TNullable<T>;
var
  Frag: TBytes;
  Len, ReadLen: SizeInt;
begin
  Len := 0;
  while Len < SizeOf(Result.Ptr^) do
  begin
    ReadLen := Receive(ASocket, @PByte(@Result.Ptr^)[Len], SizeOf(Result.Ptr^) - Len, AFlags);
    if ReadLen = 0 then
      if Len = 0 then
        Exit(null)
      else // Fragment received but non blocking afterwards
      begin
        SetLength(Frag, Len);
        Move(Result.Ptr^, Frag[0], Len);
        raise EFragmentedData.Create(Frag, SizeOf(T), 'Only fragment received in non blocking read');
      end;
    Len += ReadLen;
  end;
end;

generic function ReceiveFrom<T>(const ASocket: TFPSocket; AFlags: Integer = 0): specialize TReceiveFromMessage<T>;
var
  Frag: TBytes;
  UdpMessage: TReceiveFromResult;
begin
  Result := Default(specialize TReceiveFromMessage<T>);
  UdpMessage := ReceiveFrom(ASocket, @Result.Data, SizeOf(Result.Data), AFlags);
  if UdpMessage.DataSize < SizeOf(T) then
    if UdpMessage.DataSize = 0 then
        raise ESocketError.Create(EsockEWOULDBLOCK, 'recvfrom')
    else
    begin
      SetLength(Frag, UdpMessage.DataSize);
      Move(Result.Data, Frag[0], UdpMessage.DataSize);
      raise EFragmentedData.Create(Frag, SizeOf(T), 'Only fragment received ReceiveFrom, likely UDP Fragmentation');
    end;
  Result.FromAddr := UdpMessage.FromAddr;
  Result.FromPort := UdpMessage.FromPort;
end;

generic function ReceiveFromNonBlocking<T>(const ASocket: TFPSocket; AFlags: Integer = 0): specialize TNullable<specialize TReceiveFromMessage<T>>;
var
  Frag: TBytes;
  UdpMessage: TReceiveFromResult;
begin
  UdpMessage := ReceiveFrom(ASocket, @Result.Ptr^.Data, SizeOf(Result.Ptr^.Data), AFlags);
  if UdpMessage.DataSize < SizeOf(T) then
    if UdpMessage.DataSize = 0 then
      Exit(null)
    else
    begin
      SetLength(Frag, UdpMessage.DataSize);
      Move(Result.Ptr^.Data, Frag[0], UdpMessage.DataSize);
      raise EFragmentedData.Create(Frag, SizeOf(T), 'Only fragment received ReceiveFrom, likely UDP Fragmentation');
    end;
  Result.Ptr^.FromAddr := UdpMessage.FromAddr;
  Result.Ptr^.FromPort := UdpMessage.FromPort;
end;

generic function Send<T>(const ASocket: TFPSocket; constref AData: T; AFlags: Integer = 0): SizeInt;
begin
  Result := Send(ASocket, @AData, SizeOf(T), AFlags);
end;

generic function SendTo<T>(const ASocket: TFPSocket; const ReceiverAddr: TNetworkAddress; ReceiverPort: Word; constref AData: T; AFlags: Integer = 0): SizeInt;
begin
  Result := SendTo(ASocket, ReceiverAddr, ReceiverPort, @AData, SizeOf(T), AFlags);
end;

generic function ReceiveArray<T>(const ASocket: TFPSocket; MaxCount: SizeInt;
  AFlags: Integer = 0): specialize TArray<T>;
const
  SizeOfT = SizeOf(T);
  ReadCount = 1024 div SizeOfT;
var
  Frag: TBytes;
  Len, ReadLen: SizeInt;
begin
  Result := nil;
  if (MaxCount < 0) and (ASocket.Protocol = spDatagram) then
  {$Push}
  {$WARN 6018 off}
    if SizeOf(T) < MaxUDPPackageSize then
      MaxCount := MaxUDPPackageSize div SizeOf(T)
    else // Lets try anyway and if it fails it fails
      MaxCount := 1;
  {$Pop}

  // If MaxCount, read MaxCount
  if MaxCount > 0 then
  begin
    SetLength(Result, MaxCount);
    Len := 0;
    repeat
      ReadLen := Receive(ASocket, @PByte(@Result[0])[Len], MaxCount * SizeOf(T) - Len, AFlags);
      if ReadLen = 0 then
        if Len = 0 then
          break
        else
        begin
          SetLength(Frag, Len);
          Move(Result[0], Frag[0], Len);
          raise EFragmentedData.Create(Frag, (Len div SizeOf(T) + 1) * SizeOf(T),
            'Receiving of fragmented data is not supported by typed receive');
        end;
      Len += ReadLen;
    until (Len mod SizeOf(T)) = 0;
    SetLength(Result, Len div SizeOf(T));
    Exit;
  end;

  // Else do a (blocking) read and then read as much as in buffer, plus block to finish open blocks
  Len := 0;
  MaxCount := BytesAvailable(ASocket) div SizeOfT;
  {$Push}
  {$WARN 6018 off}
  if MaxCount = 0 then
    if ReadCount = 0 then
      MaxCount := 1
    else
      MaxCount := ReadCount;
  {$Pop}
  repeat
    SetLength(Result, Length(Result)+MaxCount);
    ReadLen := Receive(ASocket, @PByte(@Result[0])[Len], MaxCount*SizeOfT, AFlags);
    if ReadLen = 0 then
      if Len = 0 then
        break
      else
      begin
        SetLength(Frag, Len);
        Move(Result[0], Frag[0], Len);
        raise EFragmentedData.Create(Frag, (Len div SizeOf(T) + 1) * SizeOf(T),
          'Receiving of fragmented data is not supported by typed receive');
      end;
    Len += ReadLen;
    MaxCount := BytesAvailable(ASocket) div SizeOfT;
  until ((Len<Length(Result)*SizeOf(T)) Or (MaxCount = 0)) And ((Len mod SizeOf(T)) = 0);
  SetLength(Result, Len div SizeOf(T));
end;

generic function ReceiveArrayFrom<T>(const ASocket: TFPSocket; MaxCount: SizeInt;
  AFlags: Integer = 0): specialize TReceiveFromMessage<specialize TArray<T>>;
var
  Frag: TBytes;
  UdpMessage: TReceiveFromResult;
begin
  if MaxCount < 0 then
    if SizeOf(T) < MaxUDPPackageSize then
      MaxCount := MaxUDPPackageSize div SizeOf(T)
    else // Lets try anyway and if it fails it fails
      MaxCount := 1;
  Result.Data := nil;
  SetLength(Result.Data, MaxCount);
  UdpMessage := ReceiveFrom(ASocket, @Result.Data[0], MaxCount * SizeOf(T), AFlags);
  if UdpMessage.DataSize mod SizeOf(T) > 0 then
  begin
    SetLength(Frag, UdpMessage.DataSize);
    Move(Result.Data[0], Frag[0], UdpMessage.DataSize);
    raise EFragmentedData.Create(Frag, (UdpMessage.DataSize div SizeOf(T) + 1) * SizeOf(T),
      'Receiving of fragmented data is not supported by typed receive');
  end;
  SetLength(Result.Data, UdpMessage.DataSize div SizeOf(T));
  Result.FromAddr := UdpMessage.FromAddr;
  Result.FromPort := UdpMessage.FromPort;
end;

generic function ReceiveArrayFromNonBlocking<T>(const ASocket: TFPSocket;
  MaxCount: SizeInt = -1; AFlags: Integer = 0
  ): specialize TNullable<specialize TReceiveFromMessage<specialize TArray<T>>>;
var
  Frag: TBytes;
  UdpMessage: TReceiveFromResult;
begin
  if MaxCount < 0 then
    if SizeOf(T) < MaxUDPPackageSize then
      MaxCount := MaxUDPPackageSize div SizeOf(T)
    else // Lets try anyway and if it fails it fails
      MaxCount := 1;
  Result.Ptr^.Data := nil;
  SetLength(Result.Ptr^.Data, MaxCount);
  UdpMessage := ReceiveFrom(ASocket, @Result.Ptr^.Data[0], MaxCount * SizeOf(T), AFlags);
  if UdpMessage.DataSize = 0 then
    Exit(null);
  if UdpMessage.DataSize mod SizeOf(T) > 0 then
  begin
    SetLength(Frag, UdpMessage.DataSize);
    Move(Result.Ptr^.Data[0], Frag[0], UdpMessage.DataSize);
    raise EFragmentedData.Create(Frag, (UdpMessage.DataSize div SizeOf(T) + 1) * SizeOf(T),
      'Receiving of fragmented data is not supported by typed receive');
  end;
  SetLength(Result.Ptr^.Data, UdpMessage.DataSize div SizeOf(T));
  Result.Ptr^.FromAddr := UdpMessage.FromAddr;
  Result.Ptr^.FromPort := UdpMessage.FromPort;
end;

generic function SendArray<T>(const ASocket: TFPSocket; const AData: specialize TArray<T>; AFlags: Integer = 0): SizeInt;
begin
  if Length(AData) = 0 then Exit(0);
  Result := Send(ASocket, @AData[0], Length(AData) * SizeOf(T), AFlags);
end;

generic function SendArrayTo<T>(const ASocket: TFPSocket; const ReceiverAddr: TNetworkAddress; ReceiverPort: Word; const AData: specialize TArray<T>; AFlags: Integer = 0): SizeInt;
begin
  if Length(AData) = 0 then Exit(0);
  Result := SendTo(ASocket, ReceiverAddr, ReceiverPort, @AData[0], Length(AData) * SizeOf(T), AFlags);
end;

procedure SetNonBlocking(const ASocket: TFPSocket; AValue: Boolean);
{$IfDef Windows}
var
  nonblock: u_long;
begin
  nonblock := Ord(AValue);
  ioctlsocket(ASocket.FD, LongInt(FIONBIO), @nonblock);
end;
{$Else}
var
  State: cint;
begin
  State := FpFcntl(ASocket.FD, F_GetFl);
  if AValue then
    State := State Or O_NONBLOCK
  else
    State := State And not O_NONBLOCK;
  FpFcntl(ASocket.FD, F_SetFL, state);
end;
{$EndIf}

function DataAvailable(const SocketArray: specialize TArray<TFPSocket>;
  TimeOut: Integer): specialize TArray<TFPSocket>;
var
  FDSet: TFDSet;
  MaxSock: sockets.TSocket;
  timeval: TTimeVal;
  Ret: LongInt;
  i, WriteHead: Integer;
begin
  Result := nil;
  MaxSock := 0;
  {$IfDef UNIX}fpFD_ZERO{$else}FD_ZERO{$endif}(FDSet);
  for i:=0 to Length(SocketArray) - 1 do
  begin
    MaxSock := Max(MaxSock, SocketArray[i].FD);
    {$IfDef UNIX}fpFD_SET{$else}FD_SET{$endif}(SocketArray[i].FD, FDSet);
  end;
  timeval.tv_sec := TimeOut div 1000;
  timeval.tv_usec := (TimeOut mod 1000) * 1000;
  Ret := {$IfDef UNIX}fpselect{$else}select{$endif}(MaxSock + 1, @FDSet, nil, nil, @timeval);
  if Ret < 0 then
    raise ESocketError.Create(socketerror, 'select');

  SetLength(Result, Ret);
  WriteHead := 0;
  for i:=0 to Length(SocketArray) - 1 do
    if {$IfDef UNIX}fpFD_ISSET{$else}FD_ISSET{$endif}(SocketArray[i].FD, FDSet) {$Ifdef Unix}> 0{$Endif} then
    begin
      Result[WriteHead] := SocketArray[i];
      Inc(WriteHead);
    end;
end;

function DataAvailable(const ASocket: TFPSocket; TimeOut: Integer): Boolean;
var
  Arr: array of TFPSocket;
begin
  Arr := [ASocket];
  Result := Length(DataAvailable(Arr, TimeOut)) > 0;
end;

function DataAvailable(const SocketArray: array of TFPSocket; TimeOut: Integer
  ): specialize TArray<TFPSocket>;
var
  Arr: array of TFPSocket;
begin
  if Length(SocketArray) = 0 then Exit(nil);
  SetLength(Arr, Length(SocketArray));
  Move(SocketArray[0], Arr[0], Length(SocketArray) * SizeOf(SocketArray[0]));
  Result := DataAvailable(arr, TimeOut);
end;

function BytesAvailable(const ASocket: TFPSocket): SizeInt;
var
  {$IfDef WINDOWS}
  count: DWord;
  {$Else}
  count: cint;
  {$EndIf}
begin
  Result := -1;
  {$IfDef WINDOWS}
  if ioctlsocket(ASocket.FD, FIONREAD, @count) = 0 then
  {$Else}
  if FpIOCtl(ASocket.FD, FIONREAD, @count) = 0 then
  {$EndIf}
    Result := Count;
end;

function StreamClosed(const ASocket:TFPSocket):Boolean;
begin
  Result := (ASocket.Protocol <> spStream) Or (
              DataAvailable(ASocket, 0) And
              (BytesAvailable(ASocket) = 0)
            );
end;

function ConnectionState(const ASocket:TFPSocket): TConnectionState;
const
  ECONNREFUSED = {$IfDef WINDOWS}WSAECONNREFUSED{$ELSE}ESysECONNREFUSED{$EndIf};
begin
  if (ASocket.Protocol <> spStream) then
    Exit(csNotConnected);
  if (Sockets.fprecv(ASocket.FD, nil, 0, 0) = 0) And
     (Sockets.fpsend(ASocket.FD, nil, 0, 0) = 0) then
    Exit(csConnected);
  case socketerror of
  EsockEWOULDBLOCK: Result := csConnected;
  ESockENOTCONN: Result := csPending;
  ECONNREFUSED: Result := csRefused;
  else
    Result := csError;
  end;
end;

{ ESocketError }

constructor ESocketError.Create(ACode: Integer; const FunName: String);
begin
  inherited CreateFmt('[Socket Error: %d] %s call failed',  [ACode, FunName]);
  FCode := ACode;
end;

{ EFragmentedData }

constructor EFragmentedData.Create(const AFragment: TBytes; AExpected: SizeInt;
  const AMessage: String);
begin
  inherited Create(AMessage);
  FFragment := AFragment;
  FExpectedSize := AExpected;
end;

end.