BackgrounDRb::MetaWorker

    # File server/lib/meta_worker.rb, line 92
92:     def self.pool_size(size = nil)
93:       @pool_size = size if size
94:       @pool_size
95:     end

     # File server/lib/meta_worker.rb, line 98
 98:     def self.reload_on_schedule(flag = nil)
 99:       if flag
100:         self.no_auto_load = true
101:         self.reload_flag = true
102:       end
103:     end

     # File server/lib/meta_worker.rb, line 214
214:     def can_dump?(p_object)
215:       begin
216:         Marshal.dump(p_object)
217:         return true
218:       rescue TypeError
219:         return false
220:       rescue
221:         return false
222:       end
223:     end

     # File server/lib/meta_worker.rb, line 317
317:     def check_db_connection
318:       begin
319:         ActiveRecord::Base.verify_active_connections! if defined?(ActiveRecord)
320:       rescue
321:         logger.info($!.to_s)
322:         logger.info($!.backtrace.join("\n"))
323:       end
324:     end

     # File server/lib/meta_worker.rb, line 267
267:     def check_for_enqueued_tasks
268:       if worker_key && !worker_key.empty?
269:         task = BdrbJobQueue.find_next(worker_name.to_s,worker_key.to_s)
270:       else
271:         task = BdrbJobQueue.find_next(worker_name.to_s)
272:       end
273:       return unless task
274:       if self.respond_to? task.worker_method
275:         Thread.current[:persistent_job_id] = task[:id]
276:         Thread.current[:job_key] = task[:job_key]
277:         called_method_arity = self.method(task.worker_method).arity
278:         args = load_data(task.args)
279:         begin
280:           if called_method_arity != 0
281:             self.send(task.worker_method,args)
282:           else
283:             self.send(task.worker_method)
284:           end
285:         rescue
286:           logger.info($!.to_s)
287:           logger.info($!.backtrace.join("\n"))
288:         end
289:       else
290:         task.release_job
291:       end
292:     end

     # File server/lib/meta_worker.rb, line 295
295:     def check_for_timer_events
296:       super
297:       return if @worker_method_triggers.nil? or @worker_method_triggers.empty?
298:       @worker_method_triggers.delete_if { |key,value| value[:trigger].respond_to?(:end_time) && value[:trigger].end_time <= Time.now }
299: 
300:       @worker_method_triggers.each do |key,value|
301:         time_now = Time.now.to_i
302:         if value[:runtime] < time_now
303:           check_db_connection
304:           begin
305:             (t_data = value[:data]) ? send(key,t_data) : send(key)
306:           rescue
307:             logger.info($!.to_s)
308:             logger.info($!.backtrace.join("\n"))
309:           end
310:           t_time = value[:trigger].fire_after_time(Time.now)
311:           value[:runtime] = t_time.to_i
312:         end
313:       end
314:     end

     # File server/lib/meta_worker.rb, line 264
264:     def connection_completed; end

     # File server/lib/meta_worker.rb, line 137
137:     def job_key; Thread.current[:job_key]; end

     # File server/lib/meta_worker.rb, line 151
151:     def load_schedule_from_args
152:       @my_schedule = worker_options[:schedule]
153:       new_load_schedule if @my_schedule
154:     end

     # File server/lib/meta_worker.rb, line 226
226:     def new_load_schedule
227:       @worker_method_triggers = { }
228:       @my_schedule.each do |key,value|
229:         case value[:trigger_args]
230:         when String
231:           cron_args = value[:trigger_args] || "0 0 0 0 0"
232:           trigger = BackgrounDRb::CronTrigger.new(cron_args)
233:           @worker_method_triggers[key] = { :trigger => trigger,:data => value[:data],:runtime => trigger.fire_after_time(Time.now).to_i }
234:         when Hash
235:           trigger = BackgrounDRb::Trigger.new(value[:trigger_args])
236:           @worker_method_triggers[key] = { :trigger => trigger,:data => value[:trigger_args][:data],:runtime => trigger.fire_after_time(Time.now).to_i }
237:         end
238:       end
239:     end

     # File server/lib/meta_worker.rb, line 144
144:     def persistent_job
145:       job_id = Thread.current[:persistent_job_id]
146:       job_id ? BdrbJobQueue.find_by_id(job_id) : nil
147:     end

     # File server/lib/meta_worker.rb, line 183
