Add timeout options

lib/nocturne.rb

@@ -99,6 +99,26 @@ def closed?
     @conn.closed?
   end
 
+  def read_timeout=(timeout)
+    raise ConnectionClosed if @conn.closed?
+    @options[:read_timeout] = timeout
+  end
+
+  def read_timeout
+    raise ConnectionClosed if @conn.closed?
+    @options[:read_timeout]
+  end
+
+  def write_timeout=(timeout)
+    raise ConnectionClosed if @conn.closed?
+    @options[:write_timeout] = timeout
+  end
+
+  def write_timeout
+    raise ConnectionClosed if @conn.closed?
+    @options[:write_timeout]
+  end
+
   private
 
   def connect

lib/nocturne/error.rb

@@ -21,4 +21,7 @@ class ConnectionClosed < ConnectionError
 
   class QueryError < Error
   end
+
+  class TimeoutError < Error
+  end
 end

lib/nocturne/protocol.rb

@@ -13,10 +13,9 @@ def self.error(packet, klass)
     end
 
     def self.read_error(packet)
-      packet.int8
+      packet.skip(1) # Error
       code = packet.int16
-      packet.strn(1)
-      packet.strn(5)
+      packet.skip(6) # SQL state
       message = packet.eof_str
       [code, message]
     end

lib/nocturne/protocol/handshake.rb

@@ -39,7 +39,12 @@ def engage
       private
 
       def server_handshake
+        original_read_timeout = @options[:read_timeout]
+        @options[:read_timeout] = @options[:connect_timeout] || @options[:write_timeout]
+
         @conn.read_packet do |handshake|
+          raise Protocol.error(handshake, ConnectionError) if handshake.err?
+
           _protocol_version = handshake.int8
           @server_version = handshake.nulstr
           _thread_id = handshake.int32
@@ -54,6 +59,8 @@ def server_handshake
           @auth_plugin_data = auth_plugin_data + handshake.strn([13, auth_plugin_data_len - 8].max)
           @auth_plugin_name = handshake.nulstr
         end
+      ensure
+        @options[:read_timeout] = original_read_timeout
       end
 
       def ssl_request

lib/nocturne/protocol/query.rb

@@ -81,6 +81,8 @@ def read_rows(column_count)
         more_rows = true
         while more_rows
           @conn.read_packet do |row|
+            raise Protocol.error(row, QueryError) if row.err?
+
             if row.eof?
               row.skip(1)
               @conn.update_status(

lib/nocturne/socket.rb

@@ -1,5 +1,8 @@
 # frozen_string_literal:true
 
+# standard:disable Lint/MissingCopEnableDirective
+# standard:disable Style/YodaCondition
+
 require "socket"
 require "openssl"
 
@@ -17,8 +20,8 @@ def recv(buffer)
       loop do
         result = @sock.read_nonblock(MAX_BYTES, buffer, exception: false)
 
-        if :wait_readable == result # standard:disable Style/YodaCondition
-          IO.select(@select_sock)
+        if :wait_readable == result
+          IO.select(@select_sock, nil, nil, @options[:read_timeout]) || raise(TimeoutError)
         else
           return result
         end
@@ -29,8 +32,8 @@ def sendmsg(data)
       loop do
         result = @sock.write_nonblock(data, exception: false)
 
-        if :wait_writable == result # standard:disable Style/YodaCondition
-          IO.select(nil, @select_sock)
+        if :wait_writable == result
+          IO.select(nil, @select_sock, nil, @options[:write_timeout]) || raise(TimeoutError)
         else
           return result
         end
@@ -50,7 +53,8 @@ def ssl_sock
     def connect(options)
       if options[:host]
         sock = ::Socket.new(::Socket::AF_INET, ::Socket::SOCK_STREAM)
-        sock.connect ::Socket.pack_sockaddr_in(options[:port] || 3306, options[:host] || "localhost")
+        addr = ::Socket.pack_sockaddr_in(options[:port] || 3306, options[:host] || "localhost")
+        sock.connect_nonblock(addr, exception: false)
       else
         sock = ::Socket.unix(options[:socket] || "/tmp/mysql.sock")
       end
@@ -64,18 +68,17 @@ def initialize(sock, options)
       @sock = OpenSSL::SSL::SSLSocket.new(sock, ssl_context(options))
       @sock.connect
       @select_sock = [@sock]
+      @options = options
     end
 
-    MAX_BYTES = 32768
-
     def recv(buffer)
       loop do
-        result = @sock.read_nonblock(MAX_BYTES, buffer, exception: false)
+        result = @sock.read_nonblock(Nocturne::Socket::MAX_BYTES, buffer, exception: false)
 
-        if :wait_readable == result # standard:disable Style/YodaCondition
-          IO.select(@select_sock)
-        elsif :wait_writable == result # standard:disable Style/YodaCondition
-          IO.select(nil, @select_sock)
+        if :wait_readable == result
+          IO.select(@select_sock, nil, nil, @options[:read_timeout]) || raise(TimeoutError)
+        elsif :wait_writable == result
+          IO.select(nil, @select_sock, nil, @options[:write_timeout]) || raise(TimeoutError)
         else
           return result
         end
@@ -86,10 +89,10 @@ def sendmsg(data)
       loop do
         result = @sock.write_nonblock(data, exception: false)
 
-        if :wait_readable == result # standard:disable Style/YodaCondition
-          IO.select(@select_sock)
-        elsif :wait_writable == result # standard:disable Style/YodaCondition
-          IO.select(nil, @select_sock)
+        if :wait_readable == result
+          IO.select(@select_sock, nil, nil, @options[:read_timeout]) || raise(TimeoutError)
+        elsif :wait_writable == result
+          IO.select(nil, @select_sock, nil, @options[:write_timeout]) || raise(TimeoutError)
         else
           return result
         end