Contents

module Cloudist
  class Job
    attr_reader :payload, :reply_queue

    def initialize(payload)
      @payload = payload

      if payload.reply_to
        @reply_queue = ReplyQueue.new(payload.reply_to)
        reply_queue.setup
      else
        @reply_queue = nil
      end
    end

    def id
      payload.id
    end

    def data
      payload.body
    end

    def body
      data
    end

    def log
      Cloudist.log
    end

    def cleanup
      # :noop
    end

    def reply(body, headers = {}, options = {})
      raise ArgumentError, "Reply queue not ready" unless reply_queue

      options = {
        :echo => false
      }.update(options)

      headers = {
        :message_id => payload.id,
        :message_type => "reply"
      }.update(headers)

      reply_payload = Payload.new(body, headers)
      published_headers = reply_queue.publish(reply_payload)

      reply_payload
    end

    # Sends a progress update
    # Inputs: percentage - Integer
    # Optional description, this could be displayed to the user e.g. Resizing image
    def progress(percentage, description = nil)
      reply({:progress => percentage, :description => description}, {:message_type => 'progress'})
    end

    def event(event_name, event_data = {}, options = {})
      event_data ||= {}
      reply(event_data, {:event => event_name, :message_type => 'event'}, options)
    end

    def safely(&blk)
      yield
    rescue Exception => e
      handle_error(e)
    end

    def handle_error(e)
      reply({:exception => e.class.name.to_s, :message => e.message.to_s, :backtrace => e.backtrace}, {:message_type => 'error'})
    end

    def method_missing(meth, *args, &blk)
      if meth.to_s.ends_with?("!")
        event(meth.to_s.gsub(/(!)$/, ''), args.shift)
      else
        super
      end
    end

  end
end

Version data entries

1 entries across 1 versions & 1 rubygems

Version	Path
cloudist-0.5.0	lib/cloudist/job.rb