Class: Async::IO::Socket
Constant Summary
Constants inherited
from Generic
Generic::WRAPPERS
Class Method Summary
collapse
-
.accept(*args, backlog: SOMAXCONN, &block) ⇒ Object
Bind to a local address and accept connections in a loop.
-
.bind(local_address, protocol: 0, reuse_port: nil, reuse_address: true, task: Task.current, **options, &block) ⇒ Object
-
.build(*args, task: Task.current) ⇒ Object
-
.connect(remote_address, local_address = nil, reuse_port: nil, task: Task.current, **options) ⇒ Object
Establish a connection to a given ‘remote_address`.
-
.pair(*args) ⇒ Object
Instance Method Summary
collapse
Methods included from Server
#accept_each
Methods included from Peer
#connected?, #protocol, #sync, #sync=, #type
Methods inherited from Generic
#connected?, #nonblock, #nonblock?, #read, #wait, wrap, wrap_blocking_method, wraps, #write
Class Method Details
.accept(*args, backlog: SOMAXCONN, &block) ⇒ Object
Bind to a local address and accept connections in a loop.
228
229
230
231
232
233
234
|
# File 'lib/async/io/socket.rb', line 228
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
|
.bind(local_address, protocol: 0, reuse_port: nil, reuse_address: true, task: Task.current, **options, &block) ⇒ Object
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
|
# File 'lib/async/io/socket.rb', line 198
def self.bind(local_address, protocol: 0, reuse_port: nil, reuse_address: true, 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|
if reuse_address
socket.setsockopt(::Socket::SOL_SOCKET, ::Socket::SO_REUSEADDR, 1)
end
if reuse_port
socket.setsockopt(::Socket::SOL_SOCKET, ::Socket::SO_REUSEPORT, 1)
end
socket.bind(local_address.to_sockaddr)
end
return wrapper unless block_given?
task.async do |task|
task.annotate "binding to #{wrapper.local_address.inspect}"
begin
yield wrapper, task
ensure
wrapper.close
end
end
end
|
.build(*args, task: Task.current) ⇒ Object
140
141
142
143
144
145
146
147
148
149
150
|
# File 'lib/async/io/socket.rb', line 140
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
|
.connect(remote_address, local_address = nil, reuse_port: nil, task: Task.current, **options) ⇒ Object
Establish a connection to a given ‘remote_address`.
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
|
# File 'lib/async/io/socket.rb', line 158
def self.connect(remote_address, local_address = nil, reuse_port: nil, 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|
if reuse_port
socket.setsockopt(::Socket::SOL_SOCKET, ::Socket::SO_REUSEADDR, 1)
end
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 Exception
wrapper.close
raise
end
return wrapper unless block_given?
begin
yield wrapper, task
ensure
wrapper.close
end
end
|
.pair(*args) ⇒ Object
238
239
240
|
# File 'lib/async/io/socket.rb', line 238
def self.pair(*args)
::Socket.pair(*args).map(&self.method(:new))
end
|
Instance Method Details
#accept(task: Task.current) ⇒ Object
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
# File 'lib/async/io/socket.rb', line 118
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
|
#accept_nonblock ⇒ Object
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
# File 'lib/async/io/socket.rb', line 137
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
|
#connect(*args) ⇒ Object
108
109
110
111
112
113
114
|
# File 'lib/async/io/socket.rb', line 108
def connect(*args)
begin
async_send(:connect_nonblock, *args)
rescue Errno::EISCONN
end
end
|
#connect_nonblock ⇒ Object
116
117
118
119
120
121
122
|
# File 'lib/async/io/socket.rb', line 116
def connect(*args)
begin
async_send(:connect_nonblock, *args)
rescue Errno::EISCONN
end
end
|
#sysaccept ⇒ Object
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
# File 'lib/async/io/socket.rb', line 138
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
|