Class: Google::Cloud::PubSub::AsyncPublisher

Inherits:
Object
  • Object
show all
Includes:
MonitorMixin
Defined in:
lib/google/cloud/pubsub/async_publisher.rb,
lib/google/cloud/pubsub/async_publisher/batch.rb

Overview

Used to publish multiple messages in batches to a topic. See Topic#async_publisher

Examples:

require "google/cloud/pubsub"

pubsub = Google::Cloud::PubSub.new

topic = pubsub.topic "my-topic"
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!

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#callback_threads ⇒ Numeric (readonly)

The number of threads to handle the published messages' callbacks. Default is 4.

Returns:

  • (Numeric) —

    the current value of callback_threads



59
60
61
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 59

def callback_threads
  @callback_threads
end

#interval ⇒ Numeric (readonly)

The number of seconds to collect messages before the batch is published. Default is 0.01.

Returns:

  • (Numeric) —

    the current value of interval



59
60
61
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 59

def interval
  @interval
end

#max_bytes ⇒ Integer (readonly)

The maximum size of messages to be collected before the batch is published. Default is 1,000,000 (1MB).

Returns:

  • (Integer) —

    the current value of max_bytes



59
60
61
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 59

def max_bytes
  @max_bytes
end

#max_messages ⇒ Integer (readonly)

The maximum number of messages to be collected before the batch is published. Default is 100.

Returns:

  • (Integer) —

    the current value of max_messages



59
60
61
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 59

def max_messages
  @max_messages
end

#publish_threads ⇒ Numeric (readonly)

The number of threads used to publish messages. Default is 2.

Returns:

  • (Numeric) —

    the current value of publish_threads



59
60
61
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 59

def publish_threads
  @publish_threads
end

#topic_name ⇒ String (readonly)

The name of the topic the messages are published to. The value is a fully-qualified topic name in the form projects/{project_id}/topics/{topic_id}.

Returns:

  • (String) —

    the current value of topic_name



59
60
61
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 59

def topic_name
  @topic_name
end

Instance Method Details

#enable_message_ordering! ⇒ Object

Enables message ordering for messages with ordering keys. When enabled, messages published with the same ordering_key will be delivered in the order they were published.

See #message_ordering?. See Topic#publish_async, Subscription#listen, and Message#ordering_key.



241
242
243
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 241

def enable_message_ordering!
  synchronize { @ordered = true }
end

#flush ⇒ AsyncPublisher

Forces all messages in the current batch to be published immediately.

Returns:



208
209
210
211
212
213
214
215
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 208

def flush
  synchronize do
    publish_batches!
    @cond.signal
  end

  self
end

#message_ordering? ⇒ Boolean

Whether message ordering for messages with ordering keys has been enabled. When enabled, messages published with the same ordering_key will be delivered in the order they were published. When disabled, messages may be delivered in any order.

See #enable_message_ordering!. See Topic#publish_async, Subscription#listen, and Message#ordering_key.

Returns:

  • (Boolean) 


256
257
258
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 256

def message_ordering?
  synchronize { @ordered }
end

#publish(data = nil, attributes = nil, ordering_key: nil, **extra_attrs) {|result| ... } ⇒ Object

Add a message to the async publisher to be published to the topic. Messages will be collected in batches and published together. See Topic#publish_async

Parameters:

  • data (String, File) (defaults to: nil) —

    The message payload. This will be converted to bytes encoded as ASCII-8BIT.

  • attributes (Hash) (defaults to: nil) —

    Optional attributes for the message.

  • ordering_key (String) (defaults to: nil) —

    Identifies related messages for which publish order should be respected.

Yields:

  • (result) —

    the callback for when the message has been published

Yield Parameters:

  • result (PublishResult) —

    the result of the asynchronous publish

Raises:



122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 122

def publish data = nil, attributes = nil, ordering_key: nil, **extra_attrs, &callback
  msg = Convert.pubsub_message data, attributes, ordering_key, extra_attrs

  synchronize do
    raise AsyncPublisherStopped if @stopped
    raise OrderedMessagesDisabled if !@ordered && !msg.ordering_key.empty? # default is empty string

    batch = resolve_batch_for_message msg
    raise OrderingKeyError, batch.ordering_key if batch.canceled?
    batch_action = batch.add msg, callback
    if batch_action == :full
      publish_batches!
    elsif @published_at.nil?
      # Set initial time to now to start the background counter
      @published_at = Time.now
    end
    @cond.signal
  end

  nil
end

#resume_publish(ordering_key) ⇒ boolean

Resume publishing ordered messages for the provided ordering key.

Parameters:

  • ordering_key (String) —

    Identifies related messages for which publish order should be respected.

Returns:

  • (boolean) —

    true when resumed, false otherwise.



268
269
270
271
272
273
274
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 268

def resume_publish ordering_key
  synchronize do
    batch = resolve_batch_for_ordering_key ordering_key
    return if batch.nil?
    batch.resume!
  end
end

#started? ⇒ boolean

Whether the publisher has been started.

Returns:

  • (boolean) —

    true when started, false otherwise.



221
222
223
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 221

def started?
  !stopped?
end

#stop ⇒ AsyncPublisher

Begins the process of stopping the publisher. Messages already in the queue will be published, but no new messages can be added. Use #wait! to block until the publisher is fully stopped and all pending messages have been published.

Returns:



151
152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 151

def stop
  synchronize do
    break if @stopped

    @stopped = true
    publish_batches! stop: true
    @cond.signal
    @publish_thread_pool.shutdown
  end

  self
end

#stop!(timeout = nil) ⇒ AsyncPublisher

Stop this publisher and block until the publisher is fully stopped, all pending messages have been published, and all callbacks have completed, or until timeout seconds have passed.

The same as calling #stop and #wait!.

Parameters:

  • timeout (Number, nil) (defaults to: nil) —

    The number of seconds to block until the publisher is fully stopped. Default will block indefinitely.

Returns:



198
199
200
201
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 198

def stop! timeout = nil
  stop
  wait! timeout
end

#stopped? ⇒ boolean

Whether the publisher has been stopped.

Returns:

  • (boolean) —

    true when stopped, false otherwise.



229
230
231
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 229

def stopped?
  synchronize { @stopped }
end

#wait!(timeout = nil) ⇒ AsyncPublisher

Blocks until the publisher is fully stopped, all pending messages have been published, and all callbacks have completed, or until timeout seconds have passed.

Does not stop the publisher. To stop the publisher, first call #stop and then call #wait! to block until the publisher is stopped

Parameters:

  • timeout (Number, nil) (defaults to: nil) —

    The number of seconds to block until the publisher is fully stopped. Default will block indefinitely.

Returns:



176
177
178
179
180
181
182
183
184
185
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 176

def wait! timeout = nil
  synchronize do
    @publish_thread_pool.wait_for_termination timeout

    @callback_thread_pool.shutdown
    @callback_thread_pool.wait_for_termination timeout
  end

  self
end