Class: Flydata::Helper::ActionOwnershipChannel

Inherits:
Object
  • Object
show all
Includes:
FlydataCore::Logger, MonitorMixin
Defined in:
lib/flydata/helper/action_ownership_channel.rb

Overview

Take care of action ownership management

  • Receive a request from both scheduler and worker.

  • Give a action with ownership to workers

Instance Method Summary collapse

Constructor Details

#initialize ⇒ ActionOwnershipChannel

Returns a new instance of ActionOwnershipChannel.



13
14
15
16
17
 
# File 'lib/flydata/helper/action_ownership_channel.rb', line 13

def initialize
  @action_ownership_map = ActionOwnership.action_ownership_map
  @queue = {}
  super
end

Instance Method Details

#dump ⇒ Object 



82
83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/flydata/helper/action_ownership_channel.rb', line 82

def dump
  $logger.debug "\n" + <<EOT
========
action_ownership_channel dump

queue:
#{@queue}

action_ownership_map:
#{@action_ownership_map}
========
EOT
end

#map ⇒ Object 



78
79
80
 
# File 'lib/flydata/helper/action_ownership_channel.rb', line 78

def map
  @action_ownership_map.dup
end

#queue ⇒ Object

For debug



74
75
76
 
# File 'lib/flydata/helper/action_ownership_channel.rb', line 74

def queue
  @queue.dup
end

#request_action(action_name, action_info = {}) ⇒ Object

Called from schdeduler and worker action_name must be symbol



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/flydata/helper/action_ownership_channel.rb', line 21

def request_action(action_name, action_info = {})
  action_ownership = @action_ownership_map[action_name]
  if action_ownership.nil?
    # unknown action.  probably a new action
    log_warn "unsupported action `#{action_name}'.  Skip."
    return false
  end
  self.synchronize do
    if @queue.has_key?(action_name)
      false
    else
      @queue[action_name] =
                {
                  action_ownership: action_ownership,
                  action_info: action_info
                }
      true
    end
  end
end

#return_action_ownership(action_ownership) ⇒ Object

Called from worker only



66
67
68
69
70
71
 
# File 'lib/flydata/helper/action_ownership_channel.rb', line 66

def return_action_ownership(action_ownership)
  self.synchronize do
    action_ownership.owner = nil
    action_ownership.last_processed_time = Time.now.to_f
  end
end

#take_action_ownership(new_owner) ⇒ Object

Called from worker only



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/flydata/helper/action_ownership_channel.rb', line 43

def take_action_ownership(new_owner)
  self.synchronize do
    return nil if @queue.empty?

    # Wait until action will be processed
    action = @queue.values.first
    action_ownership = action[:action_ownership]
    return nil if action_ownership.processing?

    # Check resource change flag
    if action_ownership.resource_change &&
      @action_ownership_map.any? { |name, act_own|
        (act_own.resource_change && act_own.processing?) }
      return nil
    end

    @queue.shift  # delete last acion from queue
    action_ownership.owner = new_owner
    action
  end
end