Consumers
Process and store messages in Magistrala using the TimescaleDB and PostgreSQL writers.
There is no standalone cmd/consumers binary — consumers/ is a library package imported by
cmd/timescale-writer and cmd/postgres-writer.
The Consumers Service in Magistrala handles the processing and storage of messages received from various channels via Writers, which persist data to databases like TimescaleDB and PostgreSQL as messages arrive.
Writers
Writers are responsible for persisting messages into databases for storage, analysis, and future retrieval. Magistrala supports two primary database writers:
Selective persistence
Writers consume the broker's writers stream, which the Rules Engine feeds through its
save_senml and save_json outputs. A message is stored because a rule selected it, not merely
because it was published. Both editions support this path: Community Edition is limited to one
rule per deployment, while Enterprise Edition has no rule-count limit. Without a matching storage
rule, the writers can start and report healthy but persist nothing. Subscribing a writer directly
to the m stream (all channel traffic) is not supported today; see Storage
Architecture.
TimescaleDB Writer
Stores messages into TimescaleDB with optimized handling for time-series data. This makes it ideal for IoT applications where devices generate continuous streams of timestamped data (like sensor readings).
Key Features of the TimescaleDB writer
- SenML Data Handling: Inserts SenML messages with detailed timestamps.
- JSON Data Handling: Dynamically creates tables if they don’t exist.
For example:
import "github.com/absmach/magistrala/consumers/timescale"
db := // initialize your TimescaleDB connection
consumer := timescale.New(db)
consumer.ConsumeBlocking(ctx, messages)PostgreSQL Writer
Persists messages in PostgreSQL databases, using UUIDs for unique identification. It is a general-purpose relational database storage for structured data. It's widely adopted for its reliability and strong transactional support.
Key Features of the PostgreSQL writer
- Auto Table Creation: Creates tables on-the-fly when missing.
- Transaction Safety: Ensures data integrity using transactions, with rollback support in case of errors.
For example:
import "github.com/absmach/magistrala/consumers/postgres"
db := // initialize your PostgreSQL connection
consumer := postgres.New(db)
consumer.ConsumeBlocking(ctx, messages)