183:     def process_request(p_data)
184:       user_input = p_data[:data]
185:       if (user_input[:worker_method]).nil? or !respond_to?(user_input[:worker_method])
186:         result = nil
187:         send_response(p_data,result)
188:         return
189:       end
190: 
191:       called_method_arity = self.method(user_input[:worker_method]).arity
192:       result = nil
193: 
194:       Thread.current[:job_key] = user_input[:job_key]
195: 
196:       begin
197:         if called_method_arity != 0
198:           result = self.send(user_input[:worker_method],user_input[:arg])
199:         else
200:           result = self.send(user_input[:worker_method])
201:         end
202:       rescue
203:         logger.info($!.to_s)
204:         logger.info($!.backtrace.join("\n"))
205:       end
206: 
207:       if p_data[:result]
208:         result = "dummy_result" if result.nil?
209:         send_response(p_data,result) if can_dump?(result)
210:       end
211:     end

     # File server/lib/meta_worker.rb, line 165
165:     def receive_data p_data
166:       if p_data[:data][:worker_method] == :exit
167:         exit
168:       end
169:       case p_data[:type]
170:       when :request: process_request(p_data)
171:       when :response: process_response(p_data)
172:       when :get_result: return_result_object(p_data)
173:       end
174:     end

     # File server/lib/meta_worker.rb, line 157
157:     def receive_internal_data data
158:       @tokenizer.extract(data) do |b_data|
159:         data_obj = load_data(b_data)
160:         receive_data(data_obj) if data_obj
161:       end
162:     end

     # File server/lib/meta_worker.rb, line 176
176:     def return_result_object p_data
177:       user_input = p_data[:data]
178:       user_job_key = user_input[:job_key]
179:       send_response(p_data,cache[user_job_key])
180:     end

     # File server/lib/meta_worker.rb, line 243
243:     def send_response input,output
244:       input[:data] = output
245:       input[:type] = :response
246:       begin
247:         send_data(input)
248:       rescue TypeError => e
249:         logger.info(e.to_s)
250:         logger.info(e.backtrace.join("\n"))
251:         input[:data] = "invalid_result_dump_check_log"
252:         send_data(input)
253:       rescue
254:         logger.info($!.to_s)
255:         logger.info($!.backtrace.join("\n"))
256:         input[:data] = "invalid_result_dump_check_log"
257:         send_data(input)
258:       end
259:     end

     # File server/lib/meta_worker.rb, line 262
262:     def unbind; end

     # File server/lib/meta_worker.rb, line 107
107:     def worker_init
108:       raise "Invalid worker name" if !worker_name
109:       Thread.abort_on_exception = true
110: 
111:       log_flag = BDRB_CONFIG[:backgroundrb][:debug_log].nil? ? true : BDRB_CONFIG[:backgroundrb][:debug_load_rails_env]
112: 
113:       # stores the job key of currently running job
114:       Thread.current[:job_key] = nil
115:       @logger = PacketLogger.new(self,log_flag)
116:       @thread_pool = ThreadPool.new(self,pool_size || 20,@logger)
117:       t_worker_key = worker_options && worker_options[:worker_key]
118: 
119:       @cache = ResultStorage.new(worker_name,t_worker_key,BDRB_CONFIG[:backgroundrb][:result_storage])
120: 
121:       if(worker_options && worker_options[:schedule] && no_auto_load)
122:         load_schedule_from_args
123:       elsif(BDRB_CONFIG[:schedules] && BDRB_CONFIG[:schedules][worker_name.to_sym])
124:         @my_schedule = BDRB_CONFIG[:schedules][worker_name.to_sym]
125:         new_load_schedule if @my_schedule
126:       end
127:       if respond_to?(:create)
128:         create_arity = method(:create).arity
129:         (create_arity == 0) ? create : create(worker_options[:data])
130:       end
131:       return if BDRB_CONFIG[:backgroundrb][:persistent_disabled]
132:       delay = BDRB_CONFIG[:backgroundrb][:persistent_delay] || 5
133:       add_periodic_timer(delay.to_i) { check_for_enqueued_tasks }
134:     end

     # File server/lib/meta_worker.rb, line 140
140:     def worker_key; worker_options && worker_options[:worker_key]; end