module Lowkiq

# File lib/lowkiq.rb, line 89
def build_by_node_splitter(number_of_nodes, node_number)
  Lowkiq::Splitters::ByNode.new(
    number_of_nodes,
    node_number,
    Lowkiq.threads_per_node,
  )
end

# File lib/lowkiq.rb, line 85
def build_default_splitter
  Lowkiq::Splitters::Default.new Lowkiq.threads_per_node
end

# File lib/lowkiq.rb, line 72
def build_lag_scheduler
  Schedulers::Lag.new(
    ->() { sleep Lowkiq.poll_interval },
    Queue::ShardMetrics.new(self.server_redis_pool)
  )
end

# File lib/lowkiq.rb, line 79
def build_seq_scheduler
  Schedulers::Seq.new(
    ->() { sleep Lowkiq.poll_interval }
  )
end

# File lib/lowkiq.rb, line 51
def client_redis_pool
  @client_redis_pool ||= ConnectionPool.new(size: client_pool_size, timeout: pool_timeout, &redis)
end

# File lib/lowkiq.rb, line 47
def server_redis_pool
  @server_redis_pool ||= ConnectionPool.new(size: threads_per_node, timeout: pool_timeout, &redis)
end

# File lib/lowkiq.rb, line 55
def server_wrapper
  null = -> (worker, batch, &block) { block.call }
  server_middlewares.reduce(null) do |wrapper, m|
    -> (worker, batch, &block) do
      wrapper.call worker, batch do
        m.call worker, batch, &block
      end
    end
  end
end

# File lib/lowkiq.rb, line 66
def shard_handlers
  self.workers.flat_map do |w|
    ShardHandler.build_many w, self.server_wrapper
  end
end

# File lib/lowkiq/worker.rb, line 33
def client_actions
  Queue::Actions.new client_queue, client_queries
end

# File lib/lowkiq/worker.rb, line 29
def client_queries
  Queue::Queries.new Lowkiq.client_redis_pool, self.queue_name
end

# File lib/lowkiq/worker.rb, line 25
def client_queue
  Queue::Queue.new Lowkiq.client_redis_pool, self.queue_name, self.shards_count
end

# File lib/lowkiq/worker.rb, line 21
def perform(payload)
  fail "not implemented"
end

# File lib/lowkiq/worker.rb, line 37
def perform_async(batch)
  client_queue.push batch
end

# File lib/lowkiq/worker.rb, line 17
def retry_in(retry_count)
  (retry_count ** 4) + 15 + (rand(30) * (retry_count + 1))
end