Class: Nl::Async::Stream

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
nl/lib/nl/async/operation.rb

Overview

Single-pass consumer handle for a multipart Netlink operation.

Defined Under Namespace

Classes: Sink

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(mailbox:, on_close:) ⇒ Stream

Returns a new instance of Stream.



119
120
121
122
123
124
125
126
 
# File 'nl/lib/nl/async/operation.rb', line 119

def initialize(mailbox:, on_close:)
  @mailbox = mailbox
  @on_close = on_close
  @mutex = Mutex.new
  @owner = nil
  @state = :open
  @error = nil
end

Class Method Details

.build(mailbox: Mailbox.new, on_close: nil) ⇒ Object 



114
115
116
117
 
# File 'nl/lib/nl/async/operation.rb', line 114

def self.build(mailbox: Mailbox.new, on_close: nil)
  operation = new(mailbox:, on_close:)
  [operation, Sink.new(mailbox)]
end

Instance Method Details

#close ⇒ Object

Stops local delivery. The dispatcher still drains the Netlink exchange.



198
199
200
201
202
203
204
205
206
207
208
209
210
 
# File 'nl/lib/nl/async/operation.rb', line 198

def close #: nil
  closed, callback = @mutex.synchronize do
    next [false, nil] if @state != :open

    @state = :closed
    [true, @on_close]
  end
  if closed
    callback&.call
    @mailbox.close
  end
  nil
end

#closed? ⇒ Boolean

: bool

Returns:

  • (Boolean) 


212
213
214
 
# File 'nl/lib/nl/async/operation.rb', line 212

def closed? #: bool
  @mutex.synchronize { @state != :open }
end

#each ⇒ Object 



130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'nl/lib/nl/async/operation.rb', line 130

def each
  return enum_for(__method__) unless block_given?

  state, error = current_state
  case state
  when :completed then return self
  when :closed then raise ClosedError, 'stream is closed'
  when :failed then raise error
  end

  claim_consumer!
  begin
    loop do
      case event = @mailbox.pop
      when Events::Item
        yield event.value
      when Events::Done
        completed = mark_completed
        @mailbox.close
        terminal_result unless completed
        break
      when Events::Error
        failed = mark_failed(event.exception)
        @mailbox.close
        raise event.exception if failed

        terminal_result
      end
    end
    self
  rescue ClosedError
    terminal_result
  ensure
    close if open?
  end
end

#next(timeout: nil) ⇒ Object 



168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
 
# File 'nl/lib/nl/async/operation.rb', line 168

def next(timeout: nil)
  state, error = current_state
  case state
  when :completed then raise StopIteration
  when :closed then raise ClosedError, 'stream is closed'
  when :failed then raise error
  end

  claim_consumer!
  case event = @mailbox.pop(timeout:)
  when Events::Item
    event.value
  when Events::Done
    completed = mark_completed
    @mailbox.close
    raise StopIteration if completed

    terminal_result(next_item: true)
  when Events::Error
    failed = mark_failed(event.exception)
    @mailbox.close
    raise event.exception if failed

    terminal_result(next_item: true)
  end
rescue ClosedError
  terminal_result(next_item: true)
end