Class: Google::Cloud::Pubsub::Subscription

Inherits:
Object
  • Object
show all
Defined in:
lib/google/cloud/pubsub/subscription.rb,
lib/google/cloud/pubsub/subscription/list.rb

Overview

Subscription

A named resource representing the stream of messages from a single, specific Topic, to be delivered to the subscribing application.

Examples:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new

sub = pubsub.subscription "my-topic-sub"
msgs = sub.pull
msgs.each { |msg| msg.acknowledge! }

Defined Under Namespace

Classes: List

Instance Method Summary collapse

Instance Method Details

#acknowledge(*messages) ⇒ Object Also known as: ack

Acknowledges receipt of a message. After an ack, the Pub/Sub system can remove the message from the subscription. Acknowledging a message whose ack deadline has expired may succeed, although the message may have been sent again. Acknowledging a message more than once will not result in an error. This is only used for messages received via pull.

Examples:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new

sub = pubsub.subscription "my-topic-sub"
messages = sub.pull
sub.acknowledge messages

Parameters:



339
340
341
342
343
344
345
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 339

def acknowledge *messages
  ack_ids = coerce_ack_ids messages
  return true if ack_ids.empty?
  ensure_service!
  service.acknowledge name, *ack_ids
  true
end

#deadline ⇒ Object

This value is the maximum number of seconds after a subscriber receives a message before the subscriber should acknowledge the message.



95
96
97
98
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 95

def deadline
  ensure_grpc!
  @grpc.ack_deadline_seconds
end

#delay(new_deadline, *messages) ⇒ Object

Modifies the acknowledge deadline for messages.

This indicates that more time is needed to process the messages, or to make the messages available for redelivery if the processing was interrupted.

Examples:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new

sub = pubsub.subscription "my-topic-sub"
messages = sub.pull
sub.delay 120, messages

Parameters:

  • new_deadline (Integer) —

    The new ack deadline in seconds from the time this request is sent to the Pub/Sub system. Must be >= 0. For example, if the value is 10, the new ack deadline will expire 10 seconds after the call is made. Specifying 0 may immediately make the message available for another pull request.

  • messages (ReceivedMessage, String) —

    One or more ReceivedMessage objects or ack_id values.



372
373
374
375
376
377
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 372

def delay new_deadline, *messages
  ack_ids = coerce_ack_ids messages
  ensure_service!
  service.modify_ack_deadline name, ack_ids, new_deadline
  true
end

#delete ⇒ Boolean

Deletes an existing subscription. All pending messages in the subscription are immediately dropped.

Examples:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new

sub = pubsub.subscription "my-topic-sub"
sub.delete

Returns:

  • (Boolean) —

    Returns true if the subscription was deleted.



172
173
174
175
176
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 172

def delete
  ensure_service!
  service.delete_subscription name
  true
end

#endpoint ⇒ Object

Returns the URL locating the endpoint to which messages should be pushed.



103
104
105
106
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 103

def endpoint
  ensure_grpc!
  @grpc.push_config.push_endpoint if @grpc.push_config
end

#endpoint=(new_endpoint) ⇒ Object

Sets the URL locating the endpoint to which messages should be pushed.



110
111
112
113
114
115
116
117
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 110

def endpoint= new_endpoint
  ensure_service!
  service.modify_push_config name, new_endpoint, {}
  @grpc.push_config = Google::Pubsub::V1::PushConfig.new(
    push_endpoint: new_endpoint,
    attributes: {}
  ) if @grpc
end

#exists? ⇒ Boolean

Determines whether the subscription exists in the Pub/Sub service.

Examples:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new

sub = pubsub.subscription "my-topic-sub"
sub.exists? #=> true

Returns:

  • (Boolean) 


130
131
132
133
134
135
136
137
138
139
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 130

def exists?
  # Always true if we have a grpc object
  return true unless @grpc.nil?
  # If we have a value, return it
  return @exists unless @exists.nil?
  ensure_grpc!
  @exists = !@grpc.nil?
rescue Google::Cloud::NotFoundError
  @exists = false
end

