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 = {}  # key: ["action_name", {<action_info>}]
  super
end

Instance Method Details

#dump ⇒ Object 



88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/flydata/helper/action_ownership_channel.rb', line 88

def dump
  $log.info "\n" + <<EOT
========
action_ownership_channel dump

queue:
#{@queue}

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

#map ⇒ Object 



84
85
86
 
# File 'lib/flydata/helper/action_ownership_channel.rb', line 84

def map
  @action_ownership_map.dup
end

#queue ⇒ Object

For debug



80
81
82
 
# File 'lib/flydata/helper/action_ownership_channel.rb', line 80

def queue
  @queue.dup
end

#queue_key(action_name, action_info) ⇒ Object 



44
45
46
 
# File 'lib/flydata/helper/action_ownership_channel.rb', line 44

def queue_key(action_name, action_info)
  [action_name, (action_info || {})[:id]].to_json
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
41
42
 
# 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

  qkey = queue_key(action_name, action_info)
  self.synchronize do
    if @queue.has_key?(qkey)
      false
    else
      @queue[qkey] =
                {
                  action_ownership: action_ownership,
                  action_info: action_info
                }
      true
    end
  end
end

#return_action_ownership(action_ownership) ⇒ Object

Called from worker only



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

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



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/flydata/helper/action_ownership_channel.rb', line 49

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