Class: Vigilem::Core::Demultiplexer

Inherits:
Object
  • Object
show all
Extended by:
Support::ObjSpace
Includes:
Observable
Defined in:
lib/vigilem/core/demultiplexer.rb

Overview

#shift’s from input and distributes it to the observers

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(input_source = nil, observers = []) ⇒ Demultiplexer

Returns a new instance of Demultiplexer.

Parameters:

  • observers (Array<Array<observer_object, Hash{@see #add_observer}>>) (defaults to: [])
  • input_source ((Buffer) #shift) (defaults to: nil) 


37
38
39
40
41
 
# File 'lib/vigilem/core/demultiplexer.rb', line 37

def initialize(input_source=nil, observers=[])
  @input = input_source
  
  add_observers(observers)
end

Instance Attribute Details

#input ⇒ Object Also known as: source

Returns the value of attribute input.



22
23
24
 
# File 'lib/vigilem/core/demultiplexer.rb', line 22

def input
  @input
end

Class Method Details

.acquire(input_source = nil, observers = [], &block) ⇒ Demultiplexer

like first_or_new

Returns:



192
193
194
195
196
197
198
199
 
# File 'lib/vigilem/core/demultiplexer.rb', line 192

def acquire(input_source=nil, observers=[], &block)
  found = all.find(&block)
  if found
    found.add_observers(observers)
  else
    new(input_source, observers)
  end
end

.event_option_names ⇒ Array<Symbol>

Returns:

  • (Array<Symbol>) 


211
212
213
 
# File 'lib/vigilem/core/demultiplexer.rb', line 211

def event_option_names
  @event_option_names ||=  option_names - [:func]
end

.filter_events(events, event_opts) ⇒ Array

Parameters:

  • events (Array<#metadata, #type>)
  • event_opts (Hash{@see #add_observer})

Returns:

  • (Array) 


163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
 
# File 'lib/vigilem/core/demultiplexer.rb', line 163

def filter_events(events, event_opts)
  events.select do |event|
    opts = event_opts.select {|k,v| event_option_names.include? k }
    if opts.empty?
      event
    else
      opts.all? do |opt,v|
        opt_vals = [v].flatten
        case opt.to_s
        when /^devices?$/
          opt_vals.any? do |dev| 
            if (src = event.metadata[:source])
              File.identical?(dev, src)
            end
          end
        when /^device_names?$/
          opt_vals.any? {|dev_name| event.metadata[:source].respond.name =~ dev_name }
        when /^types?$/
          opt_vals.any? {|ev_type| event.type == ev_type }
        end
      end
      
    end # else
    
  end
end

.inherited(base) ⇒ Object

Parameters:

  • base

Returns:



155
156
157
 
# File 'lib/vigilem/core/demultiplexer.rb', line 155

def inherited(base)
  base.extend Support::ObjSpace
end

.new(input_source = nil, observers = []) ⇒ Object

Parameters:

  • observers (Array<Array<observer_object, Hash{@see #add_observer}>>) (defaults to: [])
  • input_source ((Buffer) #slice!) (defaults to: nil) 


30
31
32
 
# File 'lib/vigilem/core/demultiplexer.rb', line 30

def self.new(input_source=nil, observers=[])
  obj_register(super(input_source, observers))
end

.option_names ⇒ Array<Symbol>

Returns:

  • (Array<Symbol>) 


217
218
219
 
# File 'lib/vigilem/core/demultiplexer.rb', line 217

def option_names
  @options ||= [:func, :type, :types, :device_name, :device_names, :device, :devices]
end

.same_source_check ⇒ Hash

Returns:

  • (Hash) 


203
204
205
206
207
 
# File 'lib/vigilem/core/demultiplexer.rb', line 203

def same_source_check
  all.group_by do |dem| 
    dem.respond(:fileno) || dem
  end.select {|k, v| v.size > 1 }
end

Instance Method Details

#_observers ⇒ Array Also known as: _outputs

Returns:

  • (Array) 


53
54
55
 
# File 'lib/vigilem/core/demultiplexer.rb', line 53

def _observers
  (@observer_peers ||= ThreadSafe::Hash.new).keys
end

#add_observer(observer, event_opts = {}) ⇒ Array Also known as: add_output

Returns [event_opts].

Parameters:

  • event_opts (Hash) (defaults to: {})
  • :func (Hash) —

    a customizable set of options

  • :type (Hash) —

    a customizable set of options

  • :types (Hash) —

    a customizable set of options

  • :device_name (Hash) —

    a customizable set of options

  • :device_names (Hash) —

    a customizable set of options

  • :device (Hash) —

    a customizable set of options

  • :devices (Hash) —

    a customizable set of options

Returns:

  • (Array) —
    event_opts

See Also:

  • Observable#add_observer


69
70
71
72
73
74
75
76
 
# File 'lib/vigilem/core/demultiplexer.rb', line 69

def add_observer(observer, event_opts={})
  observer = Support::LazySimpleDelegator.new(observer).use_strict_eql if observer.is_a? Array
  _observers
  unless observer.respond_to?(event_opts[:func] ||= :update)
    raise NoMethodError, "observer does not respond to `#{func.to_s}'"
  end
  @observer_peers[observer] = event_opts
end

#add_observers(observers_and_opts) ⇒ Array Also known as: add_outputs

Parameters:

Returns:

  • (Array)

See Also:

  • Observable#add_observer


83
84
85
86
87
 
# File 'lib/vigilem/core/demultiplexer.rb', line 83

def add_observers(observers_and_opts)
  observers_and_opts.map do |(obs, hsh)|
    add_observer(*[obs, hsh].compact)
  end
end

#demux(max_number_of_events = 1) ⇒ Object

gets events from input and notifies_observers

Parameters:

  • max_number_of_events (Integer) (defaults to: 1)

Returns:



115
116
117
118
119
120
121
 
# File 'lib/vigilem/core/demultiplexer.rb', line 115

def demux(max_number_of_events=1)
  semaphore.synchronize {
    events = input.shift(max_number_of_events)
    changed
    notify_observers(*events)
  }
end

#filter_events(events, event_opts) ⇒ Array

Parameters:

Returns:

  • (Array) 


146
147
148
 
# File 'lib/vigilem/core/demultiplexer.rb', line 146

def filter_events(events, event_opts)
  self.class.filter_events(events, event_opts)
end

#inspect ⇒ String

Returns:

  • (String)

See Also:

  • Object#inspect


94
95
96
97
98
99
100
 
# File 'lib/vigilem/core/demultiplexer.rb', line 94

def inspect
  if input.is_a? Array  # @todo switch from regex
    super.gsub(/@input=.+?(\s+|>)/, "@input=#<#{input.class}:#{Support::Utils.inspect_id(input)} #{input}>\\1")
  else
    super
  end
end

#notify_observers(*events) ⇒ TrueClass || FalseClass Also known as: sweep, notify_outputs

Parameters:

  • (Array)

Returns:

  • (TrueClass || FalseClass)

See Also:

  • Observable#notify_observers


127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/vigilem/core/demultiplexer.rb', line 127

def notify_observers(*events)
  if defined? @observer_state and @observer_state
    if defined? @observer_peers
      @observer_peers.each do |k, event_opts|
        filtered = self.class.filter_events(events, event_opts)
        k.send event_opts[:func], filtered unless filtered.empty?
      end
    end
    @observer_state = false
  end
end

#observers ⇒ Array Also known as: outputs

the peers without the Delegators

Returns:

  • (Array) 


45
46
47
 
# File 'lib/vigilem/core/demultiplexer.rb', line 45

def observers
  _observers.map {|obj| obj.respond(:peel) || obj }
end

#semaphore ⇒ Monitor

Returns:

  • (Monitor) 


104
105
106
107
108
109
110
 
# File 'lib/vigilem/core/demultiplexer.rb', line 104

def semaphore
  if defined? super
    super
  else
    @semaphore ||= Monitor.new
  end
end