Contents

require 'forwardable'
require_relative 'strategy'
require_relative './util'
require_relative '../upperkut'

module Upperkut
  module Worker

    def self.included(base)
      base.extend(ClassMethods)
    end

    def process
      items = self.class.fetch_items.collect! do |item|
        item['body']
      end
      perform(items)
    end

    module ClassMethods
      extend Forwardable

      def_delegators :setup, :strategy
      def_delegators :strategy, :push_items, :size, :latency

      def push_items(items)
        strategy.push_items(items)
      end

      def fetch_items
        strategy.fetch_items(setup.batch_size)
      end

      def setup_upperkut(&block)
        yield(setup) if block_given?
      end

      def setup
        @@setup ||=
          begin
            default = Upperkut::Configuration.default.clone
            default.strategy ||= Upperkut::Strategy.new(self, default.redis)
            default
          end
      end
    end
  end
end

Version data entries

1 entries across 1 versions & 1 rubygems

Version	Path
upperkut-0.1.2	lib/upperkut/worker.rb