Class: Wires::Hub

Inherits:
Object
  • Object
show all
Extended by:
Hegemon
Defined in:
lib/wires/hub.rb

Overview

An Event Hub. Event/proc associations come in, and the procs get called in new threads in the order received

Class Method Summary collapse

Class Method Details

.<<(x) ⇒ Object 



90
 
# File 'lib/wires/hub.rb', line 90

def <<(x); fire(x); end

.after_kill(proc = nil, retain: false, &block) ⇒ Object

Register hook to execute after kill - can call multiple times



75
76
77
78
79
 
# File 'lib/wires/hub.rb', line 75

def after_kill(proc=nil, retain:false, &block)
  func = (block or proc)
  expect_type func, Proc
  @after_kills << [func, retain]
nil end

.alive? ⇒ Boolean

Returns:

  • (Boolean) 


36
 
# File 'lib/wires/hub.rb', line 36

def alive?; state==:alive end

.before_kill(proc = nil, retain: false, &block) ⇒ Object

Register hook to execute before kill - can call multiple times



68
69
70
71
72
 
# File 'lib/wires/hub.rb', line 68

def before_kill(proc=nil, retain:false, &block)
  func = (block or proc)
  expect_type func, Proc
  @before_kills << [func, retain]
nil end

.clone_deeper ⇒ Object

Clone all clonable instance variables as well



25
26
27
28
29
30
31
32
33
 
# File 'lib/wires/hub.rb', line 25

def clone_deeper
  result = clone
  self.instance_variables.each do |sym|
    begin varcopy = instance_variable_get(sym).clone
    rescue TypeError; end
    result.instance_variable_set(sym, varcopy)
  end
  result
end

.dead? ⇒ Boolean

Returns:

  • (Boolean) 


35
 
# File 'lib/wires/hub.rb', line 35

def dead?;  state==:dead  end

.fire(x) ⇒ Object

Put x in the queue, and block until x is processed (if Hub is running)



82
83
84
85
86
87
88
89
 
# File 'lib/wires/hub.rb', line 82

def fire(x)
  if not dead? # yield to event loop thread until awoken by it later
    @queue << [x, Thread.current]
    sleep
  else        # don't wait if Hub isn't running - would cause lockup
    @queue << [x, nil]
  end
nil end

.flush_queue ⇒ Object 



92
93
94
 
# File 'lib/wires/hub.rb', line 92

def flush_queue
  (process_item(@queue.shift) until @queue.empty?)
end

.kill(*flags) ⇒ Object

Kill the Hub event loop (optional flags change thread behavior)

valid flags:

:finish_all

Hub thread won’t be done until all child threads done

:blocking

calling thread won’t be done until Hub thread is done



61
62
63
64
65
 
# File 'lib/wires/hub.rb', line 61

def kill(*flags)
  @please_finish_all = (flags.include? :finish_all)
  @please_kill = true
  block_until_state :dead if (flags.include? :blocking)
nil end

.run(*flags) ⇒ Object

Start the Hub event loop (optional flags change thread behavior)

valid flags:

:blocking

Hub event loop will be run in calling thread, blocking until Hub is killed. If this flag is not specified, the Hub event loop is run in a new thread, which the main thread joins in at_exit.



46
47
48
49
50
51
52
53
 
# File 'lib/wires/hub.rb', line 46

def run(*flags)
  request_state :alive until alive?
  
  # If :blocking, block now, else block at exit
  (flags.include? :blocking)   ?
    (join_hegemon_auto_thread) :
    (at_exit { join_hegemon_auto_thread unless $! })
end

.state_machine_init ⇒ Object 



177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
 
# File 'lib/wires/hub.rb', line 177

def state_machine_init
  
  impose_state :dead
  
  declare_state :dead do
    
    transition_to :alive do
      after { start_hegemon_auto_thread }
    end
  end
  
  declare_state :alive do
    
    task do
      # puts "task #{Thread.current.inspect}"; 
      if @queue.empty? then sleep(0)
      else process_item(@queue.shift) end
    end
    
    transition_to :dead do
      condition {@please_kill}
      
      before { run_hooks @before_kills }
      
      before { join_children if @please_finish_all }
      
      after  { run_hooks @after_kills  }
      
      after  { @please_kill = false }
      after  { @please_finish_all = false }
      
      after  { end_hegemon_auto_thread }
      after  { do_state_tasks }
    end
  end
  
end