Class: Libuv::TCP

Inherits:
Handle show all
Includes:
Net, Stream
Defined in:
lib/libuv/tcp.rb

Defined Under Namespace

Classes: Socket4, Socket6, SocketBase

Constant Summary collapse

TLS_ERROR = 
"TLS write failed".freeze

Constants included from Stream

Stream::BACKLOG_ERROR, Stream::CLOSED_HANDLE_ERROR, Stream::STREAM_CLOSED_ERROR, Stream::WRITE_ERROR

Constants included from Net

Net::INET6_ADDRSTRLEN, Net::INET_ADDRSTRLEN, Net::IP_ARGUMENT_ERROR, Net::PORT_ARGUMENT_ERROR

Constants included from Assertions

Assertions::MSG_NO_PROC

Constants inherited from Q::Promise

Q::Promise::MAKE_PROMISE

Instance Attribute Summary collapse

Attributes inherited from Handle

#closed, #loop, #storage

Instance Method Summary collapse

Methods included from Stream

included, #listen, #progress, #readable?, #start_read, #stop_read, #try_write, #writable?

Methods inherited from Handle

#active?, #closing?, #ref, #unref

Methods included from Assertions

#assert_block, #assert_boolean, #assert_type

Methods included from Resource

#check_result, #check_result!, #resolve, #to_ptr

Methods inherited from Q::DeferredPromise

#resolved?, #then

Methods inherited from Q::Promise

#catch, #finally, #progress

Constructor Details

#initialize(loop, acceptor = nil) ⇒ TCP

Returns a new instance of TCP.



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/libuv/tcp.rb', line 23

def initialize(loop, acceptor = nil)
    @loop = loop

    tcp_ptr = ::Libuv::Ext.allocate_handle_tcp
    error = check_result(::Libuv::Ext.tcp_init(loop.handle, tcp_ptr))

    if acceptor && error.nil?
        error = check_result(::Libuv::Ext.accept(acceptor, tcp_ptr))
        @connected = true
    else
        @connected = false
    end
    
    super(tcp_ptr, error)
end

Instance Attribute Details

#connected ⇒ Object (readonly)

Returns the value of attribute connected.



16
17
18
 
# File 'lib/libuv/tcp.rb', line 16

def connected
  @connected
end

#protocol ⇒ Object (readonly)

Returns the value of attribute protocol.



17
18
19
 
# File 'lib/libuv/tcp.rb', line 17

def protocol
  @protocol
end

Instance Method Details

#accept(callback = nil, &blk) ⇒ Object 



209
210
211
212
213
214
215
216
217
218
219
220
221
222
 
# File 'lib/libuv/tcp.rb', line 209

def accept(callback = nil, &blk)
    begin
        raise RuntimeError, CLOSED_HANDLE_ERROR if @closed
        tcp = TCP.new(loop, handle)
        begin
            (callback || blk).call(tcp)
        rescue Exception => e
            @loop.log :error, :tcp_accept_cb, e
        end
    rescue Exception => e
        @loop.log :info, :tcp_accept_failed, e
    end
    nil
end

#bind(ip, port, callback = nil, &blk) ⇒ Object

END TLS Abstraction ------------------



184
185
186
187
188
189
190
191
192
193
194
195
196
 
# File 'lib/libuv/tcp.rb', line 184

def bind(ip, port, callback = nil, &blk)
    return if @closed
    @on_listen = callback || blk
    assert_type(String, ip, IP_ARGUMENT_ERROR)
    assert_type(Integer, port, PORT_ARGUMENT_ERROR)

    begin
        @tcp_socket = create_socket(IPAddr.new(ip), port)
        @tcp_socket.bind
    rescue Exception => e
        reject(e)
    end
end

#close ⇒ Object

overwrite the default close to ensure pending writes are rejected



126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/libuv/tcp.rb', line 126

