Non blocking EOF

.gitignore

@@ -1,5 +1,6 @@
 *.gem
 *.rbc
+*.ruby-*
 .bundle
 .config
 .yardoc

lib/websocket-client-simple/client.rb

@@ -19,84 +19,130 @@ def connect(url, options={})
  uri = URI.parse url
  @socket = TCPSocket.new(uri.host,
  uri.port || (uri.scheme == 'wss' ? 443 : 80))
+ @socket.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1)
  if ['https', 'wss'].include? uri.scheme
- ctx = OpenSSL::SSL::SSLContext.new
- ctx.ssl_version = options[:ssl_version] || 'SSLv23'
- ctx.verify_mode = options[:verify_mode] || OpenSSL::SSL::VERIFY_NONE #use VERIFY_PEER for verification
- cert_store = OpenSSL::X509::Store.new
- cert_store.set_default_paths
- ctx.cert_store = cert_store
- @socket = ::OpenSSL::SSL::SSLSocket.new(@socket, ctx)
+ ssl_context = options[:ssl_context] || begin
+ ctx = OpenSSL::SSL::SSLContext.new
+ ctx.ssl_version = options[:ssl_version] || 'SSLv23'
+ ctx.verify_mode = options[:verify_mode] || OpenSSL::SSL::VERIFY_NONE #use VERIFY_PEER for verification
+ cert_store = OpenSSL::X509::Store.new
+ cert_store.set_default_paths
+ ctx.cert_store = cert_store
+ ctx
+ end
+
+ @socket = ::OpenSSL::SSL::SSLSocket.new(@socket, ssl_context)
  @socket.connect
  end
+
  @handshake = ::WebSocket::Handshake::Client.new :url => url, :headers => options[:headers]
  @handshaked = false
  @pipe_broken = false
- frame = ::WebSocket::Frame::Incoming::Client.new
  @closed = false
- once :__close do |err|
- close
- emit :close, err
- end
-
- @thread = Thread.new do
- while !@closed do
- begin
- unless recv_data = @socket.getc
- sleep 1
- next
- end
- unless @handshaked
- @handshake << recv_data
- if @handshake.finished?
- @handshaked = true
- emit :open
- end
- else
- frame << recv_data
- while msg = frame.next
- emit :message, msg
- end
- end
- rescue => e
- emit :error, e
- end
- end
- end
 
- @socket.write @handshake.to_s
+ handshake
+ @thread = poll
  end
 
- def send(data, opt={:type => :text})
- return if !@handshaked or @closed
- type = opt[:type]
- frame = ::WebSocket::Frame::Outgoing::Client.new(:data => data, :type => type, :version => @handshake.version)
+ def send_data(data, opt={:type => :text})
+ return if !@handshaked || @closed
+
+ frame = ::WebSocket::Frame::Outgoing::Client.new(:data => data, :type => opt[:type], :version => @handshake.version)
+
  begin
- @socket.write frame.to_s
+ @socket.write_nonblock(frame.to_s)
+ rescue IO::WaitReadable
+ IO.select([@socket]) # OpenSSL needs to read internally
+ retry
+ rescue IO::WaitWritable, Errno::EINTR
+ IO.select(nil, [@socket])
+ retry
  rescue Errno::EPIPE => e
  @pipe_broken = true
- emit :__close, e
+ close(e)
+ rescue OpenSSL::SSL::SSLError => e
+ @pipe_broken = true
+ close(e)
  end
  end
 
- def close
+ def close(err=nil)
  return if @closed
- if !@pipe_broken
- send nil, :type => :close
- end
+
+ send_data nil, :type => :close if !@pipe_broken
+ emit :close, err
+ ensure
  @closed = true
  @socket.close if @socket
  @socket = nil
- emit :__close
  Thread.kill @thread if @thread
  end
 
+ def handshake
+ @socket.write @handshake.to_s
+
+ while !@handshaked
+ begin
+ read_sockets, _, _ = IO.select([@socket], nil, nil, 10)
+
+ if read_sockets && read_sockets[0]
+ @handshake << @socket.read_nonblock(1024)
+
+ if @socket.respond_to?(:pending) # SSLSocket
+ @handshake << @socket.read(@socket.pending) while @socket.pending > 0
+ end
+
+ @handshaked = @handshake.finished?
+ end
+ rescue IO::WaitReadable
+ # No op
+ rescue IO::WaitWritable
+ IO.select(nil, [socket])
+ retry
+ end
+ end
+ end
+
+ def poll
+ return Thread.new(@socket) do |socket|
+ frame = ::WebSocket::Frame::Incoming::Client.new
+ emit :open
+
+ while !@closed do
+ read_sockets, _, _ = IO.select([socket], nil, nil, 10)
+
+ if read_sockets && read_sockets[0]
+ begin
+ frame << socket.read_nonblock(1024)
+
+ if socket.respond_to?(:pending)
+ frame << socket.read(socket.pending) while socket.pending > 0
+ end
+
+ if msg = frame.next
+ emit :message, msg
+ frame = ::WebSocket::Frame::Incoming::Client.new
+ end
+ rescue IO::WaitReadable
+ # Nothing
+ rescue IO::WaitWritable
+ IO.select(nil, [socket])
+ retry
+ rescue EOFError => e
+ emit :error, e
+ close(e)
+ rescue => e
+ emit :error, e
+ end
+ end
+ end
+ end
+ end
+
  def open?
  @handshake.finished? and !@closed
  end
-
  end
-
  end
  end
 end

sample/client.rb

@@ -27,5 +27,5 @@
 end
 
 loop do
- ws.send STDIN.gets.strip
+ ws.send_data STDIN.gets.strip
 end

sample/echo_server.rb

@@ -12,7 +12,7 @@
  WebSocket::EventMachine::Server.start(:host => "0.0.0.0", :port => PORT) do |ws|
  ws.onopen do
  sid = @channel.subscribe do |mes|
- ws.send mes
+ ws.send_data mes
  end
  puts "<#{sid}> connect"
 

test/echo_server.rb

@@ -4,7 +4,7 @@ def self.start
  @channel = EM::Channel.new
  ws.onopen do
  sid = @channel.subscribe do |mes|
- ws.send mes # echo to client
+ ws.send_data mes # echo to client
  end
  ws.onmessage do |msg|
  @channel.push msg

test/test_connect_block.rb

@@ -13,7 +13,7 @@ def test_onopen
  EM::add_timer 1 do
  WebSocket::Client::Simple.connect EchoServer.url do |client|
  client.on :open do
- client.send "hello world"
+ client.send_data "hello world"
  end
 
  client.on :message do |msg|

test/test_websocket_client_simple.rb

@@ -27,7 +27,7 @@ def test_echo
 
  client1.on :open do
  msgs.each do |m|
- client1.send m
+ client1.send_data m
  end
  end