From 420d7e63dc674ecb13a96c98509e510edbc68337 Mon Sep 17 00:00:00 2001 From: v0idpwn Date: Thu, 6 Apr 2023 22:57:21 +0200 Subject: [PATCH] Documentation improvements --- README.md | 56 +------------------------------ guides/getting-started.md | 57 +++++++++++++++++++++++++++++++ lib/kafkaesque.ex | 61 ++++++++++++++++++++++------------ lib/kafkaesque/kafka_client.ex | 4 +++ lib/kafkaesque/query.ex | 51 +++++++++++++++------------- lib/kafkaesque/rescuer.ex | 3 +- mix.exs | 20 +++++++++-- mix.lock | 11 +++--- 8 files changed, 154 insertions(+), 109 deletions(-) create mode 100644 guides/getting-started.md diff --git a/README.md b/README.md index 6e3a454..17f0939 100644 --- a/README.md +++ b/README.md @@ -26,58 +26,4 @@ end ## Getting started -First, you need to migrate your database to create the outbox. Create a -migration with Ecto and add the following: - -```elixir -defmodule MyApp.Repo.Migrations.AddKafkaesque - def up do - Kafkaesque.Migrations.up() - end - - def down do - Kafkaesque.Migrations.down() - end -end -``` - -Then, you must define a publisher module, associated with the Repo you -ran the Kafkaesque migrations on. For example: - -```elixir -defmodule MyApp.KafkaPublisher do - use Kafkaesque, repo: MyApp.Repo - - # Optional, sets the partition for the message. Should return an integer - # Defaults to returning 0. - def partition(_topic, _body) do - Enum.random(0..1) - end - - # Optional, encodes the body of the message, should return a String. - # Defaults to the identity function. - def encode(body) do - Jason.encode!(body) - end -end -``` - -Finally, you should start a `Kafkaesque` instance in your supervision tree. -For example: - -```elixir -defmodule MyApp.Application do - def start(_type, _args) do - children = [ - MyApp.Repo, - {Kafkaesque, [repo: MyApp.Repo, client_opts: [client_id: :my_client, brokers: [{"localhost, 9092"}]]]}, - ] - end -end -``` - -Done! Now you can publish messages. Try it out with: - -```elixir -MyApp.KafkaPublisher.publish("topic", %{hello: :kafka})` -``` +Check the [Getting started](https://hexdocs.pm/kafkaesque/getting-started.html) guide in Hexdocs. diff --git a/guides/getting-started.md b/guides/getting-started.md new file mode 100644 index 0000000..41ea893 --- /dev/null +++ b/guides/getting-started.md @@ -0,0 +1,57 @@ +# Getting started + +First, you need to migrate your database to create the outbox. Create a +migration with Ecto and add the following: + +```elixir +defmodule MyApp.Repo.Migrations.AddKafkaesque + def up do + Kafkaesque.Migrations.up() + end + + def down do + Kafkaesque.Migrations.down() + end +end +``` + +Then, you must define an outbox module, associated with the Repo you +ran the Kafkaesque migrations on. For example: + +```elixir +defmodule MyApp.Outbox do + use Kafkaesque, repo: MyApp.Repo + + # Optional, sets the partition for the message. Should return an integer + # Defaults to returning 0. + def partition(_topic, _body) do + Enum.random(0..1) + end + + # Optional, encodes the body of the message, should return a String. + # Defaults to the identity function. + def encode(body) do + Jason.encode!(body) + end +end +``` + +Finally, you should start a `Kafkaesque` instance in your supervision tree. +For example: + +```elixir +defmodule MyApp.Application do + def start(_type, _args) do + children = [ + MyApp.Repo, + {MyApp.Outbox, [client_opts: [client_id: :my_client, brokers: [{"localhost, 9092"}]]]}, + ] + end +end +``` + +Done! Now you can publish messages. Try it out with: + +```elixir +MyApp.Outbox.publish("topic", %{hello: :kafka})` +``` diff --git a/lib/kafkaesque.ex b/lib/kafkaesque.ex index ce24f28..13f5bfc 100644 --- a/lib/kafkaesque.ex +++ b/lib/kafkaesque.ex @@ -1,33 +1,31 @@ defmodule Kafkaesque do @moduledoc """ - This module provides a message publishing API through `Kafkaesque.publish/3`. + This module provides the main APIs for Kafkaesque - It also allows you to `use` it and have a streamlined publishing function, - possibly providing encodin function: + ### Introduction + Kafkaesque is an outbox library built for PostgreSQL and designed, primarily, + for Kafka, though usage with other software is possible and encouraged. - defmodule MyApp.Kafka do - use Kafkaesque, repo: MyApp.Repo + - Transactional safety for messages: if they were created, they **will** be + eventually published. They're only created if the transaction commits. + - Ordering: messages are published sequentially for topic/partition combinations + - Shutdown safety: has graceful shutdown and rescue for cases where it doesn't + happen. + - Observability: all operations publish telemetry events. + - Garbage collection: outbox table is periodically cleaned. + - Multi-node safe: safety powered by PostgreSQL. - # Optional, defaults to `body` - def encode(body) do - Jason.encode!(body) - end + For a comprehensive installation guide, check the [Getting started]("getting-started.md") + guide. - def partition(_topic, %MyMessage{id: id}) do - id - end - end - - MyApp.Kafka.publish("my_topic", %{hello: :kafka}) - - `encode/1` should return a string, and defaults to the identity function. + ![Basic diagram](http://www.plantuml.com/plantuml/proxy?cache=no&src=https://raw.githubusercontent.com/v0idpwn/kafkaesque/master/diagrams/basic.iuml) - `partition/2` should return an integer, and defaults to 0. It will be used - as the partition for the message. + #### Note - See the documentation for `Kafkaesque.start_link/1` to learn about starting - Kafkaesque in your application. + For most cases, this module shouldn't be called directly. Instead, you want to + `use` it to define an outbox for your application, and call the outbox, as you + do with Ecto repos. """ alias Kafkaesque.Message @@ -60,7 +58,9 @@ defmodule Kafkaesque do end @doc """ - Starts a Kafkaesque instance. Accepts the following opts: + Starts a Kafkaesque instance. + + Accepts the following opts: - `:repo`: the repo where messages will be read from. Usually should be the same repo that you're writing to. @@ -89,11 +89,16 @@ defmodule Kafkaesque do the table. Defaults to 72 hours. - `query_opts`: Options to pass to Ecto queries. Defaults to [log: false] """ + @spec start_link(Keyword.t()) :: {:ok, pid()} def start_link(opts) do opts = Keyword.validate!(opts, [:repo] ++ @default_opts) Kafkaesque.Supervisor.start_link(opts) end + @doc """ + Child spec for a Kafkaesque instance + """ + @spec child_spec(Keyword.t()) :: Supervisor.child_spec() def child_spec(opts) do opts = Keyword.validate!(opts, [:repo] ++ @default_opts) Kafkaesque.Supervisor.child_spec(opts) @@ -120,6 +125,18 @@ defmodule Kafkaesque do 0 end + @spec start_link(Keyword.t()) :: {:ok, pid} + def start_link(opts) do + opts = Keyword.merge(unquote(opts), opts) + Kafkaesque.start_link(opts) + end + + @spec child_spec(Keyword.t()) :: Supervisor.child_spec() + def child_spec(opts) do + opts = Keyword.merge(unquote(opts), opts) + Kafkaesque.child_spec(opts) + end + defoverridable encode: 1, partition: 2 end end diff --git a/lib/kafkaesque/kafka_client.ex b/lib/kafkaesque/kafka_client.ex index cf2c83a..2eca6ed 100644 --- a/lib/kafkaesque/kafka_client.ex +++ b/lib/kafkaesque/kafka_client.ex @@ -1,4 +1,8 @@ defmodule Kafkaesque.Client do + @moduledoc """ + Behaviour for clients + """ + alias Kafkaesque.Message @type client() :: term() diff --git a/lib/kafkaesque/query.ex b/lib/kafkaesque/query.ex index 07007e2..04c2641 100644 --- a/lib/kafkaesque/query.ex +++ b/lib/kafkaesque/query.ex @@ -34,30 +34,33 @@ defmodule Kafkaesque.Query do |> order_by([m], m.id) |> limit(^demand) - repo.transaction(fn -> - repo.query!("SELECT pg_advisory_xact_lock($1)", [@xact_lock_key], query_opts) - - Message - |> where([m], m.id in subquery(subset)) - |> select([m, _], m) - |> update([m], - set: [ - state: :publishing, - attempted_at: fragment("CURRENT_TIMESTAMP"), - attempted_by: ^inspect(node()) - ], - inc: [attempt: 1] - ) - |> repo.update_all([], query_opts) - |> case do - {0, nil} -> - {0, []} - - {count, messages} -> - sorted = Enum.sort(messages, fn m1, m2 -> m1.id <= m2.id end) - {count, sorted} - end - end, query_opts) + repo.transaction( + fn -> + repo.query!("SELECT pg_advisory_xact_lock($1)", [@xact_lock_key], query_opts) + + Message + |> where([m], m.id in subquery(subset)) + |> select([m, _], m) + |> update([m], + set: [ + state: :publishing, + attempted_at: fragment("CURRENT_TIMESTAMP"), + attempted_by: ^inspect(node()) + ], + inc: [attempt: 1] + ) + |> repo.update_all([], query_opts) + |> case do + {0, nil} -> + {0, []} + + {count, messages} -> + sorted = Enum.sort(messages, fn m1, m2 -> m1.id <= m2.id end) + {count, sorted} + end + end, + query_opts + ) end @spec update_success_batch(Ecto.Repo.t(), [pos_integer()], Keyword.t()) :: :ok diff --git a/lib/kafkaesque/rescuer.ex b/lib/kafkaesque/rescuer.ex index 0c53a8f..41aae25 100644 --- a/lib/kafkaesque/rescuer.ex +++ b/lib/kafkaesque/rescuer.ex @@ -32,7 +32,8 @@ defmodule Kafkaesque.Rescuer do interval_ms = Keyword.fetch!(opts, :rescuer_interval_ms) limit_ms = Keyword.fetch!(opts, :rescuer_limit_ms) - {:ok, %{repo: repo, query_opts: query_opts, interval_ms: interval_ms, limit_ms: limit_ms}, {:continue, :rescue}} + {:ok, %{repo: repo, query_opts: query_opts, interval_ms: interval_ms, limit_ms: limit_ms}, + {:continue, :rescue}} end @impl GenServer diff --git a/mix.exs b/mix.exs index 75c6ffc..bf54dac 100644 --- a/mix.exs +++ b/mix.exs @@ -1,13 +1,13 @@ defmodule Kafkaesque.MixProject do use Mix.Project - @version "0.0.1" + @version "1.0.0-rc.0" def project do [ app: :kafkaesque, version: @version, - elixir: "~> 1.11", + elixir: "~> 1.13", start_permanent: Mix.env() == :prod, aliases: aliases(), elixirc_paths: elixirc_paths(Mix.env()), @@ -15,6 +15,7 @@ defmodule Kafkaesque.MixProject do description: description(), package: package(), name: "Kafkaesque", + docs: docs(), source_url: "https://github.com/v0idpwn/kafkaesque" ] end @@ -39,6 +40,21 @@ defmodule Kafkaesque.MixProject do ] end + defp docs do + [ + main: "Kafkaesque", + source_ref: "#v{@version}", + extra_section: "GUIDES", + extras: extras() + ] + end + + defp extras do + [ + "guides/getting-started.md" + ] + end + defp description do "Transactional outbox for kafka" end diff --git a/mix.lock b/mix.lock index 77dfeec..84064ec 100644 --- a/mix.lock +++ b/mix.lock @@ -5,17 +5,18 @@ "db_connection": {:hex, :db_connection, "2.4.3", "3b9aac9f27347ec65b271847e6baeb4443d8474289bd18c1d6f4de655b70c94d", [:mix], [{:connection, "~> 1.0", [hex: :connection, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "c127c15b0fa6cfb32eed07465e05da6c815b032508d4ed7c116122871df73c12"}, "decimal": {:hex, :decimal, "2.0.0", "a78296e617b0f5dd4c6caf57c714431347912ffb1d0842e998e9792b5642d697", [:mix], [], "hexpm", "34666e9c55dea81013e77d9d87370fe6cb6291d1ef32f46a1600230b1d44f577"}, "dialyxir": {:hex, :dialyxir, "1.2.0", "58344b3e87c2e7095304c81a9ae65cb68b613e28340690dfe1a5597fd08dec37", [:mix], [{:erlex, ">= 0.2.6", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "61072136427a851674cab81762be4dbeae7679f85b1272b6d25c3a839aff8463"}, - "earmark_parser": {:hex, :earmark_parser, "1.4.12", "b245e875ec0a311a342320da0551da407d9d2b65d98f7a9597ae078615af3449", [:mix], [], "hexpm", "711e2cc4d64abb7d566d43f54b78f7dc129308a63bc103fbd88550d2174b3160"}, + "earmark_parser": {:hex, :earmark_parser, "1.4.31", "a93921cdc6b9b869f519213d5bc79d9e218ba768d7270d46fdcf1c01bacff9e2", [:mix], [], "hexpm", "317d367ee0335ef037a87e46c91a2269fef6306413f731e8ec11fc45a7efd059"}, "ecto": {:hex, :ecto, "3.9.5", "9f0aa7ae44a1577b651c98791c6988cd1b69b21bc724e3fd67090b97f7604263", [:mix], [{:decimal, "~> 1.6 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "d4f3115d8cbacdc0bfa4b742865459fb1371d0715515842a1fb17fe31920b74c"}, "ecto_sql": {:hex, :ecto_sql, "3.9.2", "34227501abe92dba10d9c3495ab6770e75e79b836d114c41108a4bf2ce200ad5", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.9.2", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.6.0", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.16.0 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "1eb5eeb4358fdbcd42eac11c1fbd87e3affd7904e639d77903c1358b2abd3f70"}, "erlex": {:hex, :erlex, "0.2.6", "c7987d15e899c7a2f34f5420d2a2ea0d659682c06ac607572df55a43753aa12e", [:mix], [], "hexpm", "2ed2e25711feb44d52b17d2780eabf998452f6efda104877a3881c2f8c0c0c75"}, - "ex_doc": {:hex, :ex_doc, "0.23.0", "a069bc9b0bf8efe323ecde8c0d62afc13d308b1fa3d228b65bca5cf8703a529d", [:mix], [{:earmark_parser, "~> 1.4.0", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}], "hexpm", "f5e2c4702468b2fd11b10d39416ddadd2fcdd173ba2a0285ebd92c39827a5a16"}, + "ex_doc": {:hex, :ex_doc, "0.29.4", "6257ecbb20c7396b1fe5accd55b7b0d23f44b6aa18017b415cb4c2b91d997729", [:mix], [{:earmark_parser, "~> 1.4.31", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "2c6699a737ae46cb61e4ed012af931b57b699643b24dabe2400a8168414bc4f5"}, "gen_stage": {:hex, :gen_stage, "1.2.1", "19d8b5e9a5996d813b8245338a28246307fd8b9c99d1237de199d21efc4c76a1", [:mix], [], "hexpm", "83e8be657fa05b992ffa6ac1e3af6d57aa50aace8f691fcf696ff02f8335b001"}, "jason": {:hex, :jason, "1.4.0", "e855647bc964a44e2f67df589ccf49105ae039d4179db7f6271dfd3843dc27e6", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "79a3791085b2a0f743ca04cec0f7be26443738779d09302e01318f97bdb82121"}, "kafka_protocol": {:hex, :kafka_protocol, "4.1.0", "53fac8866969484f783bff204bd4e41e62a97ce9753c83f802a08d5bfc0e0c4c", [:rebar3], [{:crc32cer, "0.1.8", [hex: :crc32cer, repo: "hexpm", optional: false]}], "hexpm", "61cb8b80199bf95122cf8073e0f4c0ad62f82515b4d44c54f946a5972c3f5fa5"}, - "makeup": {:hex, :makeup, "1.0.5", "d5a830bc42c9800ce07dd97fa94669dfb93d3bf5fcf6ea7a0c67b2e0e4a7f26c", [:mix], [{:nimble_parsec, "~> 0.5 or ~> 1.0", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cfa158c02d3f5c0c665d0af11512fed3fba0144cf1aadee0f2ce17747fba2ca9"}, - "makeup_elixir": {:hex, :makeup_elixir, "0.15.1", "b5888c880d17d1cc3e598f05cdb5b5a91b7b17ac4eaf5f297cb697663a1094dd", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.1", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "db68c173234b07ab2a07f645a5acdc117b9f99d69ebf521821d89690ae6c6ec8"}, - "nimble_parsec": {:hex, :nimble_parsec, "1.1.0", "3a6fca1550363552e54c216debb6a9e95bd8d32348938e13de5eda962c0d7f89", [:mix], [], "hexpm", "08eb32d66b706e913ff748f11694b17981c0b04a33ef470e33e11b3d3ac8f54b"}, + "makeup": {:hex, :makeup, "1.1.0", "6b67c8bc2882a6b6a445859952a602afc1a41c2e08379ca057c0f525366fc3ca", [:mix], [{:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "0a45ed501f4a8897f580eabf99a2e5234ea3e75a4373c8a52824f6e873be57a6"}, + "makeup_elixir": {:hex, :makeup_elixir, "0.16.1", "cc9e3ca312f1cfeccc572b37a09980287e243648108384b97ff2b76e505c3555", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "e127a341ad1b209bd80f7bd1620a15693a9908ed780c3b763bccf7d200c767c6"}, + "makeup_erlang": {:hex, :makeup_erlang, "0.1.1", "3fcb7f09eb9d98dc4d208f49cc955a34218fc41ff6b84df7c75b3e6e533cc65f", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "174d0809e98a4ef0b3309256cbf97101c6ec01c4ab0b23e926a9e17df2077cbb"}, + "nimble_parsec": {:hex, :nimble_parsec, "1.3.0", "9e18a119d9efc3370a3ef2a937bf0b24c088d9c4bf0ba9d7c3751d49d347d035", [:mix], [], "hexpm", "7977f183127a7cbe9346981e2f480dc04c55ffddaef746bd58debd566070eef8"}, "postgrex": {:hex, :postgrex, "0.16.5", "fcc4035cc90e23933c5d69a9cd686e329469446ef7abba2cf70f08e2c4b69810", [:mix], [{:connection, "~> 1.1", [hex: :connection, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "edead639dc6e882618c01d8fc891214c481ab9a3788dfe38dd5e37fd1d5fb2e8"}, "snappyer": {:hex, :snappyer, "1.2.8", "201ce9067a33c71a6a5087c0c3a49a010b17112d461e6df696c722dcb6d0934a", [:rebar3], [], "hexpm", "35518e79a28548b56d8fd6aee2f565f12f51c2d3d053f9cfa817c83be88c4f3d"}, "supervisor3": {:hex, :supervisor3, "1.1.11", "d81cdec31d102fde407423e1d05b569572850deebed86b951d5233c387cba80b", [:rebar3], [], "hexpm", "e6c2dedbcabcba24995a218aca12db5e208b80d3252692b22ef0f1a266104b50"},