Class: Rbgo::Channel::BufferChan

Inherits:
SizedQueue
  • Object
show all
Includes:
Enumerable, Chan
Defined in:
lib/rbgo/select_chan.rb

Overview

BufferChan

Instance Method Summary collapse

Methods included from Chan

new

Constructor Details

#initialize(max) ⇒ BufferChan

Returns a new instance of BufferChan.



242
243
244
245
246
247
248
 
# File 'lib/rbgo/select_chan.rb', line 242

def initialize(max)
  super(max)
  @readable_observers = Set.new
  @readable_observers.extend(MonitorMixin)
  @writable_observers = Set.new
  @writable_observers.extend(MonitorMixin)
end

Instance Method Details

#clear ⇒ Object 



271
272
273
274
275
 
# File 'lib/rbgo/select_chan.rb', line 271

def clear
  super
  notify_writable_observers
  self
end

#close ⇒ Object 



277
278
279
280
281
282
 
# File 'lib/rbgo/select_chan.rb', line 277

def close
  super
  notify_readable_observers
  notify_writable_observers
  self
end

#each ⇒ Object 



228
229
230
231
232
233
234
235
236
237
238
239
240
 
# File 'lib/rbgo/select_chan.rb', line 228

def each
  if block_given?
    loop do
      begin
        yield pop(true)
      rescue ThreadError
        return
      end
    end
  else
    enum_for(:each)
  end
end

#pop(nonblock = false) ⇒ Object Also known as: deq, shift 



259
260
261
262
263
264
265
266
267
268
269
 
# File 'lib/rbgo/select_chan.rb', line 259

def pop(nonblock = false)
  res = nil
  begin
    res = super(nonblock)
    notify_writable_observers
    res
  rescue ThreadError
    raise unless closed?
  end
  [res, !closed?]
end

#push(obj, nonblock = false) ⇒ Object Also known as: <<, enq 



250
251
252
253
254
255
256
257
 
# File 'lib/rbgo/select_chan.rb', line 250

def push(obj, nonblock = false)
  super(obj, nonblock)
  notify_readable_observers
  self
rescue ThreadError
  raise ClosedQueueError.new if closed?
  raise
end