Class: GraphAgent::Channels::Topic

Inherits:
BaseChannel show all
Defined in:
lib/graph_agent/channels/topic.rb

Constant Summary

Constants inherited from BaseChannel

BaseChannel::MISSING

Instance Attribute Summary collapse

Attributes inherited from BaseChannel

#key

Instance Method Summary collapse

Methods inherited from BaseChannel

#consume, #finish

Constructor Details

#initialize(key: "", accumulate: false) ⇒ Topic

Returns a new instance of Topic.



8
9
10
11
12
 
# File 'lib/graph_agent/channels/topic.rb', line 8

def initialize(key: "", accumulate: false)
  super(key: key)
  @accumulate = accumulate
  @values = []
end

Instance Attribute Details

#accumulate ⇒ Object (readonly)

Returns the value of attribute accumulate.



6
7
8
 
# File 'lib/graph_agent/channels/topic.rb', line 6

def accumulate
  @accumulate
end

Instance Method Details

#available? ⇒ Boolean

Returns:

  • (Boolean) 


37
38
39
 
# File 'lib/graph_agent/channels/topic.rb', line 37

def available?
  !@values.empty?
end

#checkpoint ⇒ Object 



41
42
43
 
# File 'lib/graph_agent/channels/topic.rb', line 41

def checkpoint
  @values.dup
end

#copy ⇒ Object 



51
52
53
54
55
 
# File 'lib/graph_agent/channels/topic.rb', line 51

def copy
  ch = self.class.new(key: key, accumulate: @accumulate)
  ch.instance_variable_set(:@values, @values.dup)
  ch
end

#from_checkpoint(checkpoint) ⇒ Object 



45
46
47
48
49
 
# File 'lib/graph_agent/channels/topic.rb', line 45

def from_checkpoint(checkpoint)
  ch = self.class.new(key: key, accumulate: @accumulate)
  ch.instance_variable_set(:@values, checkpoint.equal?(MISSING) ? [] : Array(checkpoint))
  ch
end

#get ⇒ Object

Raises:



14
15
16
17
18
 
# File 'lib/graph_agent/channels/topic.rb', line 14

def get
  raise EmptyChannelError.new("Channel '#{key}' is empty") if @values.empty?

  @values.dup
end

#update(values) ⇒ Object 



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/graph_agent/channels/topic.rb', line 20

def update(values)
  updated = false

  unless @accumulate
    updated = !@values.empty?
    @values = []
  end

  flat = values.flat_map { |v| v.is_a?(Array) ? v : [v] }
  unless flat.empty?
    updated = true
    @values.concat(flat)
  end

  updated
end