A GenStage producer that continuously fetches documents from Elasticsearch based on the configured strategy.

Strategies are meant to provide flexibility based on your use case. The current strategies are:

• OffBroadway.Elasticsearch.SearchAfterStrategy - Docs: Search After
• OffBroadway.Elasticsearch.ScrollStrategy- Docs: Scroll Search
• OffBroadway.Elasticsearch.SliceStrategy - Docs: Sliced Scroll

The available options are:

• host: (String.t()) - Required. The host where Elasticsearch can be found.
• index: (String.t()) - Required. The index to be ingested.
• search: (Map.t()) - Required. The search payload to be sent to Elasticsearch.
• strategy: (Atom.t()) - One of:
  • :search_after
  • :scroll
  • :slice
• strategy_module: (Atom.t()). A custom module that implemments OffBroadway.Elasticsearch.Strategy to be used if any of the built in strategies don't fit your use case.

Example:

defmodule MyBroadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {
          OffBroadway.Elasticsearch.Producer,
          [
            # The options listed above.
            host: "http://localhost:9200",
            index: "my-index",
            strategy: :slice,
            # If `strategy_module` is provided, it will take precedence over
            # `strategy`, that can be omitted when providing this option.
            strategy_module: MySearchStrategy
            search: search() # Extracted to a private function below.
          ]
        },
        transformer: {__MODULE__, :transform, []},
        concurrency: 10
      ],
      processors: [
        default: [concurrency: 5]
      ]
    )
  end

  defp search do
    %{
      query: %{
        match_all: %{}
      },
      sort: %{
        created_at: "asc",
        _id: "asc"
      }
    }
  end
end