Module: Rbgo::Channel

Included in:
NetworkServiceFactory
Defined in:
lib/rbgo/select_chan.rb

Defined Under Namespace

Modules: Chan Classes: BufferChan, NonBufferChan

Class Method Summary collapse

Class Method Details

.on_read(chan:, &blk) ⇒ Object

on_read

Raises:

  • (ArgumentError) 


400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
 
# File 'lib/rbgo/select_chan.rb', line 400

def on_read(chan:, &blk)
  raise ArgumentError.new('chan must be a Chan') unless chan.is_a? Chan
  op = Proc.new do
    res, ok = chan.deq(true)
    if blk.nil?
      [res, ok]
    else
      blk.call(res, ok)
    end
  end
  op.define_singleton_method(:register) do |io_w|
    chan.send :register, io_w
  end
  op.define_singleton_method(:unregister) do|io_w|
    chan.send :unregister, io_w
  end
  op
end

.on_write(chan:, obj:, &blk) ⇒ Object

on_write

Raises:

  • (ArgumentError) 


425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
 
# File 'lib/rbgo/select_chan.rb', line 425

def on_write(chan:, obj:, &blk)
  raise ArgumentError.new('chan must be a Chan') unless chan.is_a? Chan
  op = Proc.new do
    res = chan.enq(obj, true)
    res = blk.call unless blk.nil?
    res
  end
  op.define_singleton_method(:register) do |io_w|
    chan.send :register, io_w
  end
  op.define_singleton_method(:unregister) do|io_w|
    chan.send :unregister, io_w
  end
  op
end

.select_chan(*ops) ⇒ Object

select_chan



347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
 
# File 'lib/rbgo/select_chan.rb', line 347

def select_chan(*ops)
  ops.shuffle!

  io_hash      = {}
  close_io_blk = proc do
    io_hash.each_pair do |io_r, (op, io_w)|
      io_r.close rescue nil
      io_w.close rescue nil
      op.unregister(io_w)
    end
  end

  begin
    while true do

      close_io_blk.call
      io_hash.clear

      ops.each do |op|
        io_r, io_w = IO.pipe
        op.register(io_w)
        io_hash[io_r] = [op, io_w]
      end

      ops.each do |op|
        begin
          return op.call
        rescue ThreadError
        end
      end

      return yield if block_given?

      read_ios = IO.select(io_hash.keys).first rescue []

      read_ios.each do |io_r|
        op = io_hash[io_r].first
        begin
          return op.call
        rescue ThreadError
        end
      end
    end
  ensure
    close_io_blk.call
  end
end