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) 


45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/karafka/web/pro/commanding/handlers/partitions/executor.rb', line 45

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:



68
69
70
71
72
73
74
 
# File 'lib/karafka/web/pro/commanding/handlers/partitions/executor.rb', line 68

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