Sha256: 62a1c17d4ed4f7104e2f8f4b61ccad7ebd180853b88c1ea949c09861f98cf50b

Contents?: true

Size: 728 Bytes

Versions: 4

Compression:

Stored size: 728 Bytes

Contents

class GooglePubsubEnhancer::Middleware::Publisher

  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

  def call(env)
    @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)
      end
    end
  rescue => ex
    @logger.error("Retry publisher: #{ex}")
    retry
    @app.call(env)
  end

end

Version data entries

4 entries across 4 versions & 1 rubygems

Version Path
google-pubsub-enhancer-0.5.3 lib/google_pubsub_enhancer/middleware/publisher.rb
google-pubsub-enhancer-0.5.2 lib/google_pubsub_enhancer/middleware/publisher.rb
google-pubsub-enhancer-0.5.1 lib/google_pubsub_enhancer/middleware/publisher.rb
google-pubsub-enhancer-0.5.0 lib/google_pubsub_enhancer/middleware/publisher.rb