# `BroadwayCloudPubSub.Pull.Producer`
[🔗](https://github.com/dashbitco/broadway_cloud_pub_sub/blob/v2.0.0-rc.0/lib/broadway_cloud_pub_sub/pull/producer.ex#L1)

A GenStage producer that continuously receives messages from a Google Cloud Pub/Sub
topic and acknowledges them after being successfully processed.

By default this producer uses `BroadwayCloudPubSub.Pull.FinchClient` to talk to Cloud
Pub/Sub, but you can provide your client by implementing the `BroadwayCloudPubSub.Pull.Client`
behaviour.

For a quick getting started on using Broadway with Cloud Pub/Sub, please see
the [Google Cloud Pub/Sub Guide](https://hexdocs.pm/broadway/google-cloud-pubsub.html).

## Options

Aside from `:receive_interval` and `:client` which are generic and apply to all
producers (regardless of the client implementation), all other options are specific to
`BroadwayCloudPubSub.PullClient`, which is the default client.

* `:client` - A module that implements the `BroadwayCloudPubSub.Pull.Client` behaviour.
  This module is responsible for fetching and acknowledging the messages.
  Pay attention that all options passed to the producer will be forwarded
  to the client. It's up to the client to normalize the options it needs.

  `BroadwayCloudPubSub.Pull.FinchClient` is the default client and will
  automatically retry the following errors [408, 500, 502, 503, 504, 522,
  524] up to 10 times with a 500ms pause between retries. This can be
  configured by passing the module with options to the client:

    {BroadwayCloudPubSub.Pull.FinchClient,
      retry_codes: [502, 503],
      retry_delay_ms: 300,
      max_retries: 5}

  These options will be merged with the options to the producer and passed
  to the client init/1 function.

  The default value is `BroadwayCloudPubSub.Pull.FinchClient`.

* `:subscription` - Required. The name of the subscription, including the project.
  For example, if you project is named `"my-project"` and your
  subscription is named `"my-subscription"`, then your subscription
  name is `"projects/my-project/subscriptions/my-subscription"`.

* `:max_number_of_messages` (`t:pos_integer/0`) - The maximum number of messages to be fetched per request. The default value is `10`.

* `:on_failure` - Configures the acking behaviour for failed messages.
  See the "Acknowledgements" section below for all the possible values.
  This option can also be changed for each message through
  `Broadway.Message.configure_ack/2`.

  Defaults to `{:nack, 0}`, which makes failed messages immediately
  available for redelivery. This matches the behaviour of the official
  Google Cloud Pub/Sub client libraries.

  The default value is `{:nack, 0}`.

* `:on_success` - Configures the acking behaviour for successful messages.
  See the "Acknowledgements" section below for all the possible values.
  This option can also be changed for each message through
  `Broadway.Message.configure_ack/2`. The default value is `:ack`.

* `:receive_interval` (`t:integer/0`) - The duration (in milliseconds) for which the producer waits
  before making a request for more messages. The default value is `5000`.

* `:receive_timeout` - The maximum time (in milliseconds) to wait for a response
  before the pull client returns an error. The default value is `:infinity`.

* `:goth` (`t:atom/0`) - The `Goth` module to use for authentication. Note that this option only applies to the
  default token generator.

* `:token_generator` - An MFArgs tuple that will be called before each request
  to fetch an authentication token. It should return `{:ok, String.t()} | {:error, any()}`.
  By default this will invoke `Goth.fetch/1` with the `:goth` option.
  See the "Custom token generator" section below for more information.

* `:base_url` - The base URL for the Cloud Pub/Sub service.
  This option is mostly useful for testing via the Pub/Sub emulator. The default value is `"https://pubsub.googleapis.com"`.

* `:finch` (`t:atom/0`) - The name of the `Finch` pool. If no name is provided, then a default
  pool will be started by the pipeline's supervisor. The default value is `nil`.

### Custom token generator

A custom token generator can be given as a MFArgs tuple.

For example, define a `MyApp.fetch_token/0` function:

    defmodule MyApp do

      @doc "Fetches a Google auth token"
      def fetch_token do
        with {:ok, token} <- Goth.fetch(MyApp.Goth)
          {:ok, token.token}
        end
      end
    end

and configure the producer to use it:

    token_generator: {MyApp, :fetch_token, []}

## Acknowledgements

You can use the `:on_success` and `:on_failure` options to control how
messages are acknowledged with the Pub/Sub system.

By default successful messages are acknowledged and failed messages are ignored
(`:noop`). You can set `:on_success` and `:on_failure` when starting this producer,
or change them for each message through `Broadway.Message.configure_ack/2`.

The following values are supported by both `:on_success` and `:on_failure`:

* `:ack` - Acknowledge the message. Pub/Sub can remove the message from
   the subscription.

* `:noop` - Do nothing. No requests will be made to Pub/Sub, and the
   message will be rescheduled according to the subscription-level
   `ackDeadlineSeconds`.

* `:nack` - Make a request to Pub/Sub to set `ackDeadlineSeconds` to `0`,
   which may cause the message to be immediately redelivered to another
   connected consumer. Note that this does not modify the subscription-level
   `ackDeadlineSeconds` used for subsequent messages.

* `{:nack, integer}` - Modifies the `ackDeadlineSeconds` for a particular
   message. Note that this does not modify the subscription-level
   `ackDeadlineSeconds` used for subsequent messages.

### Batching

Even if you are not interested in working with Broadway batches via the
`handle_batch/3` callback, we recommend all Broadway pipelines with Pub/Sub
producers to define a default batcher with `batch_size` set to 10, so
messages can be acknowledged in batches, which improves the performance
and reduces the cost of integrating with Google Cloud Pub/Sub. In addition,
you should ensure that `batch_timeout` is set to a value less than
the acknowledgement deadline on the subscription. Otherwise you could
potentially have messages that remain in the subscription and are never
acknowledged successfully.

### Example

    Broadway.start_link(MyBroadway,
      name: MyBroadway,
      producer: [
        module: {BroadwayCloudPubSub.Pull.Producer,
          goth: MyApp.Goth,
          subscription: "projects/my-project/subscriptions/my_subscription"
        }
      ],
      processors: [
        default: []
      ],
      batchers: [
        default: [
          batch_size: 10,
          batch_timeout: 2_000
        ]
      ]
    )

The above configuration will set up a producer that continuously receives
messages from `"projects/my-project/subscriptions/my_subscription"` and sends
them downstream.

## Telemetry

This producer emits a few [Telemetry](https://github.com/beam-telemetry/telemetry)
events which are listed below.

  * `[:broadway_cloud_pub_sub, :pull_client, :receive_messages, :start | :stop | :exception]` spans -
    these events are emitted in "span style" when executing pull requests to GCP PubSub.
    See `:telemetry.span/3`.

    All these events have the measurements described in `:telemetry.span/3`. The events
    contain the following metadata:

    * `:max_messages` - the number of messages requested after applying the `max_messages`
    config option to the existing demand
    * `:demand` - the total demand accumulated into the producer
    * `:name` - the name of the Broadway topology

  * `[:broadway_cloud_pub_sub, :pull_client, :ack, :start | :stop | :exception]` span - these events
    are emitted in "span style" when acking messages on GCP PubSub. See `:telemetry.span/3`.

    All these events have the measurements described in `:telemetry.span/3`. The events
    contain the following metadata:

    * `:name` - the name of the Broadway topology

---

*Consult [api-reference.md](api-reference.md) for complete listing*