Handle more IO errors (especially, ECONNRESET)

Also, address the possibility of the listen thread dying and needing to
be respawned. As a bonus, we now defer construction of the thread until
we are first given something to monitor.

2 files changed, 31 insertions, 0 deletions

diff --git a/actioncable/test/client/echo_channel.rb b/actioncable/test/client/echo_channel.rb
index 9a54080d4d..63e35f194a 100644
--- a/actioncable/test/client/echo_channel.rb
+++ b/actioncable/test/client/echo_channel.rb
@@ -7,6 +7,11 @@ class EchoChannel < ActionCable::Channel::Base
     transmit(dong: data['message'])
   end
 
+  def delay(data)
+    sleep 1
+    transmit(dong: data['message'])
+  end
+
   def bulk(data)
     ActionCable.server.broadcast "global", wide: data['message']
   end
diff --git a/actioncable/test/client_test.rb b/actioncable/test/client_test.rb
index b185654c71..7617e93426 100644
--- a/actioncable/test/client_test.rb
+++ b/actioncable/test/client_test.rb
@@ -181,6 +181,15 @@ class ClientTest < ActionCable::TestCase
       @ws.close
       @closed.wait(WAIT_WHEN_EXPECTING_EVENT)
     end
+
+    def close!
+      sock = BasicSocket.for_fd(@ws.instance_variable_get(:@stream).detach)
+
+      # Force a TCP reset
+      sock.setsockopt(Socket::SOL_SOCKET, Socket::SO_LINGER, [1, 0].pack('ii'))
+
+      sock.close
+    end
   end
 
   def faye_client(port)
@@ -235,4 +244,21 @@ class ClientTest < ActionCable::TestCase
       clients.map {|c| Concurrent::Future.execute { c.close } }.each(&:wait!)
     end
   end
+
+  def test_disappearing_client
+    with_puma_server do |port|
+      c = faye_client(port)
+      c.send_message command: 'subscribe', identifier: JSON.dump(channel: 'EchoChannel')
+      assert_equal({"identifier"=>"{\"channel\":\"EchoChannel\"}", "type"=>"confirm_subscription"}, c.read_message)
+      c.send_message command: 'message', identifier: JSON.dump(channel: 'EchoChannel'), data: JSON.dump(action: 'delay', message: 'hello')
+      c.close! # disappear before write
+
+      c = faye_client(port)
+      c.send_message command: 'subscribe', identifier: JSON.dump(channel: 'EchoChannel')
+      assert_equal({"identifier"=>"{\"channel\":\"EchoChannel\"}", "type"=>"confirm_subscription"}, c.read_message)
+      c.send_message command: 'message', identifier: JSON.dump(channel: 'EchoChannel'), data: JSON.dump(action: 'ding', message: 'hello')
+      assert_equal({"identifier"=>'{"channel":"EchoChannel"}', "message"=>{"dong"=>"hello"}}, c.read_message)
+      c.close! # disappear before read
+    end
+  end
 end