5 files changed, 37 insertions, 14 deletions
diff --git a/actioncable/lib/action_cable/channel/base.rb b/actioncable/lib/action_cable/channel/base.rb
index 874ebe2e71..86c8f12f6e 100644
--- a/actioncable/lib/action_cable/channel/base.rb
+++ b/actioncable/lib/action_cable/channel/base.rb
@@ -191,7 +191,7 @@ module ActionCable
         # Transmit a hash of data to the subscriber. The hash will automatically be wrapped in a JSON envelope with
         # the proper channel identifier marked as the recipient.
         def transmit(data, via: nil)
-          logger.info "#{self.class.name} transmitting #{data.inspect}".tap { |m| m << " (via #{via})" if via }
+          logger.info "#{self.class.name} transmitting #{data.inspect.truncate(300)}".tap { |m| m << " (via #{via})" if via }
           connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, message: data)
         end
 
diff --git a/actioncable/lib/action_cable/connection/client_socket.rb b/actioncable/lib/action_cable/connection/client_socket.rb
index ef937d7c16..95e1ac4c16 100644
--- a/actioncable/lib/action_cable/connection/client_socket.rb
+++ b/actioncable/lib/action_cable/connection/client_socket.rb
@@ -132,11 +132,8 @@ module ActionCable
           @ready_state = CLOSING
           @close_params = [reason, code]
 
-          if @stream
-            @stream.shutdown
-          else
-            finalize_close
-          end
+          @stream.shutdown if @stream
+          finalize_close
         end
 
         def finalize_close
diff --git a/actioncable/lib/action_cable/connection/subscriptions.rb b/actioncable/lib/action_cable/connection/subscriptions.rb
index d7f95e6a62..24934e12f2 100644
--- a/actioncable/lib/action_cable/connection/subscriptions.rb
+++ b/actioncable/lib/action_cable/connection/subscriptions.rb
@@ -54,7 +54,7 @@ module ActionCable
       end
 
       def unsubscribe_from_all
-        subscriptions.each { |id, channel| channel.unsubscribe_from_channel }
+        subscriptions.each { |id, channel| remove_subscription(channel) }
       end
 
       protected
diff --git a/actioncable/test/client/echo_channel.rb b/actioncable/test/client/echo_channel.rb
index 63e35f194a..5a7bac25c5 100644
--- a/actioncable/test/client/echo_channel.rb
+++ b/actioncable/test/client/echo_channel.rb
@@ -3,6 +3,10 @@ class EchoChannel < ActionCable::Channel::Base
     stream_from "global"
   end
 
+  def unsubscribed
+    'Goodbye from EchoChannel!'
+  end
+
   def ding(data)
     transmit(dong: data['message'])
   end
diff --git a/actioncable/test/client_test.rb b/actioncable/test/client_test.rb
index d30c381131..1b07689127 100644
--- a/actioncable/test/client_test.rb
+++ b/actioncable/test/client_test.rb
@@ -54,7 +54,7 @@ class ClientTest < ActionCable::TestCase
       @ws = Faye::WebSocket::Client.new("ws://127.0.0.1:#{port}/")
       @messages = Queue.new
       @closed = Concurrent::Event.new
-      @has_messages = Concurrent::Event.new
+      @has_messages = Concurrent::Semaphore.new(0)
       @pings = 0
 
       open = Concurrent::Event.new
@@ -79,7 +79,7 @@ class ClientTest < ActionCable::TestCase
           @pings += 1
         else
           @messages << hash
-          @has_messages.set
+          @has_messages.release
         end
       end
 
@@ -92,8 +92,7 @@ class ClientTest < ActionCable::TestCase
     end
 
     def read_message
-      @has_messages.wait(WAIT_WHEN_EXPECTING_EVENT) if @messages.empty?
-      @has_messages.reset if @messages.size < 2
+      @has_messages.try_acquire(1, WAIT_WHEN_EXPECTING_EVENT)
 
       msg = @messages.pop(true)
       raise msg if msg.is_a?(Exception)
@@ -104,9 +103,11 @@ class ClientTest < ActionCable::TestCase
     def read_messages(expected_size = 0)
       list = []
       loop do
-        @has_messages.wait(list.size < expected_size ? WAIT_WHEN_EXPECTING_EVENT : WAIT_WHEN_NOT_EXPECTING_EVENT)
-        if @has_messages.set?
-          list << read_message
+        if @has_messages.try_acquire(1, list.size < expected_size ? WAIT_WHEN_EXPECTING_EVENT : WAIT_WHEN_NOT_EXPECTING_EVENT)
+          msg = @messages.pop(true)
+          raise msg if msg.is_a?(Exception)
+
+          list << msg
         else
           break
         end
@@ -198,4 +199,25 @@ class ClientTest < ActionCable::TestCase
       c.close # disappear before read
     end
   end
+
+  def test_unsubscribe_client
+    with_puma_server do |port|
+      app = ActionCable.server
+      identifier = JSON.dump(channel: 'EchoChannel')
+
+      c = faye_client(port)
+      c.send_message command: 'subscribe', identifier: identifier
+      assert_equal({"identifier"=>"{\"channel\":\"EchoChannel\"}", "type"=>"confirm_subscription"}, c.read_message)
+      assert_equal(1, app.connections.count)
+      assert(app.remote_connections.where(identifier: identifier))
+
+      channel = app.connections.first.subscriptions.send(:subscriptions).first[1]
+      channel.expects(:unsubscribed)
+      c.close
+      sleep 0.1 # Data takes a moment to process
+
+      # All data is removed: No more connection or subscription information!
+      assert_equal(0, app.connections.count)
+    end
+  end
 end