Class: Nl::Async::Mailbox

Inherits:
Object
  • Object
show all
Defined in:
nl/lib/nl/async/mailbox.rb

Overview

A thread-safe, scheduler-aware, single-consumer mailbox.

Defined Under Namespace

Classes: FullError

Instance Method Summary collapse

Constructor Details

#initialize(capacity: nil) ⇒ Mailbox

Returns a new instance of Mailbox.

Raises:

  • (ArgumentError) 


9
10
11
12
13
14
15
16
17
 
# File 'nl/lib/nl/async/mailbox.rb', line 9

def initialize(capacity: nil)
  raise ArgumentError, 'capacity must be positive' if capacity && capacity <= 0

  @mutex = Mutex.new
  @condition = ConditionVariable.new
  @queue = []
  @closed = false
  @capacity = capacity
end

Instance Method Details

#close ⇒ Object 



62
63
64
65
66
67
68
69
70
 
# File 'nl/lib/nl/async/mailbox.rb', line 62

def close
  @mutex.synchronize do
    return nil if @closed

    @closed = true
    @condition.broadcast
  end
  nil
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


72
73
74
 
# File 'nl/lib/nl/async/mailbox.rb', line 72

def closed?
  @mutex.synchronize { @closed }
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


76
77
78
 
# File 'nl/lib/nl/async/mailbox.rb', line 76

def empty?
  @mutex.synchronize { @queue.empty? }
end

#pop(timeout: nil) ⇒ Object

Removes the next value, suspending through the current thread or Fiber scheduler if empty.



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'nl/lib/nl/async/mailbox.rb', line 46

def pop(timeout: nil)
  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout if timeout

  @mutex.synchronize do
    loop do
      return @queue.shift unless @queue.empty?
      raise ClosedError if @closed

      remaining = deadline && deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
      raise TimeoutError if remaining && remaining <= 0

      @condition.wait(@mutex, remaining)
    end
  end
end

#push(value) ⇒ Object

Adds a value without executing consumer code.



21
22
23
 
# File 'nl/lib/nl/async/mailbox.rb', line 21

def push(value)
  push_value(value, enforce_capacity: true)
end

#push_terminal(value) ⇒ Object

Adds a completion/error event even when the value capacity is full.



26
27
28
 
# File 'nl/lib/nl/async/mailbox.rb', line 26

def push_terminal(value)
  push_value(value, enforce_capacity: false)
end