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

#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.



18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/libuv/tcp.rb', line 18

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.



13
14
15
 
# File 'lib/libuv/tcp.rb', line 13

def connected
  @connected
end

Instance Method Details

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



191
192
193
194
195
196
197
198
199
200
201
202
203
204
 
# File 'lib/libuv/tcp.rb', line 191

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 ------------------



166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/libuv/tcp.rb', line 166

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



108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/libuv/tcp.rb', line 108

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



83
84
85
86
87
88
89
90
91
 
# File 'lib/libuv/tcp.rb', line 83

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 



206
207
208
209
210
211
212
213
214
215
216
217
218
 
# File 'lib/libuv/tcp.rb', line 206

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))
    rescue Exception => e
        reject(e)
    end
end

#direct_write ⇒ Object 



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

alias_method :direct_write, :write

#disable_keepalive ⇒ Object 



249
250
251
252
 
# File 'lib/libuv/tcp.rb', line 249

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

#disable_nodelay ⇒ Object 



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

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

#disable_simultaneous_accepts ⇒ Object 



259
260
261
262
 
# File 'lib/libuv/tcp.rb', line 259

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



63
64
65
66
67
68
69
 
# File 'lib/libuv/tcp.rb', line 63

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

#do_shutdown ⇒ Object 



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

alias_method :do_shutdown, :shutdown

#enable_keepalive(delay) ⇒ Object 



244
245
246
247
 
# File 'lib/libuv/tcp.rb', line 244

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 



234
235
236
237
 
# File 'lib/libuv/tcp.rb', line 234

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

#enable_simultaneous_accepts ⇒ Object 



254
255
256
257
 
# File 'lib/libuv/tcp.rb', line 254

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

#handshake_cb ⇒ Object

Push through any pending writes when handshake has completed



51
52
53
54
55
56
57
58
59
 
# File 'lib/libuv/tcp.rb', line 51

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

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



180
181
182
183
184
185
186
187
188
189
 
# File 'lib/libuv/tcp.rb', line 180

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 



227
228
229
230
231
232
 
# File 'lib/libuv/tcp.rb', line 227

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 



154
155
156
157
158
159
160
 
# File 'lib/libuv/tcp.rb', line 154

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

#sockname ⇒ Object 



220
221
222
223
224
225
 
# File 'lib/libuv/tcp.rb', line 220

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 ----------------------



39
40
41
42
43
44
45
46
47
48
 
# File 'lib/libuv/tcp.rb', line 39

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) 


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

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)



73
74
75
76
77
78
79
80
 
# File 'lib/libuv/tcp.rb', line 73

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 



93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/libuv/tcp.rb', line 93

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



131
132
133
 
# File 'lib/libuv/tcp.rb', line 131

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

#write(data) ⇒ Object 



136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/libuv/tcp.rb', line 136

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