Task

# File lib/symphony/task.rb, line 416
def check_for_idle_timeout

        # If it's unset, it means it's running now
        return unless self.last_worked && self.exit_on_idle?

        seconds_idle = Time.now - self.last_worked
        self.log.debug "%p: idle %0.2fs" % [ self.class, seconds_idle ]

        if seconds_idle > self.class.idle_timeout
                self.log.debug "Sending stop signal due to idle timeout"
                self.stop_gracefully
        end
end

# File lib/symphony/task.rb, line 439
def handle_signal( sig )
        self.log.debug "Handling signal %s" % [ sig ]
        case sig
        when :TERM
                self.on_terminate
        when :INT
                self.on_interrupt
        when :HUP
                self.on_hangup
        else
                self.log.warn "Unhandled signal %s" % [ sig ]
        end
end

# File lib/symphony/task.rb, line 475
def make_consumer_tag
        return "%s.%s.%d" % [
                self.queue_name,
                Socket.gethostname.gsub(/\..*$/, ''),
                Process.pid,
        ]
end

# File lib/symphony/task.rb, line 517
def on_hangup
        self.log.info "Hangup signal."
        self.restart
end

# File lib/symphony/task.rb, line 524
def on_terminate
        self.log.debug "Signalled to shut down."

        if self.shutting_down?
                self.stop_immediately
        else
                self.stop_gracefully
        end
end

# File lib/symphony/task.rb, line 456
def preprocess_payload( payload, metadata )
        self.log.debug "Got a %0.2fK %s payload" %
                [ payload.bytesize / 1024.0, metadata[:content_type] ]
        work_payload = case metadata[:content_type]
                when 'application/x-msgpack'
                        MessagePack.unpack( payload )
                when 'application/json', 'text/javascript'
                        JSON.parse( payload )
                when 'application/x-yaml', 'text/x-yaml'
                        YAML.load( payload )
                else
                        payload
                end

        return work_payload
end

# File lib/symphony/task.rb, line 505
def procname
        return "%s %s: Symphony: %p (%s) -> %s" % [
                RUBY_ENGINE,
                RUBY_VERSION,
                self.class,
                self.class.work_model,
                self.class.queue_name
        ]
end

# File lib/symphony/task.rb, line 338
def restart
        self.restarting = true
        self.log.warn "Restarting..."

        if Symphony.config.reload
                self.log.info "  config reloaded"
        else
                self.log.info "  no config changes"
        end

        self.log.info "  resetting queue"
        Symphony::Queue.reset
        self.queue.shutdown
end

# File lib/symphony/task.rb, line 315
def start
        rval = nil

        Process.setproctitle( self.procname )

        begin
                self.restarting = false
                rval = self.with_signal_handler( *SIGNALS ) do
                        self.start_handling_messages
                end
        end while self.restarting?

        return rval ? 0 : 1

rescue Exception => err
        self.log.fatal "%p in %p: %s: %s" % [ err.class, self.class, err.message, err.backtrace.first ]
        self.log.debug { '  ' + err.backtrace.join("  \n") }

        return :software
end

# File lib/symphony/task.rb, line 372
def start_handling_messages
        oneshot = self.class.work_model == :oneshot

        rval = nil
        self.queue.wait_for_message( oneshot ) do |payload, metadata|
                begin
                        self.last_worked = nil
                        work_payload = self.preprocess_payload( payload, metadata )

                        rval = if self.class.timeout
                                self.work_with_timeout( work_payload, metadata )
                        else
                                self.work( work_payload, metadata )
                        end
                ensure
                        self.last_worked = Time.now
                end

                rval
        end

        return rval
end

# File lib/symphony/task.rb, line 399
def start_signal_handler
        @signal_handler = Thread.new do
                Thread.current.abort_on_exception = true
                loop do
                        # self.log.debug "Signal handler: waiting for new signals in the queue."
                        self.wait_for_signals( IDLE_CHECK_INTERVAL )
                        self.check_for_idle_timeout
                end
        end
rescue => err
        self.log.fatal "Signal handler thread crashed: %p: %s" % [ err.class, err.message ]
        self.stop_immediately
end

# File lib/symphony/task.rb, line 364
def stop_gracefully
        self.log.warn "Attempting to shut down gracefully."
        self.shutting_down = true
        self.queue.shutdown
end

# File lib/symphony/task.rb, line 355
def stop_immediately
        self.log.warn "Already in shutdown -- halting immediately."
        self.shutting_down = true
        self.ignore_signals( *SIGNALS )
        self.queue.halt
end

# File lib/symphony/task.rb, line 432
def stop_signal_handler
        @signal_handler.exit if @signal_handler
end

# File lib/symphony/task.rb, line 485
def work( payload, metadata )
        raise NotImplementedError,
                "%p doesn't implement required method #work" % [ self.class ]
end

# File lib/symphony/task.rb, line 493
def work_with_timeout( payload, metadata )
        Timeout.timeout( self.class.timeout ) do
                return self.work( payload, metadata )
        end
rescue Timeout::Error
        self.log.error "Timed out while performing work"
        raise if self.class.timeout_action == :reject
        return false
end