#listen(max: 100, autoack: false, delay: 1) {|msg| ... } ⇒ Object

Poll the backend for new messages. This runs a loop to ping the API, blocking indefinitely, yielding retrieved messages as they are received.

Examples:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new

sub = pubsub.subscription "my-topic-sub"
sub.listen do |msg|
  # process msg
end

Limit number of messages pulled per API request with max:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new

sub = pubsub.subscription "my-topic-sub"
sub.listen max: 20 do |msg|
  # process msg
end

Automatically acknowledge messages with autoack:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new

sub = pubsub.subscription "my-topic-sub"
sub.listen autoack: true do |msg|
  # process msg
end

Parameters:

  • max (Integer) —

    The maximum number of messages to return for this request. The Pub/Sub system may return fewer than the number specified. The default value is 100, the maximum value is 1000.

  • autoack (Boolean) —

    Automatically acknowledge the message as it is pulled. The default value is false.

  • delay (Number) —

    The number of seconds to pause between requests when the Google Cloud service has no messages to return. The default value is 1.

Yields:

  • (msg) —

    a block for processing new messages

Yield Parameters:



308
309
310
311
312
313
314
315
316
317
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 308

def listen max: 100, autoack: false, delay: 1
  loop do
    msgs = wait_for_messages max: max, autoack: autoack
    if msgs.any?
      msgs.each { |msg| yield msg }
    else
      sleep delay
    end
  end
end

#name ⇒ Object

The name of the subscription.



69
70
71
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 69

def name
  @grpc ? @grpc.name : @name
end

#policy(force: nil) {|policy| ... } ⇒ Policy

Gets the Cloud IAM access control policy for this subscription.

Examples:

Policy values are memoized to reduce the number of API calls:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new
sub = pubsub.subscription "my-subscription"

policy = sub.policy # API call
policy_2 = sub.policy # No API call

Use force to retrieve the latest policy from the service:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new
sub = pubsub.subscription "my-subscription"

policy = sub.policy force: true # API call
policy_2 = sub.policy force: true # API call

Update the policy by passing a block:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new
sub = pubsub.subscription "my-subscription"

policy = sub.policy do |p|
  p.add "roles/owner", "user:owner@example.com"
end # 2 API calls

Parameters:

  • force (Boolean) —

    Force the latest policy to be retrieved from the Pub/Sub service when true. Otherwise the policy will be memoized to reduce the number of API calls made to the Pub/Sub service. The default is false.

Yields:

  • (policy) —

    A block for updating the policy. The latest policy will be read from the Pub/Sub service and passed to the block. After the block completes, the modified policy will be written to the service.

Yield Parameters:

  • policy (Policy) —

    the current Cloud IAM Policy for this subscription

Returns:

  • (Policy) —

    the current Cloud IAM Policy for this subscription

See Also:



428
429
430
431
432
433
434
435
436
437
438
439
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 428

def policy force: nil
  @policy = nil if force || block_given?
  @policy ||= begin
    ensure_service!
    grpc = service.get_subscription_policy name
    Policy.from_grpc grpc
  end
  return @policy unless block_given?
  p = @policy.deep_dup
  yield p
  self.policy = p
end

#policy=(new_policy) ⇒ Object

Updates the Cloud IAM access control policy for this subscription. The policy should be read from #policy. See Policy for an explanation of the policy etag property and how to modify policies.

You can also update the policy by passing a block to #policy, which will call this method internally after the block completes.

Examples:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new
sub = pubsub.subscription "my-subscription"

policy = sub.policy # API call

policy.add "roles/owner", "user:owner@example.com"

sub.policy = policy # API call

Parameters:

  • new_policy (Policy) —

    a new or modified Cloud IAM Policy for this subscription

See Also:



468
469
470
471
472
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 468

def policy= new_policy
  ensure_service!
  grpc = service.set_subscription_policy name, new_policy.to_grpc
  @policy = Policy.from_grpc grpc
end

#pull(immediate: true, max: 100, autoack: false) ⇒ Array<Google::Cloud::Pubsub::ReceivedMessage>