def close
    return if @closed

    # Free tls memory
    # Next tick as may recieve data after closing
    if @tls
        @loop.next_tick do
            @tls.cleanup
        end
    end
    @connected = false

    if @pending_writes
        @pending_writes.each do |deferred, data|
            deferred.reject(TLS_ERROR)
        end
        @pending_writes = nil
    end

    super
end

#close_cb ⇒ Object

Close can be called multiple times



101
102
103
104
105
106
107
108
109
 
# File 'lib/libuv/tcp.rb', line 101

def close_cb
    if @pending_write
        @pending_write.reject(TLS_ERROR)
        @pending_write = nil
    end

    # Shutdown the stream
    close
end

#connect(ip, port, callback = nil, &blk) ⇒ Object 



224
225
226
227
228
229
230
231
232
233
234
235
236
 
# File 'lib/libuv/tcp.rb', line 224

def connect(ip, port, callback = nil, &blk)
    return if @closed
    @callback = callback || blk
    assert_type(String, ip, IP_ARGUMENT_ERROR)
    assert_type(Integer, port, PORT_ARGUMENT_ERROR)
    
    begin
        @tcp_socket = create_socket(IPAddr.new(ip), port)
        @tcp_socket.connect(callback(:on_connect, @tcp_socket.connect_req.address))
    rescue Exception => e
        reject(e)
    end
end

#direct_write ⇒ Object 



153
 
# File 'lib/libuv/tcp.rb', line 153

alias_method :direct_write, :write

#disable_keepalive ⇒ Object 



267
268
269
270
 
# File 'lib/libuv/tcp.rb', line 267

def disable_keepalive
    return if @closed
    check_result ::Libuv::Ext.tcp_keepalive(handle, 0, 0)
end

#disable_nodelay ⇒ Object 



257
258
259
260
 
# File 'lib/libuv/tcp.rb', line 257

def disable_nodelay
    return if @closed
    check_result ::Libuv::Ext.tcp_nodelay(handle, 0)
end

#disable_simultaneous_accepts ⇒ Object 



277
278
279
280
 
# File 'lib/libuv/tcp.rb', line 277

def disable_simultaneous_accepts
    return if @closed
    check_result ::Libuv::Ext.tcp_simultaneous_accepts(handle, 0)
end

#dispatch_cb(data) ⇒ Object

This is clear text data that has been decrypted Same as stream.rb on_read for clear text



81
82
83
84
85
86
87
 
# File 'lib/libuv/tcp.rb', line 81

def dispatch_cb(data)
    begin
        @progress.call data, self
    rescue Exception => e
        @loop.log :error, :stream_progress_cb, e
    end
end

#do_shutdown ⇒ Object 



171
 
# File 'lib/libuv/tcp.rb', line 171

alias_method :do_shutdown, :shutdown

#enable_keepalive(delay) ⇒ Object 



262
263
264
265
 
# File 'lib/libuv/tcp.rb', line 262

def enable_keepalive(delay)
    return if @closed                   # The to_i asserts integer
    check_result ::Libuv::Ext.tcp_keepalive(handle, 1, delay.to_i)
end

#enable_nodelay ⇒ Object 



252
253
254
255
 
# File 'lib/libuv/tcp.rb', line 252

def enable_nodelay
    return if @closed
    check_result ::Libuv::Ext.tcp_nodelay(handle, 1)
end

#enable_simultaneous_accepts ⇒ Object 



272
273
274
275
 
# File 'lib/libuv/tcp.rb', line 272

def enable_simultaneous_accepts
    return if @closed
    check_result ::Libuv::Ext.tcp_simultaneous_accepts(handle, 1)
end

#handshake_cb(protocol = nil) ⇒ Object

Push through any pending writes when handshake has completed



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/libuv/tcp.rb', line 56

def handshake_cb(protocol = nil)
    @handshake = true
    @protocol = protocol

    writes = @pending_writes
    @pending_writes = nil
    writes.each do |deferred, data|
        @pending_write = deferred
        @tls.encrypt(data)
    end

    begin
        @on_handshake.call(self, protocol) if @on_handshake
    rescue => e
        @loop.log :warn, :tls_handshake_callback_error, e
    end
