Class: Nl::NotificationChannel

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

Overview

Thread-safe queue shared by a family's blocking and asynchronous facades.

Instance Method Summary collapse

Constructor Details

#initialize(capacity:) ⇒ NotificationChannel

Returns a new instance of NotificationChannel.

Raises:

  • (ArgumentError) 


10
11
12
13
14
15
16
17
18
19
 
# File 'nl/lib/nl/notification.rb', line 10

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

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

Instance Method Details

#close ⇒ Object 



73
74
75
76
77
78
79
80
81
82
83
 
# File 'nl/lib/nl/notification.rb', line 73

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

    @closed = true
    @queue.clear
    @error = nil
    @condition.broadcast
  end
  nil
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


69
70
71
 
# File 'nl/lib/nl/notification.rb', line 69

def empty?
  @mutex.synchronize { @queue.empty? && !@error }
end

#fail(error) ⇒ Object

Records a broken notification boundary while allowing later delivery to resume after the consumer observes the error and resynchronizes state.



40
41
42
43
44
45
46
47
 
# File 'nl/lib/nl/notification.rb', line 40

def fail(error)
  @mutex.synchronize do
    return false if @closed

    lose!(error)
  end
  true
end

#pop(timeout: nil) ⇒ Object 



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'nl/lib/nl/notification.rb', line 49

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

  @mutex.synchronize do
    loop do
      if error = @error
        @error = nil
        raise error
      end
      return @queue.shift unless @queue.empty?
      raise ClosedError, 'notification channel is closed' if @closed

      remaining = deadline && deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
      raise TimeoutError, 'notification receive timed out' if remaining && remaining <= 0

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

#push(notification) ⇒ Object

Adds a notification without ever blocking the socket receive loop. Returns false after the channel has been closed.



23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'nl/lib/nl/notification.rb', line 23

def push(notification)
  @mutex.synchronize do
    return false if @closed

    if @capacity && @queue.length >= @capacity
      lose!(NotificationLossError.new('notification queue capacity exceeded'))
    else
      wake = @queue.empty? && !@error
      @queue << notification
      @condition.signal if wake
    end
  end
  true
end