Module: Karafka::Pro::Routing::Features::Throttling::Topic

Defined in:
lib/karafka/pro/routing/features/throttling/topic.rb

Overview

Topic throttling API extensions

Instance Method Summary collapse

Instance Method Details

#throttle(**args) ⇒ Object

Just an alias for nice API

Parameters:

  • args (Array) —

    Anything #throttling accepts



41
42
43
 
# File 'lib/karafka/pro/routing/features/throttling/topic.rb', line 41

def throttle(**args)
  throttling(**args)
end

#throttling(limit: Float::INFINITY, interval: 60_000) ⇒ Object

Parameters:

  • limit (Integer) (defaults to: Float::INFINITY) —

    max messages to process in an time interval

  • interval (Integer) (defaults to: 60_000) —

    time interval for processing



15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/karafka/pro/routing/features/throttling/topic.rb', line 15

def throttling(
  limit: Float::INFINITY,
  interval: 60_000
)
  # Those settings are used for validation
  @throttling ||= begin
    config = Config.new(
      active: limit != Float::INFINITY,
      limit: limit,
      interval: interval
    )

    # If someone defined throttling setup, we need to create appropriate filter for it
    # and inject it via filtering feature
    if config.active?
      factory = ->(*) { Pro::Processing::Filters::Throttler.new(limit, interval) }
      filter(factory)
    end

    config
  end
end

#throttling? ⇒ Boolean

Returns is a given job throttled.

Returns:

  • (Boolean) —

    is a given job throttled



46
47
48
 
# File 'lib/karafka/pro/routing/features/throttling/topic.rb', line 46

def throttling?
  throttling.active?
end

#to_h ⇒ Hash

Returns topic with all its native configuration options plus throttling.

Returns:

  • (Hash) —

    topic with all its native configuration options plus throttling



51
52
53
54
55
 
# File 'lib/karafka/pro/routing/features/throttling/topic.rb', line 51

def to_h
  super.merge(
    throttling: throttling.to_h
  ).freeze
end