# Copyright 2015 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
require "google/cloud/errors"
require "google/cloud/pubsub/service"
require "google/cloud/pubsub/credentials"
require "google/cloud/pubsub/topic"
require "google/cloud/pubsub/batch_publisher"
require "google/cloud/pubsub/snapshot"
module Google
module Cloud
module Pubsub
##
# # Project
#
# Represents the project that pubsub messages are pushed to and pulled
# from. {Topic} is a named resource to which messages are sent by
# publishers. {Subscription} is a named resource representing the stream
# of messages from a single, specific topic, to be delivered to the
# subscribing application. {Message} is a combination of data and
# attributes that a publisher sends to a topic and is eventually delivered
# to subscribers.
#
# See {Google::Cloud#pubsub}
#
# @example
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
#
# topic = pubsub.topic "my-topic"
# topic.publish "task completed"
#
class Project
##
# @private The Service object.
attr_accessor :service
##
# @private Creates a new Pub/Sub Project instance.
def initialize service
@service = service
end
# The Pub/Sub project connected to.
#
# @example
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new(
# project_id: "my-project",
# credentials: "/path/to/keyfile.json"
# )
#
# pubsub.project_id #=> "my-project"
#
def project_id
service.project
end
alias project project_id
##
# Retrieves topic by name.
#
# @param [String] topic_name Name of a topic.
# @param [String] project If the topic belongs to a project other than
# the one currently connected to, the alternate project ID can be
# specified here. Optional.
# @param [Boolean] skip_lookup Optionally create a {Topic} object
# without verifying the topic resource exists on the Pub/Sub service.
# Calls made on this object will raise errors if the topic resource
# does not exist. Default is `false`. Optional.
# @param [Hash] async A hash of values to configure the topic's
# {AsyncPublisher} that is created when {Topic#publish_async}
# is called. Optional.
#
# Hash keys and values may include the following:
#
# * `:max_bytes` (Integer) The maximum size of messages to be
# collected before the batch is published. Default is 10,000,000
# (10MB).
# * `:max_messages` (Integer) The maximum number of messages to be
# collected before the batch is published. Default is 1,000.
# * `:interval` (Numeric) The number of seconds to collect messages
# before the batch is published. Default is 0.25.
# * `:threads` (Hash) The number of threads to create to handle
# concurrent calls by the publisher:
# * `:publish` (Integer) The number of threads used to publish
# messages. Default is 4.
# * `:callback` (Integer) The number of threads to handle the
# published messages' callbacks. Default is 8.
#
# @return [Google::Cloud::Pubsub::Topic, nil] Returns `nil` if topic
# does not exist.
#
# @example
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
# topic = pubsub.topic "existing-topic"
#
# @example By default `nil` will be returned if topic does not exist.
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
# topic = pubsub.topic "non-existing-topic" # nil
#
# @example Create topic in a different project with the `project` flag.
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
# topic = pubsub.topic "another-topic", project: "another-project"
#
# @example Skip the lookup against the service with `skip_lookup`:
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
# topic = pubsub.topic "another-topic", skip_lookup: true
#
# @example Configuring AsyncPublisher to increase concurrent callbacks:
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
# topic = pubsub.topic "my-topic",
# async: { threads: { callback: 16 } }
#
# topic.publish_async "task completed" do |result|
# if result.succeeded?
# log_publish_success result.data
# else
# log_publish_failure result.data, result.error
# end
# end
#
# topic.async_publisher.stop.wait!
#
def topic topic_name, project: nil, skip_lookup: nil, async: nil
ensure_service!
options = { project: project }
return Topic.new_lazy(topic_name, service, options) if skip_lookup
grpc = service.get_topic topic_name
Topic.from_grpc grpc, service, async: async
rescue Google::Cloud::NotFoundError
nil
end
alias get_topic topic
alias find_topic topic
##
# Creates a new topic.
#
# @param [String] topic_name Name of a topic.
# @param [Hash] async A hash of values to configure the topic's
# {AsyncPublisher} that is created when {Topic#publish_async}
# is called. Optional.
#
# Hash keys and values may include the following:
#
# * `:max_bytes` (Integer) The maximum size of messages to be
# collected before the batch is published. Default is 10,000,000
# (10MB).
# * `:max_messages` (Integer) The maximum number of messages to be
# collected before the batch is published. Default is 1,000.
# * `:interval` (Numeric) The number of seconds to collect messages
# before the batch is published. Default is 0.25.
# * `:threads` (Hash) The number of threads to create to handle
# concurrent calls by the publisher:
# * `:publish` (Integer) The number of threads used to publish
# messages. Default is 4.
# * `:callback` (Integer) The number of threads to handle the
# published messages' callbacks. Default is 8.
#
# @return [Google::Cloud::Pubsub::Topic]
#
# @example
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
# topic = pubsub.create_topic "my-topic"
#
def create_topic topic_name, async: nil
ensure_service!
grpc = service.create_topic topic_name
Topic.from_grpc grpc, service, async: async
end
alias new_topic create_topic
##
# Retrieves a list of topics for the given project.
#
# @param [String] token The `token` value returned by the last call to
# `topics`; indicates that this is a continuation of a call, and that
# the system should return the next page of data.
# @param [Integer] max Maximum number of topics to return.
#
# @return [Array<Google::Cloud::Pubsub::Topic>] (See
# {Google::Cloud::Pubsub::Topic::List})
#
# @example
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
#
# topics = pubsub.topics
# topics.each do |topic|
# puts topic.name
# end
#
# @example Retrieve all topics: (See {Topic::List#all})
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
#
# topics = pubsub.topics
# topics.all do |topic|
# puts topic.name
# end
#
def topics token: nil, max: nil
ensure_service!
options = { token: token, max: max }
grpc = service.list_topics options
Topic::List.from_grpc grpc, service, max
end
alias find_topics topics
alias list_topics topics
##
# Retrieves subscription by name.
#
# @param [String] subscription_name Name of a subscription.
# @param [String] project If the subscription belongs to a project other
# than the one currently connected to, the alternate project ID can be
# specified here.
# @param [Boolean] skip_lookup Optionally create a {Subscription} object
# without verifying the subscription resource exists on the Pub/Sub
# service. Calls made on this object will raise errors if the service
# resource does not exist. Default is `false`.
#
# @return [Google::Cloud::Pubsub::Subscription, nil] Returns `nil` if
# the subscription does not exist
#
# @example
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
#
# sub = pubsub.subscription "my-sub"
# sub.name #=> "projects/my-project/subscriptions/my-sub"
#
# @example Skip the lookup against the service with `skip_lookup`:
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
#
# # No API call is made to retrieve the subscription information.
# sub = pubsub.subscription "my-sub", skip_lookup: true
# sub.name #=> "projects/my-project/subscriptions/my-sub"
#
def subscription subscription_name, project: nil, skip_lookup: nil
ensure_service!
options = { project: project }
if skip_lookup
return Subscription.new_lazy subscription_name, service, options
end
grpc = service.get_subscription subscription_name
Subscription.from_grpc grpc, service
rescue Google::Cloud::NotFoundError
nil
end
alias get_subscription subscription
alias find_subscription subscription
##
# Retrieves a list of subscriptions for the given project.
#
# @param [String] token A previously-returned page token representing
# part of the larger set of results to view.
# @param [Integer] max Maximum number of subscriptions to return.
#
# @return [Array<Google::Cloud::Pubsub::Subscription>] (See
# {Google::Cloud::Pubsub::Subscription::List})
#
# @example
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
#
# subs = pubsub.subscriptions
# subs.each do |sub|
# puts sub.name
# end
#
# @example Retrieve all subscriptions: (See {Subscription::List#all})
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
#
# subs = pubsub.subscriptions
# subs.all do |sub|
# puts sub.name
# end
#
def subscriptions token: nil, max: nil
ensure_service!
options = { token: token, max: max }
grpc = service.list_subscriptions options
Subscription::List.from_grpc grpc, service, max
end
alias find_subscriptions subscriptions
alias list_subscriptions subscriptions
##
# Retrieves a list of snapshots for the given project.
#
# @param [String] token A previously-returned page token representing
# part of the larger set of results to view.
# @param [Integer] max Maximum number of snapshots to return.
#
# @return [Array<Google::Cloud::Pubsub::Snapshot>] (See
# {Google::Cloud::Pubsub::Snapshot::List})
#
# @example
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
#
# snapshots = pubsub.snapshots
# snapshots.each do |snapshot|
# puts snapshot.name
# end
#
# @example Retrieve all snapshots: (See {Snapshot::List#all})
# require "google/cloud/pubsub"
#
# pubsub = Google::Cloud::Pubsub.new
#
# snapshots = pubsub.snapshots
# snapshots.all do |snapshot|
# puts snapshot.name
# end
#
def snapshots token: nil, max: nil
ensure_service!
options = { token: token, max: max }
grpc = service.list_snapshots options
Snapshot::List.from_grpc grpc, service, max
end
alias find_snapshots snapshots
alias list_snapshots snapshots
protected
##
# @private Raise an error unless an active connection to the service is
# available.
def ensure_service!
raise "Must have active connection to service" unless service
end
##
# Call the publish API with arrays of data data and attrs.
def publish_batch_messages topic_name, batch
grpc = service.publish topic_name, batch.messages
batch.to_gcloud_messages Array(grpc.message_ids)
end
end
end
end
end