about summary refs log tree commit homepage
path: root/lib/rainbows/stream_response_epoll/client.rb
blob: b7574693c038a009cd977d968b31fc45b4e007c2 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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
# -*- encoding: binary -*-
# :enddoc:
class Rainbows::StreamResponseEpoll::Client
  OUT = SleepyPenguin::Epoll::OUT
  N = Raindrops.new(1)
  EP = SleepyPenguin::Epoll.new
  timeout = Rainbows.server.timeout
  thr = Thread.new do
    begin
      EP.wait(nil, timeout) { |_,client| client.epoll_run }
    rescue Errno::EINTR
    rescue => e
      Rainbows::Error.listen_loop(e)
    end while Rainbows.alive || N[0] > 0
  end
  Rainbows.at_quit { thr.join(timeout) }

  attr_reader :to_io

  def initialize(io, unwritten)
    @finish = false
    @to_io = io
    @wr_queue = [ unwritten.dup ]
    EP.set(self, OUT)
  end

  def write(str)
    @wr_queue << str.dup
  end

  def close
    @finish = true
  end

  def hijack(hijack)
    @finish = hijack
  end

  def epoll_run
    return if @to_io.closed?
    buf = @wr_queue.shift or return on_write_complete
    case rv = @to_io.kgio_trywrite(buf)
    when nil
      buf = @wr_queue.shift or return on_write_complete
    when String # retry, socket buffer may grow
      buf = rv
    when :wait_writable
      return @wr_queue.unshift(buf)
    end while true
  rescue
    @to_io.close
    N.decr(0, 1)
  end

  def on_write_complete
    if true == @finish
      @to_io.shutdown
      @to_io.close
      N.decr(0, 1)
    elsif @finish.respond_to?(:call) # hijacked
      EP.delete(self)
      N.decr(0, 1)
      @finish.call(@to_io)
    end
  end
end