Class: Karafka::Web::Pro::Commanding::Handlers::Partitions::Executor

Inherits:
Object
  • Object
show all
Defined in:
lib/karafka/web/pro/commanding/handlers/partitions/executor.rb

Overview

Selects proper command for running prior to next poll and executes its flow

Instance Method Summary collapse

Instance Method Details

#call(listener, client, request) ⇒ Object

Parameters:

  • listener (Karafka::Connection::Listener)
  • client (Karafka::Connection::Client)
  • request (Request) 


20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/karafka/web/pro/commanding/handlers/partitions/executor.rb', line 20

def call(listener, client, request)
  command = case request.name
            when Commanding::Commands::Partitions::Resume.name
              Commands::Resume
            when Commanding::Commands::Partitions::Pause.name
              Commands::Pause
            when Commanding::Commands::Partitions::Seek.name
              Commands::Seek
            else
              raise ::Karafka::Errors::UnsupportedCaseError, request.name
            end

  command.new(
    listener,
    client,
    request
  ).call
end

#reject(request) ⇒ Object

Publishes the reject event as the final result. Used to indicate, that given request will not be processed because it is not valid anymore.

Parameters:



43
44
45
46
47
48
49
 
# File 'lib/karafka/web/pro/commanding/handlers/partitions/executor.rb', line 43

def reject(request)
  Dispatcher.result(
    request.name,
    process_id,
    request.to_h.merge(status: 'rebalance_rejected')
  )
end