Contents

require 'google/cloud/pubsub'

module Fluent
  module GcloudPubSub
    class Error < StandardError; end

    class Publisher
      def initialize(project, key, topic, autocreate_topic)
        pubsub = Google::Cloud::Pubsub.new project: project, keyfile: key

        @client = pubsub.topic topic, autocreate: autocreate_topic
        raise Fluent::GcloudPubSub::Error.new "topic:#{topic} does not exist." if @client.nil?
      end

      def publish(messages)
        @client.publish do |batch|
          messages.each do |m|
            batch.publish m
          end
        end
      end
    end

    class Subscriber
      def initialize(project, key, topic, subscription)
        pubsub = Google::Cloud::Pubsub.new project: project, keyfile: key
        topic = pubsub.topic topic
        @client = topic.subscription subscription
        raise Fluent::GcloudPubSub::Error.new "subscription:#{subscription} does not exist." if @client.nil?
      end

      def pull(immediate, max)
        @client.pull immediate: immediate, max: max
      end

      def acknowledge(messages)
        @client.acknowledge messages
      end
    end
  end
end

Version data entries

1 entries across 1 versions & 1 rubygems

Version	Path
fluent-plugin-gcloud-pubsub-custom-0.3.0	lib/fluent/plugin/gcloud_pubsub/client.rb