Class: Gcpc::Interceptors::Subscriber::DecodeInterceptor

Inherits:
Subscriber::BaseInterceptor
  • Object
show all
Defined in:
lib/gcpc/interceptors/subscriber/decode_interceptor.rb

Overview

‘DecodeInterceptor` decodes the message according to the strategy and sets it in the attributes.

Defined Under Namespace

Classes: BaseStrategy, JSONStrategy

Instance Method Summary collapse

Constructor Details

#initialize(strategy:, logger: Logger.new(STDOUT), ignore_on_error: true) ⇒ DecodeInterceptor

Returns a new instance of DecodeInterceptor.

Parameters:

  • strategy (BaseStrategy)
  • logger (Logger) (defaults to: Logger.new(STDOUT))
  • ignore_on_error (Boolean) (defaults to: true) —

    Ignore the message when decode failed



24
25
26
27
28
 
# File 'lib/gcpc/interceptors/subscriber/decode_interceptor.rb', line 24

def initialize(strategy:, logger: Logger.new(STDOUT), ignore_on_error: true)
  @strategy        = strategy
  @logger          = logger
  @ignore_on_error = ignore_on_error
end

Instance Method Details

#handle(data, attributes, message, &block) {|m, attributes, message| ... } ⇒ Object

Parameters:

  • data (String)
  • attributes (Hash)
  • message (Google::Cloud::Pubsub::ReceivedMessage)
  • block (Proc)

Yields:

  • (m, attributes, message) 


34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/gcpc/interceptors/subscriber/decode_interceptor.rb', line 34

def handle(data, attributes, message, &block)
  begin
    m = @strategy.decode(data, attributes, message)
  rescue => e
    @logger.error(e)

    if @ignore_on_error
      @logger.info("Ack a message{data=#{message.data}, attributes=#{message.attributes}} because it can't be decoded!")
      message.ack!  # Ack immediately if decode failed
      return
    else
      raise e
    end
  end

  yield m, attributes, message
end