Class: Gcloud::Pubsub::Subscription

Inherits:
Object
  • Object
show all
Defined in:
lib/gcloud/pubsub/subscription.rb,
lib/gcloud/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 "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

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 "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

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

Parameters:


358
359
360
361
362
363
364
365
366
367
# File 'lib/gcloud/pubsub/subscription.rb', line 358

def acknowledge *messages
  ack_ids = coerce_ack_ids messages
  ensure_connection!
  resp = connection.acknowledge name, *ack_ids
  if resp.success?
    true
  else
    fail ApiError.from_response(resp)
  end
end

#deadlineObject

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


97
98
99
100
# File 'lib/gcloud/pubsub/subscription.rb', line 97

def deadline
  ensure_gapi!
  @gapi["ackDeadlineSeconds"]
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 "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

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.


395
396
397
398
399
400
401
402
403
404
# File 'lib/gcloud/pubsub/subscription.rb', line 395

def delay new_deadline, *messages
  ack_ids = coerce_ack_ids messages
  ensure_connection!
  resp = connection.modify_ack_deadline name, ack_ids, new_deadline
  if resp.success?
    true
  else
    fail ApiError.from_response(resp)
  end
end

#deleteBoolean

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

Examples:

require "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

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

Returns:

  • (Boolean)

    Returns true if the subscription was deleted.


178
179
180
181
182
183
184
185
186
# File 'lib/gcloud/pubsub/subscription.rb', line 178

def delete
  ensure_connection!
  resp = connection.delete_subscription name
  if resp.success?
    true
  else
    fail ApiError.from_response(resp)
  end
end

#endpointObject

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


105
106
107
108
# File 'lib/gcloud/pubsub/subscription.rb', line 105

def endpoint
  ensure_gapi!
  @gapi["pushConfig"]["pushEndpoint"] if @gapi["pushConfig"]
end

#endpoint=(new_endpoint) ⇒ Object

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


112
113
114
115
116
117
118
119
120
# File 'lib/gcloud/pubsub/subscription.rb', line 112

def endpoint= new_endpoint
  ensure_connection!
  resp = connection.modify_push_config name, new_endpoint, {}
  if resp.success?
    @gapi["pushConfig"]["pushEndpoint"] = new_endpoint if @gapi
  else
    fail ApiError.from_response(resp)
  end
end

#exists?Boolean

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

Examples:

require "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

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

Returns:

  • (Boolean)

134
135
136
137
138
139
140
141
142
143
# File 'lib/gcloud/pubsub/subscription.rb', line 134

def exists?
  # Always true if we have a gapi object
  return true unless @gapi.nil?
  # If we have a value, return it
  return @exists unless @exists.nil?
  ensure_gapi!
  @exists = !@gapi.nil?
rescue NotFoundError
  @exists = false
end

#listen(max: 100, autoack: false, delay: 1) ⇒ 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 "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

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

Limit the number of messages pulled per batch with max:

require "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

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

Automatically acknowledge messages with autoack:

require "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

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.


326
327
328
329
330
331
332
333
334
335
# File 'lib/gcloud/pubsub/subscription.rb', line 326

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

#nameObject

The name of the subscription.


71
72
73
# File 'lib/gcloud/pubsub/subscription.rb', line 71

def name
  @gapi ? @gapi["name"] : @name
end

#policy(force: nil) ⇒ Hash

Gets the access control policy.

By default, the policy values are memoized to reduce the number of API calls to the Pub/Sub service.

Examples:

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

require "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

subscription = pubsub.subscription "my-subscription"
puts subscription.policy["bindings"]
puts subscription.policy["rules"]

Use force to retrieve the latest policy from the service:

require "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

subscription = pubsub.subscription "my-subscription"
policy = subscription.policy force: true

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.

Returns:

  • (Hash)

    Returns a hash that conforms to the following structure:

    { "etag"=>"CAE=", "bindings" => [{ "role" => "roles/viewer", "members" => ["serviceAccount:your-service-account"] }] }


446
447
448
449
450
451
452
453
454
455
# File 'lib/gcloud/pubsub/subscription.rb', line 446

def policy force: nil
  @policy = nil if force
  @policy ||= begin
    ensure_connection!
    resp = connection.get_subscription_policy name
    policy = resp.data
    policy = policy.to_hash if policy.respond_to? :to_hash
    policy
  end
end

#policy=(new_policy) ⇒ Object

Sets the access control policy.

Examples:

require "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

viewer_policy = {
  "bindings" => [{
    "role" => "roles/viewer",
    "members" => ["serviceAccount:your-service-account"]
  }]
}
subscription = pubsub.subscription "my-subscription"
subscription.policy = viewer_policy

Parameters:

  • new_policy (String)

    A hash that conforms to the following structure:

    { "bindings" => [{ "role" => "roles/viewer", "members" => ["serviceAccount:your-service-account"] }] }


485
486
487
488
489
490
491
492
493
494
# File 'lib/gcloud/pubsub/subscription.rb', line 485

def policy= new_policy
  ensure_connection!
  resp = connection.set_subscription_policy name, new_policy
  if resp.success?
    @policy = resp.data["policy"]
    @policy = @policy.to_hash if @policy.respond_to? :to_hash
  else
    fail ApiError.from_response(resp)
  end
end

#pull(immediate: true, max: 100, autoack: false) ⇒ Array<Gcloud::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 "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

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

A maximum number of messages returned can also be specified:

require "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

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

The call can block until messages are available:

require "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

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:


235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
# File 'lib/gcloud/pubsub/subscription.rb', line 235

def pull immediate: true, max: 100, autoack: false
  ensure_connection!
  options = { immediate: immediate, max: max }
  resp = connection.pull name, options
  if resp.success?
    messages = Array(resp.data["receivedMessages"]).map do |gapi|
      ReceivedMessage.from_gapi gapi, self
    end
    acknowledge messages if autoack
    messages
  else
    fail ApiError.from_response(resp)
  end
rescue Faraday::TimeoutError
  []
end

#test_permissions(*permissions) ⇒ Array<String>

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

Examples:

require "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub
sub = pubsub.subscription "my-subscription"
perms = sub.test_permissions "projects.subscriptions.list",
                             "projects.subscriptions.pull"
perms.include? "projects.subscriptions.list" #=> true
perms.include? "projects.subscriptions.pull" #=> false

Parameters:

  • *permissions (String, Array<String>)

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

Returns:

  • (Array<String>)

    The permissions that have access.

See Also:


520
521
522
523
524
525
526
527
528
529
# File 'lib/gcloud/pubsub/subscription.rb', line 520

def test_permissions *permissions
  permissions = Array(permissions).flatten
  ensure_connection!
  resp = connection.test_subscription_permissions name, permissions
  if resp.success?
    Array(resp.data["permissions"])
  else
    fail ApiError.from_response(resp)
  end
end

#topicTopic

The Topic from which this subscription receives messages.

Examples:

require "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

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

Returns:


89
90
91
92
# File 'lib/gcloud/pubsub/subscription.rb', line 89

def topic
  ensure_gapi!
  Topic.new_lazy @gapi["topic"], connection
end

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

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

subscription.pull immediate: false

Examples:

require "gcloud"

gcloud = Gcloud.new
pubsub = gcloud.pubsub

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:


276
277
278
# File 'lib/gcloud/pubsub/subscription.rb', line 276

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