# Changelog

## [0.3.0](https://github.com/intility/off_broadway_emqtt/compare/v0.2.1...v0.3.0) (2026-04-20)

_Breaking changes and reliability rework._

This release removes the ETS/disk-log buffer layer and replaces it with proper
MQTT protocol-level backpressure. The result is a simpler, more reliable producer
that correctly participates in QoS 1/2 guarantees.

### Breaking changes

- **ETS buffer removed.** The options `buffer_size`, `buffer_overflow_strategy`,
  `buffer_durability`, and `buffer_log_dir` no longer exist. Remove them from
  your configuration. Backpressure is now handled entirely by `max_inflight` and
  the MQTT protocol.
- **`clean_start` defaults to `false`** (was `true`). With the old default, the
  broker discarded the session on every reconnect, silently losing all unACKed
  QoS 1/2 messages. The new default preserves the session so the broker redelivers
  unACKed messages after a producer restart. If you explicitly want a fresh session
  on each connect, set `config: [clean_start: true]`.
- **`topics` is now required.** Previously it defaulted to `[]`, starting a
  producer that subscribed to nothing. Omitting `topics` now raises at startup.
- **`MessageHandler` behaviour simplified.** The callbacks `handle_connect/1`,
  `handle_disconnect/1`, and `handle_pubrel/1` have been removed. Only
  `handle_message/3` remains. Custom handlers implementing the removed callbacks
  must delete them.
- **`concurrency > 1` requires `shared_group`.** Starting a pipeline with multiple
  producer instances and no `shared_group` now raises at startup. Without shared
  subscriptions every producer receives every message, causing duplicates.
- **Client ID suffix `_N` is now always appended.** Each producer instance connects
  with `{clientid}_0`, `{clientid}_1`, etc. A pipeline that previously connected as
  `my-client` now connects as `my-client_0`. If you have a persistent broker session
  keyed by the exact client ID, update the `clientid` in your config to match the new
  suffix (e.g. `clientid: "my-client_0"`) or accept that the session will be treated
  as new on first connect.
- **emqtt bumped to `~> 1.14`**, cowlib to `~> 2.13.0`.

### Bug fixes

- **QoS 2 acknowledgement fixed.** The `pubcomp` step was incorrectly calling
  `:emqtt.pubrec` instead of `:emqtt.pubcomp`, breaking the QoS 2 handshake
  entirely. QoS 2 exactly-once delivery now works correctly.

### New features

- **Protocol-level backpressure via `max_inflight`.** The broker stops delivering
  new messages once `max_inflight` unACKed QoS 1/2 messages are outstanding.
  Default is 100. For MQTT v5, `Receive-Maximum` is automatically set in the
  CONNECT properties so the broker enforces the limit server-side.
- **`shared_group` option** for distributing messages across multiple producer
  instances using MQTT shared subscriptions (`$share/{group}/{topic}`).
- **New `config` options:** `reconnect`, `reconnect_timeout`, `low_mem`.
- **`connection: :down` telemetry event** emitted when the MQTT connection is lost,
  with `%{client_id: string, producer_index: integer, reason: term}` metadata.
- **Clearer startup errors.** A `Logger.error` message including host and port is
  emitted when the producer fails to connect to the broker.
- **`persistent_term` cleanup.** The ack-options entry written at pipeline startup
  is now erased when the pipeline stops, preventing accumulation in long-running
  applications that start and stop pipelines dynamically.

### Migration from v0.2.x

#### Remove buffer options

The ETS/disk-log buffer has been removed. Delete these options from your producer config:

```elixir
# Remove all of these:
buffer_size: 10_000,
buffer_overflow_strategy: :drop_head,
buffer_durability: :durable,
buffer_log_dir: System.tmp_dir!(),
```

Backpressure is now handled by `max_inflight`.

#### Review clean_start

`clean_start` now defaults to `false` (was `true`). This is the safer default: the broker
redelivers unACKed QoS 1/2 messages after a restart instead of discarding them.

If your pipeline was relying on the broker discarding the session on reconnect, add
`config: [clean_start: true]` explicitly. Be aware this means unACKed messages are lost
on every restart.

#### Remove MessageHandler callbacks

If you implemented a custom `MessageHandler`, delete any `handle_connect/1`, `handle_disconnect/1`,
or `handle_pubrel/1` callbacks. Only `handle_message/3` is part of the behaviour.

#### Add shared_group for concurrency > 1

If you were running with `concurrency > 1`, you must now add `shared_group`:

```elixir
# Before (would cause duplicate messages):
producer: [
  module: {OffBroadway.EMQTT.Producer, topics: [{"my/topic", 1}], config: [...]},
  concurrency: 3
]

# After:
producer: [
  module: {OffBroadway.EMQTT.Producer,
    topics: [{"my/topic", 1}],
    shared_group: "my-pipeline",
    config: [...]
  },
  concurrency: 3
]
```

#### Update dependencies

```elixir
{:off_broadway_emqtt, "~> 0.3.0"}
```

---

## [0.2.1](https://github.com/intility/off_broadway_emqtt/compare/v0.2.0...v0.2.1) (2025-06-05)

- If `:clean_start` option is `true`, truncate the buffer log file and skip replay when the producer starts.
- Properly disconnect from the MQTT broker on terminate.

## [0.2.0](https://github.com/intility/off_broadway_emqtt/compare/v0.1.0...v0.2.0) (2025-06-03)

- Add support for wrapping the ETS buffer cache with [:disk_log](https://www.erlang.org/docs/17/man/disk_log) to persist cached messages for producer.
  - Introduced new option `buffer_durability` which can be either `:durable` or `:transient`. When `:durable`, 
    messages will be persisted to disk to ensure messages are not lost if the producer crashes. Defaults to
    `:transient` (in-memory buffer only).
  - New option `buffer_log_dir` can be either a string, or a zero-arity function that returns the directory to
    store buffer logs.
  - Added new telemetry events for `:durable` buffer operations.

## 0.1.1 (unreleased)

_Never tagged. Changes rolled into later releases._

- Emitting `off_broadway_emqtt.receive_message.ack` reads message topic from message receipt instead of from the message body.
 This ensures that topic is included in telemetry events even if the message has been altered during dispatch.
- Move `emqtt.start_link/1` and `emqtt.connect/1` to a `handle_continue/2` callback to prevent blocking `GenServer.init/1`.
- Convert `host` and `server_name_indication` to charlist when validating options.
- Return new state from `handle_continue` on connection error.
- Publish the `payload` field as message data, and the rest as metadata.

## 0.1.0 (2024-09-24)

_Initial release._

The initial release supports connecting to an MQTT broker using  [emqtt](https://github.com/emqx/emqtt), 
and consume messages using a Broadway pipeline.

**Supported features**
- [x] Support most  `emqtt` configurable options as producer config options.
- [x] Specify buffer size and overflow strategy for the `ets` table buffer.
- [x] `OffBroadway.EMQTT.MessageHandler` behaviour to support overriding default implementation.
- [x] Telemetry events for observability.