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 



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

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

Instance Method Details

#close ⇒ Object

Closes the selector.



121
122
123
124
125
126
127
128
129
 
# File 'lib/httpx/selector.rb', line 121

def close
  return if @closed

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

#deregister(io) ⇒ Object

deregisters io from selectables.



57
58
59
60
 
# File 'lib/httpx/selector.rb', line 57

def deregister(io)
  monitor = @selectables.delete(io)
  monitor.close(false) if monitor
end

#register(io, interests) ⇒ Object

register io for interests events.



63
64
65
66
67
68
69
70
71
72
 
# File 'lib/httpx/selector.rb', line 63

def register(io, interests)
  monitor = @selectables[io]
  if monitor
    monitor.interests = interests
  else
    monitor = Monitor.new(io, interests, self)
    @selectables[io] = monitor
  end
  monitor
end

#select(interval) ⇒ Object

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



77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/httpx/selector.rb', line 77

def select(interval)
  begin
    r = [@__r__]
    w = []

    @selectables.each do |io, monitor|
      r << io if monitor.interests == :r || monitor.interests == :rw
      w << io if monitor.interests == :w || monitor.interests == :rw
      monitor.readiness = nil
    end

    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
    @selectables.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? ? @selectables.delete(io) : @selectables[io]
      next unless monitor

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

  writers.each do |io|
    monitor = io.closed? ? @selectables.delete(io) : @selectables[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.



132
133
134
 
# File 'lib/httpx/selector.rb', line 132

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