class Async::IO::Socket
def self.accept(*args, backlog: SOMAXCONN, &block)
def self.accept(*args, backlog: SOMAXCONN, &block) bind(*args) do |server, task| server.listen(backlog) if backlog server.accept_each(task: task, &block) end end
def self.bind(local_address, protocol: 0, reuse_port: false, task: Task.current, **options, &block)
(**reuse_port)-
Allow(Boolean) -- this port to be bound in multiple processes. -
The(Integer) -- socket protocol to use.
Parameters:
-
local_address(Address) -- The local address to bind to.
def self.bind(local_address, protocol: 0, reuse_port: false, task: Task.current, **options, &block) Async.logger.debug(self) {"Binding to #{local_address.inspect}"} wrapper = build(local_address.afamily, local_address.socktype, protocol, **options) do |socket| socket.setsockopt(::Socket::SOL_SOCKET, ::Socket::SO_REUSEADDR, true) socket.setsockopt(::Socket::SOL_SOCKET, ::Socket::SO_REUSEPORT, true) if reuse_port socket.bind(local_address.to_sockaddr) end return wrapper unless block_given? task.async do |task| task.annotate "binding to #{local_address.inspect}" begin yield wrapper, task ensure wrapper.close end end end
def self.build(*args, task: Task.current)
def self.build(*args, task: Task.current) socket = wrapped_klass.new(*args) yield socket return self.new(socket, task.reactor) rescue Exception socket.close if socket raise end
def self.connect(remote_address, local_address = nil, reuse_port: false, task: Task.current, **options)
(**protcol)-
The(Integer) -- socket protocol to use.
Parameters:
-
local_address(Addrinfo) -- The local address to bind to before connecting. -
remote_address(Addrinfo) -- The remote address to connect to.
def self.connect(remote_address, local_address = nil, reuse_port: false, task: Task.current, **options) # Async.logger.debug(self) {"Connecting to #{remote_address.inspect}"} task.annotate "connecting to #{remote_address.inspect}" wrapper = build(remote_address.afamily, remote_address.socktype, remote_address.protocol, **options) do |socket| socket.setsockopt(::Socket::SOL_SOCKET, ::Socket::SO_REUSEADDR, reuse_port) if local_address socket.bind(local_address.to_sockaddr) end self.new(socket, task.reactor) end begin wrapper.connect(remote_address.to_sockaddr) task.annotate "connected to #{remote_address.inspect}" rescue wrapper.close raise end return wrapper unless block_given? begin yield wrapper, task ensure wrapper.close end end
def self.pair(*args)
def self.pair(*args) ::Socket.pair(*args).map(&self.method(:new)) end
def accept(task: Task.current)
def accept(task: Task.current) peer, address = async_send(:accept_nonblock) wrapper = Socket.new(peer, task.reactor) return wrapper, address unless block_given? task.async do |task| task.annotate "incoming connection #{address.inspect}" begin yield wrapper, address rescue Async.logger.error(self) {$!} ensure wrapper.close end end end
def connect(*args)
def connect(*args) begin async_send(:connect_nonblock, *args) rescue Errno::EISCONN # We are now connected. end end