end

#on_handshake(callback = nil, &blk) ⇒ Object

Provide a callback once the TLS handshake has completed



75
76
77
 
# File 'lib/libuv/tcp.rb', line 75

def on_handshake(callback = nil, &blk)
    @on_handshake = callback || blk
end

#open(fd, binding = true, callback = nil, &blk) ⇒ Object 



198
199
200
201
202
203
204
205
206
207
 
# File 'lib/libuv/tcp.rb', line 198

def open(fd, binding = true, callback = nil, &blk)
    return if @closed
    if binding
        @on_listen = callback || blk
    else
        @callback = callback || blk
    end
    error = check_result UV.tcp_open(handle, fd)
    reject(error) if error
end

#peername ⇒ Object 



245
246
247
248
249
250
 
# File 'lib/libuv/tcp.rb', line 245

def peername
    return [] if @closed
    sockaddr, len = get_sockaddr_and_len
    check_result! ::Libuv::Ext.tcp_getpeername(handle, sockaddr, len)
    get_ip_and_port(::Libuv::Ext::Sockaddr.new(sockaddr), len.get_int(0))
end

#shutdown ⇒ Object 



172
173
174
175
176
177
178
 
# File 'lib/libuv/tcp.rb', line 172

def shutdown
    if @pending_writes && @pending_writes.length > 0
        @pending_writes[-1][0].finally method(:do_shutdown)
    else
        do_shutdown
    end
end

#sockname ⇒ Object 



238
239
240
241
242
243
 
# File 'lib/libuv/tcp.rb', line 238

def sockname
    return [] if @closed
    sockaddr, len = get_sockaddr_and_len
    check_result! ::Libuv::Ext.tcp_getsockname(handle, sockaddr, len)
    get_ip_and_port(::Libuv::Ext::Sockaddr.new(sockaddr), len.get_int(0))
end

#start_tls(args = {}) ⇒ Object

TLS Abstraction ----------------------



44
45
46
47
48
49
50
51
52
53
 
# File 'lib/libuv/tcp.rb', line 44

def start_tls(args = {})
    return unless @connected && @tls.nil?

    args[:verify_peer] = true if @on_verify

    @handshake = false
    @pending_writes = []
    @tls = ::RubyTls::SSL::Box.new(args[:server], self, args)
    @tls.start
end

#tls? ⇒ Boolean

Check if tls active on the socket

Returns:

  • (Boolean) 


20
 
# File 'lib/libuv/tcp.rb', line 20

def tls?; !@tls.nil?; end

#transmit_cb(data) ⇒ Object

We resolve the existing tls write promise with a the real writes promise (a close may have occurred)



91
92
93
94
95
96
97
98
 
# File 'lib/libuv/tcp.rb', line 91

def transmit_cb(data)
    if @pending_write
        @pending_write.resolve(direct_write(data))
        @pending_write = nil
    else
        direct_write(data)
    end
end

#verify_cb(cert) ⇒ Object 



111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/libuv/tcp.rb', line 111

def verify_cb(cert)
    if @on_verify
        begin
            return @on_verify.call cert
        rescue => e
            @loop.log :warn, :tls_verify_callback_failed, e
            return false
        end
    end

    true
end

#verify_peer(callback = nil, &blk) ⇒ Object

Verify peers will be called for each cert in the chain



149
150
151
 
# File 'lib/libuv/tcp.rb', line 149

def verify_peer(callback = nil, &blk)
    @on_verify = callback || blk
end

#write(data) ⇒ Object 



154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/libuv/tcp.rb', line 154

def write(data)
    if @tls.nil?
        direct_write(data)
    else
        deferred = @loop.defer
        
        if @handshake == true
            @pending_write = deferred
            @tls.encrypt(data)
        else
            @pending_writes << [deferred, data]
        end

        deferred.promise
    end
end