Class: ActionCable::Connection::Stream

Inherits:
Object
  • Object
show all
Defined in:
actioncable/lib/action_cable/connection/stream.rb

Overview

– This class is heavily based on faye-websocket-ruby

Copyright © 2010-2015 James Coglan

Instance Method Summary collapse

Constructor Details

#initialize(event_loop, socket) ⇒ Stream

Returns a new instance of Stream.



10
11
12
13
14
15
16
17
18
19
20
 
# File 'actioncable/lib/action_cable/connection/stream.rb', line 10

def initialize(event_loop, socket)
  @event_loop    = event_loop
  @socket_object = socket
  @stream_send   = socket.env["stream.send"]

  @rack_hijack_io = nil
  @write_lock = Mutex.new

  @write_head = nil
  @write_buffer = Queue.new
end

Instance Method Details

#close ⇒ Object 



26
27
28
29
 
# File 'actioncable/lib/action_cable/connection/stream.rb', line 26

def close
  shutdown
  @socket_object.client_gone
end

#each(&callback) ⇒ Object 



22
23
24
 
# File 'actioncable/lib/action_cable/connection/stream.rb', line 22

def each(&callback)
  @stream_send ||= callback
end

#flush_write_buffer ⇒ Object 



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'actioncable/lib/action_cable/connection/stream.rb', line 70

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

#hijack_rack_socket ⇒ Object 



96
97
98
99
100
101
102
103
104
105
 
# File 'actioncable/lib/action_cable/connection/stream.rb', line 96

def hijack_rack_socket
  return unless @socket_object.env["rack.hijack"]

  # This should return the underlying io according to the SPEC:
  @rack_hijack_io = @socket_object.env["rack.hijack"].call
  # Retain existing behavior if required:
  @rack_hijack_io ||= @socket_object.env["rack.hijack_io"]

  @event_loop.attach(@rack_hijack_io, self)
end

#receive(data) ⇒ Object 



92
93
94
 
# File 'actioncable/lib/action_cable/connection/stream.rb', line 92

def receive(data)
  @socket_object.parse(data)
end

#shutdown ⇒ Object 



31
32
33
 
# File 'actioncable/lib/action_cable/connection/stream.rb', line 31

def shutdown
  clean_rack_hijack
end

#write(data) ⇒ Object 



35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'actioncable/lib/action_cable/connection/stream.rb', line 35

def write(data)
  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