Class: CapistranoMulticonfigParallel::RakeWorker

Inherits:
Object
  • Object
show all
Includes:
Celluloid, Celluloid::Logger
Defined in:
lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb

Overview

class that handles the rake task and waits for approval from the celluloid worker

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#action ⇒ Object

Returns the value of attribute action.



7
8
9
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 7

def action
  @action
end

#client ⇒ Object

Returns the value of attribute client.



7
8
9
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 7

def client
  @client
end

#env ⇒ Object

Returns the value of attribute env.



7
8
9
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 7

def env
  @env
end

#job_id ⇒ Object

Returns the value of attribute job_id.



7
8
9
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 7

def job_id
  @job_id
end

#publisher_channel ⇒ Object

Returns the value of attribute publisher_channel.



7
8
9
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 7

def publisher_channel
  @publisher_channel
end

#subscription_channel ⇒ Object

Returns the value of attribute subscription_channel.



7
8
9
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 7

def subscription_channel
  @subscription_channel
end

#successfull_subscription ⇒ Object

Returns the value of attribute successfull_subscription.



7
8
9
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 7

def successfull_subscription
  @successfull_subscription
end

#task ⇒ Object

Returns the value of attribute task.



7
8
9
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 7

def task
  @task
end

#task_approved ⇒ Object

Returns the value of attribute task_approved.



7
8
9
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 7

def task_approved
  @task_approved
end

Instance Method Details

#debug_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


24
25
26
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 24

def debug_enabled?
  @client.debug_enabled?
end

#find_job_id ⇒ Object 



32
33
34
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 32

def find_job_id
  @env[CapistranoMulticonfigParallel::ENV_KEY_JOB_ID]
end

#on_close(code, reason) ⇒ Object 



78
79
80
81
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 78

def on_close(code, reason)
  debug("websocket connection closed: #{code.inspect}, #{reason.inspect}") if debug_enabled?
  terminate
end

#on_message(message) ⇒ Object 



53
54
55
56
57
58
59
60
61
62
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 53

def on_message(message)
  debug("Rake worker #{find_job_id} received after parse #{message}") if debug_enabled?
  if @client.succesfull_subscription?(message)
    publish_subscription_successfull
  elsif message.present? && message['client_action'].blank?
    task_approval(message)
  else
    warn "unknown action: #{message.inspect}" if debug_enabled?
  end
end

#publish_new_work(env, task) ⇒ Object 



44
45
46
47
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 44

def publish_new_work(env, task)
  work(env, task, rake_actor_id: @options['rake_actor_id'])
  publish_to_worker(task_data)
end

#publish_subscription_successfull ⇒ Object 



64
65
66
67
68
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 64

def publish_subscription_successfull
  debug("Rake worker #{find_job_id} received  parse #{message}") if debug_enabled?
  publish_to_worker(task_data)
  @successfull_subscription = true
end

#publish_to_worker(data) ⇒ Object 



49
50
51
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 49

def publish_to_worker(data)
  @client.publish(@publisher_channel, data)
end

#task_approval(message) ⇒ Object 



70
71
72
73
74
75
76
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 70

def task_approval(message)
  if @job_id.to_i == message['job_id'].to_i && message['task'] == task_name && message['approved'] == 'yes'
    @task_approved = true
  else
    warn "unknown invocation #{message.inspect}" if debug_enabled?
  end
end

#task_data ⇒ Object 



36
37
38
39
40
41
42
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 36

def task_data
  {
    action: @action,
    task: task_name,
    job_id: find_job_id
  }
end

#task_name ⇒ Object 



28
29
30
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 28

def task_name
  @task.name
end

#work(env, task, options = {}) ⇒ Object 



9
10
11
12
13
14
15
16
17
18
19
20
21
22
 
# File 'lib/capistrano_multiconfig_parallel/celluloid/rake_worker.rb', line 9

def work(env, task, options = {})
  @options = options.stringify_keys
  @env = env
  @job_id = find_job_id
  @subscription_channel = @options['rake_actor_id']
  @publisher_channel = "worker_#{find_job_id}"
  @action = @options['rake_actor_id'].include?('_count') ? 'count' : 'invoke'
  @task = task
  @task_approved = false
  @successfull_subscription = false
  @client = CelluloidPubsub::Client.connect(actor: Actor.current, enable_debug: CapistranoMulticonfigParallel::CelluloidManager.debug_websocket?) do |ws|
    ws.subscribe(@subscription_channel)
  end if !defined?(@client) || @client.nil?
end