1 files changed, 54 insertions, 3 deletions
diff --git a/actioncable/lib/action_cable/connection/stream.rb b/actioncable/lib/action_cable/connection/stream.rb
index 5a2aace0ba..d66e1b4e41 100644
--- a/actioncable/lib/action_cable/connection/stream.rb
+++ b/actioncable/lib/action_cable/connection/stream.rb
@@ -14,6 +14,9 @@ module ActionCable
 
         @rack_hijack_io = nil
         @write_lock = Mutex.new
+
+        @write_head = nil
+        @write_buffer = Queue.new
       end
 
       def each(&callback)
@@ -30,14 +33,62 @@ module ActionCable
       end
 
       def write(data)
-        @write_lock.synchronize do
-          return @rack_hijack_io.write(data) if @rack_hijack_io
-          return @stream_send.call(data) if @stream_send
+        if @stream_send
+          return @stream_send.call(data)
         end
+
+        if @write_lock.try_lock
+          begin
+            if @write_head.nil? && @write_buffer.empty?
+              written = @rack_hijack_io.write_nonblock(data, exception: false)
+
+              case written
+              when :wait_writable
+                # proceed below
+              when data.bytesize
+                return data.bytesize
+              else
+                @write_head = data.byteslice(written, data.bytesize)
+                @event_loop.writes_pending @rack_hijack_io
+
+                return data.bytesize
+              end
+            end
+          ensure
+            @write_lock.unlock
+          end
+        end
+
+        @write_buffer << data
+        @event_loop.writes_pending @rack_hijack_io
+
+        data.bytesize
       rescue EOFError, Errno::ECONNRESET
         @socket_object.client_gone
       end
 
+      def flush_write_buffer
+        @write_lock.synchronize do
+          loop do
+            if @write_head.nil?
+              return true if @write_buffer.empty?
+              @write_head = @write_buffer.pop
+            end
+
+            written = @rack_hijack_io.write_nonblock(@write_head, exception: false)
+            case written
+            when :wait_writable
+              return false
+            when @write_head.bytesize
+              @write_head = nil
+            else
+              @write_head = @write_head.byteslice(written, @write_head.bytesize)
+              return false
+            end
+          end
+        end
+      end
+
       def receive(data)
         @socket_object.parse(data)
       end