BroadwayCloudPubSub

A Google Cloud Pub/Sub connector for Broadway.

Documentation can be found at https://hexdocs.pm/broadway_cloud_pub_sub.

What's in the box

• BroadwayCloudPubSub.Producer: Broadway producer using the gRPC StreamingPull API. Messages are pushed by the server over a persistent bidirectional stream, giving low latency and high throughput with automatic lease extension and server-side flow control. This is the recommended producer, in line with Google's own guidance that StreamingPull is what their first-party client libraries use "where possible".
• BroadwayCloudPubSub.Pull.Producer: Broadway producer using the unary HTTP Pull API. Retained for environments where gRPC is unavailable or undesired, and for the cases Google lists as Pull-only: when you need strict control over the number of messages pulled per request, tight control over client memory and CPU, or when your subscriber acts as a proxy to another pull-oriented system.
• BroadwayCloudPubSub.Streaming.Client: Behaviour for custom gRPC client implementations.
• BroadwayCloudPubSub.Pull.Client: Behaviour for custom HTTP pull client implementations.

Installation

Add :broadway_cloud_pub_sub to your dependencies, along with an HTTP/2 adapter for :grpc:

def deps do
  [
    {:broadway_cloud_pub_sub, "~> 2.0"},
    {:goth, "~> 1.3"},
    {:grpc, "~> 1.0"},
    {:protobuf, "~> 0.12"},
    # Pick one HTTP/2 adapter:
    {:gun, "~> 2.2"},
    # or
    # {:mint, "~> 1.9"},
    # {:castore, "~> 1.0"}
  ]
end

The goth package handles Google Authentication and is required for the default token generator.

The grpc and protobuf packages are required by BroadwayCloudPubSub.Producer. You must pick one HTTP/2 adapter for the gRPC connection and add it to your mix.exs: either :gun, or :mint together with :castore.

If you only use BroadwayCloudPubSub.Pull.Producer you may omit :grpc, :protobuf, and the adapter packages.

Usage

Broadway.start_link(MyBroadway,
  name: MyBroadway,
  producer: [
    module: {BroadwayCloudPubSub.Producer,
      goth: MyGoth,
      subscription: "projects/my-project/subscriptions/my-subscription",
      max_outstanding_messages: 1000
    }
  ],
  processors: [default: [concurrency: 10]]
)

See BroadwayCloudPubSub.Producer for the full option reference, including flow control, reconnection backoff, graceful shutdown, and telemetry.

HTTP/2 adapter

The producer supports two adapters. Both are optional dependencies of :grpc, so you select one by adding it to your application's mix.exs (see Installation).

• :gun (default): Gun HTTP/2 client. Add {:gun, "~> 2.2"} to your deps.
• :mint: Mint HTTP/2 client. Add {:mint, "~> 1.9"} and {:castore, "~> 1.0"} to your deps.

Then select the adapter in your producer config:

{BroadwayCloudPubSub.Producer,
 goth: MyGoth,
 subscription: "projects/my-project/subscriptions/my-subscription",
 adapter: :mint}

Using the HTTP pull producer

If gRPC is not available in your environment or you prefer to use the HTTP pull method, use BroadwayCloudPubSub.Pull.Producer:

Broadway.start_link(MyBroadway,
  name: MyBroadway,
  producer: [
    module: {BroadwayCloudPubSub.Pull.Producer,
      goth: MyGoth,
      subscription: "projects/my-project/subscriptions/my-subscription"
    }
  ],
  processors: [default: [concurrency: 10]]
)

Upgrading from 1.x

2.0 is a major release with breaking changes. The three-line summary is below; the full upgrade guide has step-by-step instructions, option mapping tables, and rationale for every change.

Breaking change 1: BroadwayCloudPubSub.Producer is now the gRPC streaming producer

The biggest change: the module name BroadwayCloudPubSub.Producer now refers to the new gRPC StreamingPull producer. The 1.x HTTP pull producer lives on under BroadwayCloudPubSub.Pull.Producer.

# 1.x — HTTP pull producer
{BroadwayCloudPubSub.Producer, goth: MyApp.Goth, subscription: "..."}

# 2.0 option A — switch to streaming (recommended, lower latency)
{BroadwayCloudPubSub.Producer,
 goth: MyApp.Goth,
 subscription: "...",
 max_outstanding_messages: 1000}

# 2.0 option B — keep HTTP pull, one-line change
{BroadwayCloudPubSub.Pull.Producer, goth: MyApp.Goth, subscription: "..."}

The streaming producer requires :grpc, :protobuf, and an HTTP/2 adapter (:gun or :mint + :castore). If you stay on the pull producer those packages are not needed.

Breaking change 2: two modules renamed (only if referenced directly)

These only matter if you passed the module explicitly (e.g. client: BroadwayCloudPubSub.PullClient) or implemented a custom pull client with @behaviour BroadwayCloudPubSub.Client.

Breaking change 3: on_failure default changed from :noop to {:nack, 0}

Failed messages are now immediately made available for redelivery instead of waiting for the subscription's ackDeadlineSeconds to expire. This matches the behaviour of Google's own first-party client libraries.

To keep the 1.x behaviour, set the option explicitly:

{BroadwayCloudPubSub.Pull.Producer,
 goth: MyApp.Goth,
 subscription: "...",
 on_failure: :noop}

See the full upgrade guide for all details.

License

Copyright 2019 Michael Crumm
Copyright 2020 Dashbit

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.