Class: Karafka::Processing::Scheduler

Inherits:

Object

• Object
• Karafka::Processing::Scheduler

Defined in:

lib/karafka/processing/scheduler.rb

Overview

FIFO scheduler for messages coming from various topics and partitions

Direct Known Subclasses

Karafka::Pro::Processing::Scheduler

Instance Method Summary

• #clear(_group_id) ⇒ Object

  This scheduler does not need to be cleared because it is stateless.

• #initialize(queue) ⇒ Scheduler constructor

  A new instance of Scheduler.

• #manage ⇒ Object

  This scheduler does not have anything to manage as it is a pass through and has no state.

• #schedule_consumption(jobs_array) ⇒ Object (also: #schedule_revocation, #schedule_shutdown)

  Schedules jobs in the fifo order.

Constructor Details

#initialize(queue) ⇒ Scheduler

Returns a new instance of Scheduler.

Parameters:

• queue (Karafka::Processing::JobsQueue) —

  queue where we want to put the jobs

# File 'lib/karafka/processing/scheduler.rb', line 8

def initialize(queue)
  @queue = queue
end

Instance Method Details

#clear(_group_id) ⇒ Object

This scheduler does not need to be cleared because it is stateless

Parameters:

• _group_id (String) —

  Subscription group id

# File 'lib/karafka/processing/scheduler.rb', line 33

def clear(_group_id)
  nil
end

#manage ⇒ Object

This scheduler does not have anything to manage as it is a pass through and has no state

# File 'lib/karafka/processing/scheduler.rb', line 26

def manage
  nil
end

#schedule_consumption(jobs_array) ⇒ Object Also known as: schedule_revocation, schedule_shutdown

Schedules jobs in the fifo order

Parameters:

• jobs_array (Array<Karafka::Processing::Jobs::Base>) —

  jobs we want to schedule

# File 'lib/karafka/processing/scheduler.rb', line 15

def schedule_consumption(jobs_array)
  jobs_array.each do |job|
    @queue << job
  end
end