Class: Async::Channel

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

Overview

Channel is a wrapper around Async::Queue that provides a protocol and handy tools for passing data, exceptions and closing. It is designed to be used only with one publisher and one subscriber

Defined Under Namespace

Classes: ChannelClosedError, ChannelError

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Channel

Returns a new instance of Channel.



11
12
13
14
 
# File 'lib/async/channel.rb', line 11

def initialize
  @queue = Async::Queue.new
  @closed = false
end

Instance Method Details

#<<(payload) ⇒ Object

Methods for a publisher

Raises:



25
26
27
28
29
 
# File 'lib/async/channel.rb', line 25

def <<(payload)
  raise(ChannelClosedError, "Cannot send to a closed channel") if @closed

  @queue << [:payload, payload]
end

#close ⇒ Object 



37
38
39
40
41
42
 
# File 'lib/async/channel.rb', line 37

def close
  return if closed?

  @queue << [:close]
  @closed = true
end

#close! ⇒ Object 



44
45
46
47
48
49
 
# File 'lib/async/channel.rb', line 44

def close!
  return if closed?

  exception(ChannelClosedError.new("Channel was forcefully closed"))
  close
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


16
17
18
 
# File 'lib/async/channel.rb', line 16

def closed?
  @closed
end

#dequeue ⇒ Object

Methods for a subscriber



52
53
54
55
56
 
# File 'lib/async/channel.rb', line 52

def dequeue
  each do |payload| # rubocop:disable Lint/UnreachableLoop this is intended
    return payload
  end
end

#each ⇒ Object

Raises:



58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/async/channel.rb', line 58

def each
  raise(ChannelClosedError, "Cannot receive from a closed channel") if closed?

  @queue.each do |type, payload|
    case type
    when :exception
      payload.set_backtrace(caller + (payload.backtrace || [])) # A hack to preserve full backtrace
      raise payload
    when :payload
      yield payload
    when :close
      break
    end
  end
end

#exception(exception) ⇒ Object

Raises:



31
32
33
34
35
 
# File 'lib/async/channel.rb', line 31

def exception(exception)
  raise(ChannelClosedError, "Cannot send to a closed channel") if closed?

  @queue << [:exception, exception]
end

#open? ⇒ Boolean

Returns:

  • (Boolean) 


20
21
22
 
# File 'lib/async/channel.rb', line 20

def open?
  !@closed
end