class Riffer::Rig::Runtime::JobInbox

# File lib/riffer/rig/runtime/job_inbox.rb, line 13
def initialize(clock: -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) })
  @mutex = Mutex.new
  @arrived = ConditionVariable.new
  @clock = clock
  @pending = []
  @subscribers = []
  @closed = false
end

# File lib/riffer/rig/runtime/job_inbox.rb, line 54
def await(timeout)
  deadline = timeout && (@clock.call + timeout)
  @mutex.synchronize do
    until @pending.any? || @closed
      remaining = deadline && (deadline - @clock.call)
      break if remaining && !remaining.positive?

      @arrived.wait(@mutex, remaining)
    end
    @pending.any?
  end
end

# File lib/riffer/rig/runtime/job_inbox.rb, line 68
def close
  @mutex.synchronize do
    @closed = true
    @arrived.broadcast
  end
end

# File lib/riffer/rig/runtime/job_inbox.rb, line 48
def drain
  @mutex.synchronize { @pending.shift(@pending.size) }
end

# File lib/riffer/rig/runtime/job_inbox.rb, line 30
def push(completion)
  # Subscribers run under the lock so that once close returns, none is
  # running and none will start.
  @mutex.synchronize do
    return if @closed

    @pending << completion
    @arrived.broadcast
    @subscribers.each { |subscriber| subscriber.call(completion) }
  end
end

# File lib/riffer/rig/runtime/job_inbox.rb, line 43
def size
  @mutex.synchronize { @pending.size }
end

# File lib/riffer/rig/runtime/job_inbox.rb, line 24
def subscribe(&block)
  @mutex.synchronize { @subscribers << block }
end