class Puma::Server
def closed_socket?(socket)
Experimental RBS support (using type sampling data from the type_fusion
project).
def closed_socket?: (TCPSocket socket) -> false
This signature was generated using 1 sample from 1 application.
def closed_socket?(socket) skt = socket.to_io return false unless skt.kind_of?(TCPSocket) && @precheck_closing begin tcp_info = skt.getsockopt(Socket::IPPROTO_TCP, Socket::TCP_INFO) rescue IOError, SystemCallError Puma::Util.purge_interrupt_queue @precheck_closing = false false else state = tcp_info.unpack(UNPACK_TCP_STATE_FROM_TCP_INFO)[0] # TIME_WAIT: 6, CLOSE: 7, CLOSE_WAIT: 8, LAST_ACK: 9, CLOSING: 11 (state >= 6 && state <= 9) || state == 11 end end