peburrows/kane

Google Pub/Sub client for Elixir

Kane

Kane. Citizen Kane. Charles Foster Kane, to be exact, Publisher extraordinaire. Rosebud.

Kane is for publishing and subscribing to topics using Google Cloud Pub/Sub.

Installation

1. Add Kane to your list of dependencies in mix.exs:

def deps do
  [{:kane, "~> 0.9.0"}]
end

2. Configure Goth (Kane's underlying token storage and retrieval library) with your Google JSON credentials.

Usage

Pull, process and acknowledge messages via a pre-existing subscription:

{:ok, token} = Goth.fetch(MyApp.Goth)

kane = %Kane{
  project_id: my_app_gcp_credentials["project_id"],
  token: token
}

subscription = %Kane.Subscription{
                  name: "my-sub",
                  topic: %Kane.Topic{
                    name: "my-topic"
                  }
                }

{:ok, messages} = Kane.Subscription.pull(kane, subscription)

Enum.each messages, fn(mess)->
  process_message(mess)
end

# acknowledge message receipt in bulk
Kane.Subscription.ack(kane, subscription, messages)

Send message via pre-existing subscription:

topic   = %Kane.Topic{name: "my-topic"}
message = %Kane.Message{data: %{"hello": "world"}, attributes: %{"random" => "attr"}}

result  = Kane.Message.publish(kane, message, topic)

case result do
  {:ok, _return}    -> IO.puts("It worked!")
  {:error, _reason} -> IO.puts("Should we try again?")
end

Hints:

• Attributes have to be Strings (https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage)
• We base64-encode the message by default (only mandatory when using json - https://cloud.google.com/pubsub/publisher)

For more details, see the documentation.

kane's Issues

Skipping goth auth with PUBSUB_EMULATOR_HOST

Is it possible to skip authentication if $PUBSUB_EMULATOR_HOST is defined, and just use :project_id (and :endpoint via #15 (comment))? This would require changes to the kane and/or goth libraries.

Opened this ticket for goth with more info: peburrows/goth#56

List all topics using pagination

Currently there's a limit of about 100 topics when calling Kane.Topic.all function, as the pubsub API paginates the results. There are several ways of removing this 100 topic limit, but the easiest one is, by default, traversing the next pages.

When listing topics, we get a map of the following kind:

%{
  "topics" => list(%Kane.Topic{}),
  "nextPageToken" => str
}

We can GET projects/#{project}/topics?pageToken=#{page_token} to get the next page. If we use this recursively we can get all of the tokens until the map no longer has the nextPageToken present and we conclude that's the last page.

GenServer.call(Goth.Config, {:get, "project_id"}, 5000)

I'm trying to run the following task and I keep getting this exception

** (exit) exited in: GenServer.call(Goth.Config, {:get, "project_id"}, 5000)
    ** (EXIT) no process: the process is not alive or there's no process currently associated with the given name, possibly because its application isn't started
    (elixir) lib/gen_server.ex:821: GenServer.call/3
    lib/kane/message.ex:90: Kane.Message.path/1
    lib/kane/message.ex:30: Kane.Message.publish/2
    lib/kane/message.ex:22: Kane.Message.publish/2
    lib/mix/tasks/pubsub.ex:9: Mix.Tasks.Pubsub.run/1
    (mix) lib/mix/task.ex:314: Mix.Task.run_task/3
    (mix) lib/mix/cli.ex:80: Mix.CLI.run_task/2
    (elixir) lib/code.ex:677: Code.require_file/2

defmodule Mix.Tasks.Pubsub do
  use Mix.Task

  @shortdoc "Playground for gcloud pubsub"
  def run(_) do
    topic = %Kane.Topic{name: 'foobar'}
    message = %Kane.Message{data: %{"hello": "world"}, attributes: %{"random": "attr"}}

    result = Kane.Message.publish(message, topic)

    case result do
      {:ok, _return} -> IO.puts("It worked!")
      {:error, _reason} -> IO.puts("Failed to publish message")
    end
  end
end

The following is my mix.ex file

defmodule Broadcaster.Mixfile do
  use Mix.Project

  def project do
    [
      app: :broadcaster,
      version: "0.0.1",
      elixir: "~> 1.4",
      elixirc_paths: elixirc_paths(Mix.env),
      compilers: [:phoenix, :gettext] ++ Mix.compilers,
      start_permanent: Mix.env == :prod,
      aliases: aliases(),
      deps: deps()
    ]
  end

  # Configuration for the OTP application.
  #
  # Type `mix help compile.app` for more information.
  def application do
    [
      mod: {Broadcaster.Application, []},
      extra_applications: [:logger, :runtime_tools],
      applications: [:kane]
    ]
  end

  # Specifies which paths to compile per environment.
  defp elixirc_paths(:test), do: ["lib", "test/support"]
  defp elixirc_paths(_),     do: ["lib"]

  # Specifies your project dependencies.
  #
  # Type `mix help deps` for examples and options.
  defp deps do
    [
      {:phoenix, "~> 1.3.2"},
      {:phoenix_pubsub, "~> 1.0"},
      {:phoenix_ecto, "~> 3.2"},
      {:postgrex, ">= 0.0.0"},
      {:phoenix_html, "~> 2.10"},
      {:phoenix_live_reload, "~> 1.0", only: :dev},
      {:gettext, "~> 0.11"},
      {:cowboy, "~> 1.0"},
      {:kane, "~> 0.2.0"}
    ]
  end

  # Aliases are shortcuts or tasks specific to the current project.
  # For example, to create, migrate and run the seeds file at once:
  #
  #     $ mix ecto.setup
  #
  # See the documentation for `Mix` for more info on aliases.
  defp aliases do
    [
      "ecto.setup": ["ecto.create", "ecto.migrate", "run priv/repo/seeds.exs"],
      "ecto.reset": ["ecto.drop", "ecto.setup"],
      "test": ["ecto.create --quiet", "ecto.migrate", "test"]
    ]
  end
end

And finally my dev.exs

...
config :goth,
  json: "/Users/sandboxws/Documents/Unii-prod-daba6d92db8d.json" |> File.read!

Option to wait and not return immediately on pull

I notice that Kane.Subscription.pull has returnImmediately set to true.

Would you accept a PR which (optionally) sets this to false so that it will wait for new messages to enter the queue before returning?

I assume this would also need to increase the timeout in HTTPoison.post (possibly to :infinite?).

LICENSE file?

hex.pm claims MIT, but github doesn't know and I can't find a reference anywhere in the code to MIT.

Incorrect example in README

Using this example code in the README for subscribing we received a 404 response from GCP:

subscription = %Kane.Subscription{topic: %Kane.Topic{name: "my-topic"}

Changing it to:

subscription = %Kane.Subscription{name: "my-topic"}

seems to work better.

Switch to using protocol buffers instead of JSON

Assuming [de]serialization of protocol buffers is faster than [de]serialization of JSON, let's switch to using protocol buffers for interacting with the API.

Allow :system tuple for endpoint config

Related to #15, it would be great if we could use a {:system, "ENV_VAR"} approach for loading the endpoint value at runtime.

Would you be open to this enhancement? If so, I'd be happy to open a PR.

Add ability to stream messages

Instead of requiring the calling code to call Kane.Subscription.pull/1 repeatedly, we should add something like Kane.Subscription.stream

1.0.0 is not tagged

https://hex.pm/packages/kane refers to release 1.0.0, which isn't tagged. It's unclear what the correct commit is for the 1.0.0 release.

Switch to gRPC API

Once we have a decent http/2 client library & protocol buffer support added, we should switch to gRPC.

Usage locally against GCP PubSub emulator

I would like to be able to develop locally and have Kane use the GCP PubSub emulator (gcloud beta emulators pubsub start) endpoint.

This seems like something I could set up via the following config;

config :kane,
  endpoint: "http://127.0.0.1:8085"

...which seems to work OK, however I’m stuck with errors coming out of Goth.

I’m wondering if this restriction is intentional or just an oversight?