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. In the form of "/projects/project-identifier/topics/topic-name".

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.



237
238
239
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 237

def enable_message_ordering!
  synchronize { @ordered = true }
end

#flush ⇒ AsyncPublisher

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

Returns:



204
205
206
207
208
209
210
211
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 204

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) 


252
253
254
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 252

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:



118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 118

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.



264
265
266
267
268
269
270
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 264

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.



217
218
219
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 217

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:



147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 147

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:



194
195
196
197
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 194

def stop! timeout = nil
  stop
  wait! timeout
end

#stopped? ⇒ boolean

Whether the publisher has been stopped.

Returns:

  • (boolean) —

    true when stopped, false otherwise.



225
226
227
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 225

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:



172
173
174
175
176
177
178
179
180
181
 
# File 'lib/google/cloud/pubsub/async_publisher.rb', line 172

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