Class: HTTPX::Selector

Inherits:
Object
  • Object
show all
Defined in:
lib/httpx/selector.rb

Defined Under Namespace

Classes: Monitor

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Selector

Returns a new instance of Selector.



50
51
52
53
54
55
 
# File 'lib/httpx/selector.rb', line 50

def initialize
  @readers = {}
  @writers = {}
  @__r__, @__w__ = IO.pipe
  @closed = false
end

Instance Method Details

#close ⇒ Object

Closes the selector.



136
137
138
139
140
141
142
143
144
 
# File 'lib/httpx/selector.rb', line 136

def close
  return if @closed

  @__r__.close
  @__w__.close
rescue IOError
ensure
  @closed = true
end

#deregister(io) ⇒ Object

deregisters io from selectables.



58
59
60
61
62
63
 
# File 'lib/httpx/selector.rb', line 58

def deregister(io)
  rmonitor = @readers.delete(io)
  wmonitor = @writers.delete(io)
  monitor = rmonitor || wmonitor
  monitor.close(false) if monitor
end

#register(io, interests) ⇒ Object

register io for interests events.



66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/httpx/selector.rb', line 66

def register(io, interests)
  readable = READABLE.include?(interests)
  writable = WRITABLE.include?(interests)
  if readable
    monitor = @readers[io]
    if monitor
      monitor.interests = interests
    else
      monitor = Monitor.new(io, interests, self)
    end
    @readers[io] = monitor
    @writers.delete(io) unless writable
  end
  if writable
    monitor = @writers[io]
    if monitor
      monitor.interests = interests
    else
      # reuse object
      monitor = readable ? @readers[io] : Monitor.new(io, interests, self)
    end
    @writers[io] = monitor
    @readers.delete(io) unless readable
  end
  monitor
end

#select(interval) ⇒ Object

waits for read/write events for interval. Yields for monitors of selected IO objects.



96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
 
# File 'lib/httpx/selector.rb', line 96

def select(interval)
  begin
    r = @readers.keys
    w = @writers.keys
    r.unshift(@__r__)

    readers, writers = IO.select(r, w, nil, interval)

    raise HTTPX::TimeoutError.new(interval, "timed out while waiting on select") if readers.nil? && writers.nil?
  rescue IOError, SystemCallError
    @readers.reject! { |io, _| io.closed? }
    @writers.reject! { |io, _| io.closed? }
    retry
  end

  readers.each do |io|
    if io == @__r__
      # clean up wakeups
      @__r__.read(@__r__.stat.size)
    else
      monitor = io.closed? ? @readers.delete(io) : @readers[io]
      next unless monitor

      monitor.readiness = writers.delete(io) ? :rw : :r
      yield monitor
    end
  end if readers

  writers.each do |io|
    monitor = io.closed? ? @writers.delete(io) : @writers[io]
    next unless monitor

    # don't double run this, the last iteration might have run this task already
    monitor.readiness = :w
    yield monitor
  end if writers
end

#wakeup ⇒ Object

interrupts the select call.



147
148
149
 
# File 'lib/httpx/selector.rb', line 147

def wakeup
  @__w__.write_nonblock("\0", exception: false)
end