Pulls messages from the server. Returns an empty list if there are no messages available in the backlog. Raises an ApiError with status UNAVAILABLE if there are too many concurrent pull requests pending for the given subscription.

Examples:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new

sub = pubsub.subscription "my-topic-sub"
sub.pull.each { |msg| msg.acknowledge! }

A maximum number of messages returned can also be specified:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new

sub = pubsub.subscription "my-topic-sub", max: 10
sub.pull.each { |msg| msg.acknowledge! }

The call can block until messages are available:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new

sub = pubsub.subscription "my-topic-sub"
msgs = sub.pull immediate: false
msgs.each { |msg| msg.acknowledge! }

Parameters:

  • immediate (Boolean) —

    When true the system will respond immediately even if it is not able to return messages. When false the system is allowed to wait until it can return least one message. No messages are returned when a request times out. The default value is true.

  • max (Integer) —

    The maximum number of messages to return for this request. The Pub/Sub system may return fewer than the number specified. The default value is 100, the maximum value is 1000.

  • autoack (Boolean) —

    Automatically acknowledge the message as it is pulled. The default value is false.

Returns:



222
223
224
225
226
227
228
229
230
231
232
233
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 222

def pull immediate: true, max: 100, autoack: false
  ensure_service!
  options = { immediate: immediate, max: max }
  list_grpc = service.pull name, options
  messages = Array(list_grpc.received_messages).map do |msg_grpc|
    ReceivedMessage.from_grpc msg_grpc, self
  end
  acknowledge messages if autoack
  messages
rescue Google::Cloud::DeadlineExceededError
  []
end

#test_permissions(*permissions) ⇒ Array<String>

Tests the specified permissions against the Cloud IAM access control policy.

Examples:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new
sub = pubsub.subscription "my-subscription"
perms = sub.test_permissions "pubsub.subscriptions.get",
                             "pubsub.subscriptions.consume"
perms.include? "pubsub.subscriptions.get" #=> true
perms.include? "pubsub.subscriptions.consume" #=> false

Parameters:

  • permissions (String, Array<String>) —

    The set of permissions to check access for. Permissions with wildcards (such as * or storage.*) are not allowed.

    The permissions that can be checked on a subscription are:

    • pubsub.subscriptions.consume
    • pubsub.subscriptions.get
    • pubsub.subscriptions.delete
    • pubsub.subscriptions.update
    • pubsub.subscriptions.getIamPolicy
    • pubsub.subscriptions.setIamPolicy

Returns:

  • (Array<String>) —

    The permissions that have access.

See Also:



506
507
508
509
510
511
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 506

def test_permissions *permissions
  permissions = Array(permissions).flatten
  ensure_service!
  grpc = service.test_subscription_permissions name, permissions
  grpc.permissions
end

#topic ⇒ Topic

The Topic from which this subscription receives messages.

Examples:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new

sub = pubsub.subscription "my-topic-sub"
sub.topic.name #=> "projects/my-project/topics/my-topic"

Returns:



86
87
88
89
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 86

def topic
  ensure_grpc!
  Topic.new_lazy @grpc.topic, service
end

#wait_for_messages(max: 100, autoack: false) ⇒ Array<Google::Cloud::Pubsub::ReceivedMessage>

Pulls from the server while waiting for messages to become available. This is the same as:

subscription.pull immediate: false

Examples:

require "google/cloud/pubsub"

pubsub = Google::Cloud::Pubsub.new

sub = pubsub.subscription "my-topic-sub"
msgs = sub.wait_for_messages
msgs.each { |msg| msg.acknowledge! }

Parameters:

  • max (Integer) —

    The maximum number of messages to return for this request. The Pub/Sub system may return fewer than the number specified. The default value is 100, the maximum value is 1000.

  • autoack (Boolean) —

    Automatically acknowledge the message as it is pulled. The default value is false.

Returns:



258
259
260
 
# File 'lib/google/cloud/pubsub/subscription.rb', line 258

def wait_for_messages max: 100, autoack: false
  pull immediate: false, max: max, autoack: autoack
end