Class: Thor::Node

Inherits:
Application show all
Defined in:
lib/ThorNode.rb

Defined Under Namespace

Classes: StructJob

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods inherited from Application

#amqp_handle_failure, #amqp_start, #amqp_stop, #run

Constructor Details

#initialize(opts = {}) ⇒ Node

C-tor



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/ThorNode.rb', line 38

def initialize(opts = {})
	super(opts)
	
	@jobs = {}
	@jobs_lock = Mutex.new

	options[:jobs_dir] = File.join(File.dirname(__FILE__), 'jobs')
	options[:boot_file] = ""

	initialize_optparser { |opts|
	############################
	# AMQP Section
	############################
		# Jobs Dir
		opts.on( '-j', '--jobs-dir STRING', "AMQP Virtual Host") do |jobs_dir|
			options[:jobs_dir] = jobs_dir
		end

		# Boot file
		opts.on( '-b', '--boot-file STRING', "Boot file specifying which jobs run automaticaly") do |boot_file|
			options[:boot_file] = boot_file
		end
	}
end

Instance Attribute Details

#jobs ⇒ Object

Returns the value of attribute jobs.



33
34
35
 
# File 'lib/ThorNode.rb', line 33

def jobs
  @jobs
end

#jobs_lock ⇒ Object

Returns the value of attribute jobs_lock.



33
34
35
 
# File 'lib/ThorNode.rb', line 33

def jobs_lock
  @jobs_lock
end

#request_exit ⇒ Object

Returns the value of attribute request_exit.



33
34
35
 
# File 'lib/ThorNode.rb', line 33

def request_exit
  @request_exit
end

Instance Method Details

#amqp_loop(amqp) ⇒ Object 



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/ThorNode.rb', line 63

def amqp_loop(amqp)
	super(amqp)
	
	# Start boot time jobs, move somewhere else!
	boot_file = options[:boot_file]
	if(boot_file != nil && boot_file != "")
		boot_jobs = load_boot_file(boot_file)
		if(boot_jobs != nil)
			start_boot_time_jobs(boot_jobs, amqp)
		end
	end

	#amqp_exchange_create_direct(guid) # Create local direct exchange
	#amqp_exchange_connect_master(options[:amqp_channel_master], guid)
end

#enumerate_jobs(dir, exclude = []) ⇒ Object

Enumerate existing jobs



80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/ThorNode.rb', line 80

def enumerate_jobs(dir, exclude = [])
	res_jobs = {}

	if(File.directory?(dir) == false)
		return nil
	end	

	Dir[dir + "/*"].each do |path|
		next if (File.directory?(path) == false)
		basename = File.basename(path)
		dash_index = basename.index("-")
		if(dash_index == nil || dash_index < 0 || (basename.length - dash_index - 1) < 5)
			Bsl::Logger::Log "Invalid job dir name - #{path}"
			next
		end

		job_name = basename.slice(0,dash_index)
		job_name_version = basename.slice(dash_index+1, basename.length - dash_index)

		job_main = File.join(path + "/main.rb")
		if(File.exists?(job_main) == false)
			Bsl::Logger::Log "Job entry point does not exists - #{job_main}"
			next
		end

		require job_main

		job = nil
		begin
			job = Thor::Jobs::const_get(job_name)
		rescue
			Bsl::Logger::Log "Unable to find class 'Thor::Jobs::#{job_name}' in file '#{job_main}'"
			next
		end

		job_author = job.author
		job_description = job.description
		job_license = job.license
		job_version = job.version

		if(job_version != job_name_version)
			Bsl::Logger::Log "Job '#{job_name}' inconsistent versions, filename: '#{job_name_version}', class: '#{job_version}, skipping.'"
			next
		end

		Bsl::Logger::Log "Adding job, name: '#{job_name}', version: '#{job_version}', basename: '#{basename}'"
		res_jobs[basename] = StructJob.new(job, job_main)
	end

	return res_jobs
end

#load_boot_file(boot_file) ⇒ Object

Load boot file if specified



133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/ThorNode.rb', line 133

def load_boot_file(boot_file)
	Bsl::Logger::Log "Loading boot file '#{boot_file}'."
	if(File.exists?(boot_file) == false)
		Bsl::Logger::Log "Boot file '#{boot_file}' is not valid file path!"
		return nil
	end
	
    require boot_file
    basename = File.basename(boot_file, ".*")
    return ThorBoot::const_get(basename)::boot_jobs
end

#main ⇒ Object

Main entry-point



175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
 
# File 'lib/ThorNode.rb', line 175

def main
	super()

	# Refresh jobs
	new_jobs = enumerate_jobs(options[:jobs_dir])
	@jobs_lock. synchronize {
		@jobs = new_jobs
	}

	# Run loop while exit is not requested
	while(@request_exit == false)
		begin
			amqp_start()
		rescue SystemExit, Interrupt
			Bsl::Logger::Log "Received interrupt, quitting!"
			@request_exit = true
			amqp_stop()
		rescue Exception => e
			amqp_handle_failure(e)
		end
		puts "."
	end
end

#start_boot_time_jobs(boot_time_jobs, amqp) ⇒ Object

Starts boot time jobs



146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
 
# File 'lib/ThorNode.rb', line 146

def start_boot_time_jobs(boot_time_jobs, amqp)
	boot_time_jobs.each do |boot_time_job|
		@jobs_lock.synchronize {
			job_name = boot_time_job[:name]
			job = @jobs[job_name]
			if(job != nil)
				job_klass = job[:klass]
				
				job_opts = options						
				job_opts = job_opts.merge(boot_time_job[:options])						
				job_opts = job_opts.merge({:amqp => {:conn => amqp}})	    
				job_opts.merge!({:amqp => {:conn => amqp}}) # TODO: Use Thor::AmqpNode defaults
				
				if(options[:verbose])
					Bsl::Logger::Log "Starting boot job '#{job_klass.name}'."
					#Bsl::Logger::Log "Job options '#{job_opts.inspect}'."
				end
				
				job_instance = job_klass.new(job_opts)
				job_instance.start(job_opts)
				Bsl::Logger::Log "Boot job '#{job_klass.name}' started."
			else
				Bsl::Logger::Log "Invalid boot job specified, name: '#{job_name}'"
			end
		}
	end
end