1  from __future__ import annotations
   2  
   3  import select
   4  import socket
   5  import sys
   6  
   7  
   8  def is_socket_readable(sock: socket.socket | None) -> bool:
   9      """
  10      Return whether a socket, as identifed by its file descriptor, is readable.
  11      "A socket is readable" means that the read buffer isn't empty, i.e. that calling
  12      .recv() on it would immediately return some data.
  13      """
  14      # NOTE: we want check for readability without actually attempting to read, because
  15      # we don't want to block forever if it's not readable.
  16  
  17      # In the case that the socket no longer exists, or cannot return a file
  18      # descriptor, we treat it as being readable, as if it the next read operation
  19      # on it is ready to return the terminating `b""`.
  20      sock_fd = None if sock is None else sock.fileno()
  21      if sock_fd is None or sock_fd < 0:  # pragma: nocover
  22          return True
  23  
  24      # The implementation below was stolen from:
  25      # https://github.com/python-trio/trio/blob/20ee2b1b7376db637435d80e266212a35837ddcc/trio/_socket.py#L471-L478
  26      # See also: https://github.com/encode/httpcore/pull/193#issuecomment-703129316
  27  
  28      # Use select.select on Windows, and when poll is unavailable and select.poll
  29      # everywhere else. (E.g. When eventlet is in use. See #327)
  30      if (
  31          sys.platform == "win32" or getattr(select, "poll", None) is None
  32      ):  # pragma: nocover
  33          rready, _, _ = select.select([sock_fd], [], [], 0)
  34          return bool(rready)
  35      p = select.poll()
  36      p.register(sock_fd, select.POLLIN)
  37      return bool(p.poll(0))
  38