class GooglePubsubEnhancer::Middleware::Publisher

Public Class Methods

new(app, opts={}) click to toggle source
# File lib/google_pubsub_enhancer/middleware/publisher.rb, line 3
def initialize(app, opts={})
  @app = app
  @short_topic_name = opts[:short_topic_name] || raise
  @full_topic_name = GooglePubsubEnhancer.name_by('topics',@short_topic_name)
  @messages_key = opts[:messages] || raise
  @logger = opts[:logger] || Logger.new(STDOUT)
  @google_cloud_pubsub ||= Google::Cloud::Pubsub.new
end

Public Instance Methods

call(env) click to toggle source
# File lib/google_pubsub_enhancer/middleware/publisher.rb, line 12
def call(env)
  begin
    @logger.debug("#{env[@messages_key].length} messages published")
    @google_cloud_pubsub.publish(@full_topic_name) do |publisher|
      [*env[@messages_key]].each do |m|
        publisher.publish(m, {recordId: Digest::MD5.hexdigest(m)} )
      end
    end
  rescue => ex
    @logger.error("Retry publisher: #{ex}")
    retry
  end
  @app.call(env)
end