blocks

package
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 4, 2026 License: Apache-2.0 Imports: 23 Imported by: 0

Documentation

Overview

Package blocks registers the built-in Flow block catalog used to instantiate stored pipelines (milestone v2.0: block factory).

Index

Constants

View Source 
const (
	TypeAstarteSource = "astarte_source"
	TypeNullSink      = "null_sink"
	TypeLogSink       = "log_sink"
	// TypeContainer is the Docker-backed custom processing block (Design B / #43).
	TypeContainer = container.Type

	// Pure transforms and the random source (#80, astarte_flow parity).
	TypeToJSON         = "to_json"
	TypeUpdateMetadata = "update_metadata"
	TypeSplitMap       = "split_map"
	TypeRandomSource   = "random_source"
	TypeSort           = "sort"

	// TypeJSONPathMap reshapes JSON messages through a template (#81,
	// astarte_flow parity); declared in jsonpathmap.go's neighborhood.
	TypeJSONPathMap = "json_path_map"

	// TypeVirtualDevicePool publishes messages as registered virtual devices
	// (#84, astarte_flow parity); declared in the virtualdevicepool package.
	TypeVirtualDevicePool = virtualdevicepool.Type
)

Well-known block_type strings stored in pipeline definitions.

View Source 
const (
	TypeHTTPSource = "http_source"
	TypeHTTPSink   = "http_sink"
)

HTTP block types (issue #82, astarte_flow parity): http_source polls GET URLs round-robin, http_sink POSTs message payloads to a URL. Both use net/http only and talk to whatever endpoints the operator configures.

View Source 
const (
	TypeMQTTSource = "mqtt_source"
	TypeMQTTSink   = "mqtt_sink"
)

MQTT block types (issue #83, astarte_flow parity): mqtt_source subscribes to broker topics and emits one binary message per delivery, mqtt_sink publishes message payloads to a topic. Both talk plain MQTT over TCP via paho.mqtt.golang, like the internal/broker client side.

View Source 
const (
	TypeFilter = "filter"
	TypeMap    = "map"
)

Well-known transform block_type strings.

Variables

This section is empty.

Functions

func AstarteSource 

func AstarteSource(name string, config map[string]any, deps flow.Deps) (flow.Block, error)

AstarteSource constructs an astartesource.Source. Config keys:

  • realm (string): tenant; defaults to deps.Realm
  • interface (string): optional interface filter
  • path (string): optional path prefix filter

func DefaultRegistry 

func DefaultRegistry() *flow.Registry

DefaultRegistry returns a registry with the minimum useful built-in set: AstarteSource (bus → Message), filter/map transforms, container (custom image via local Docker HTTP bridge), and null/log sinks so operators can compose a complete source→transform→sink pipeline.

func Filter 

func Filter(name string, config map[string]any, _ flow.Deps) (flow.Block, error)

Filter keeps or drops messages. All non-empty conditions must match (AND). Config keys (all optional; at least one required at construct time):

  • key_prefix (string): message Key must have this prefix
  • key_contains (string): message Key must contain this substring
  • type (string): wire type name — integer|real|boolean|datetime|binary|string|map
  • metadata (object string→string): every key must equal the given value

Matching messages pass through unchanged; non-matching messages are dropped (zero outputs, no error).

func HTTPSink 

func HTTPSink(name string, config map[string]any, _ flow.Deps) (flow.Block, error)

HTTPSink constructs a Sink that sends each message payload to a URL.

Config keys:

  • url (required string): target of every request
  • method (string, default "POST"): any non-empty token allowed
  • timeout_ms (int, default 5000): per-request timeout; must be positive
  • headers (object string→string): set verbatim on each request

Payload mapping: binary messages send their bytes with Subtype (or application/octet-stream) as Content-Type; strings send text/plain; everything else is JSON-encoded with application/json. Status >= 400 fails the message; 2xx/3xx succeed.

func HTTPSource 

func HTTPSource(name string, config map[string]any, _ flow.Deps) (flow.Block, error)

HTTPSource constructs a Source that polls configured GET URLs round-robin, emitting one binary message per response.

Config keys:

  • urls (required string array): at least one URL to poll; empty-string entries are rejected
  • interval_ms (int, default 1000): wait before every Emit (including the first); must be positive
  • timeout_ms (int, default 5000): per-request timeout; must be positive

func JSONPathMap 

func JSONPathMap(name string, config map[string]any, _ flow.Deps) (flow.Block, error)

JSONPathMap reshapes JSON messages through a template: placeholders in the template are substituted with the original payload or metadata values and the result is parsed as a JSON object payload.

Config keys:

  • template (string, required): a JSON document containing placeholders.

Placeholder grammar:

  • $MESSAGE — the entire original payload, rendered as raw JSON
  • $MESSAGE.<seg>(.<seg>)* — dot-separated path into the payload; each seg is one or more of [A-Za-z0-9_-]. Unresolvable paths fail Process.
  • $METADATA.<name> — metadata lookup; missing names fail Process. $MESSAGE followed by anything not matching .<seg> is the whole-payload token ($MESSAGES renders $MESSAGE then a literal "S").

Whole-message and path tokens insert raw (unquoted) JSON; time.Time becomes an RFC3339Nano string and []byte a base64 string before marshaling. Metadata tokens substitute the value JSON-escaped, so a template occurrence like "$METADATA.user" lands in the rendered document as the JSON string "alice". The output message is TypeMap with Key/Timestamp inherited, Metadata cloned, and per-field types inferred: float64→real, bool→boolean, string→string; nested objects, arrays and null become strings carrying their compact JSON rendering. The input message is never mutated.

func LogSink 

func LogSink(name string, _ map[string]any, _ flow.Deps) (flow.Block, error)

LogSink logs each message at debug level via slog.Default().

func MQTTSink 

func MQTTSink(name string, config map[string]any, _ flow.Deps) (flow.Block, error)

MQTTSink constructs a Sink that publishes each message payload to an MQTT topic (issue #83). Payload mapping follows http_sink's rule: binary sends its bytes as-is, strings send their text bytes, everything else is JSON-encoded.

Config keys:

  • url (required string): broker URL
  • topic (required string): target of every publish
  • qos (int, default 0), retained (bool, default false)
  • client_id, username, password (optional strings)

Construction connects eagerly; Stop disconnects on teardown.

func MQTTSource 

func MQTTSource(name string, config map[string]any, _ flow.Deps) (flow.Block, error)

MQTTSource constructs a Source that subscribes to MQTT topics and emits one binary message per received publish (issue #83).

Config keys:

  • url (required string): broker URL, e.g. tcp://127.0.0.1:1883
  • topics (required string array): at least one topic to subscribe to; empty-string entries are rejected
  • qos (int, default 0): subscription QoS, must be 0, 1 or 2
  • client_id, username, password (optional strings)

Construction connects eagerly and subscribes before returning, so config and connectivity errors fail fast.

func Map 

func Map(name string, config map[string]any, _ flow.Deps) (flow.Block, error)

Map rewrites key and/or metadata. The payload (Type/Data) is never changed. Config keys (all optional; at least one required):

  • key (string): new stream key; supports placeholders {key} and {metadata.<name>} (missing metadata expands to empty)
  • set_metadata (object string→string): merge into Metadata (overwrites)
  • delete_metadata ([]string): remove these Metadata keys after set

The message is shallow-copied so concurrent lanes never share Metadata maps.

func NullSink 

func NullSink(name string, _ map[string]any, _ flow.Deps) (flow.Block, error)

NullSink discards every message. Useful as a placeholder sink in tests and for pipelines that only need side effects from transforms.

func RandomSource 

func RandomSource(name string, config map[string]any, _ flow.Deps) (flow.Block, error)

RandomSource emits one random value per configured interval.

Config keys:

  • type (required string): integer | real | boolean
  • interval_ms (int, default 1000): must be positive; every Emit (including the first) waits one interval before returning a message
  • min / max (numbers, optional): bounds — inclusive [min,max] for integer (defaults 0..100), [min,max] for real (defaults 0.0..1.0); both ignored for boolean
  • key (string, default "random"): Key of the emitted messages

func Sort 

func Sort(name string, config map[string]any, _ flow.Deps) (flow.Block, error)

Sort buffers messages and releases them in ascending Timestamp order once a newer arrival makes them older than the window behind the newest buffered timestamp.

Config keys:

  • window_ms (int, default 1000): how long a message may trail the newest buffered timestamp before it is released; must be >= 0
  • dedup (bool, default false): drop a message whose full wire encoding equals one currently buffered

Window semantics: a message is held until either a newer arrival pushes it behind the window edge of the newest timestamp, or it is superseded in the buffer. The len>1 flush guard keeps at least the newest message buffered — the tail is never emitted by this block itself; it is only released when a later arrival overtakes it (buffered state is lost on teardown).

func SplitMap 

func SplitMap(name string, config map[string]any, _ flow.Deps) (flow.Block, error)

SplitMap explodes a TypeMap payload into one message per field. Non-map messages are dropped silently (zero outputs, no error).

Config keys (all optional):

  • key_template (string): output Key; placeholders {key} (original key) and {field} (field name); default "{key}/{field}"

Each output message carries a single field value as its payload: the type comes from FieldTypes when present, else is inferred from the Go value (bool→boolean, int64→integer, float64→real, anything else → string via fmt.Sprint). FieldSubtypes carry over per field; Metadata is cloned; Timestamp is inherited.

func ToJSON 

func ToJSON(name string, _ map[string]any, _ flow.Deps) (flow.Block, error)

ToJSON converts a message payload into its JSON representation, emitted as a TypeBinary message with subtype "application/json".

Config keys: none.

Map payloads are converted field-by-field (TypeBinary fields become base64 strings, TypeDatetime fields become RFC3339Nano strings); scalar datetime and binary payloads get the same treatment. The input message is never mutated.

func UpdateMetadata 

func UpdateMetadata(name string, config map[string]any, _ flow.Deps) (flow.Block, error)

UpdateMetadata merges set_metadata into a message's Metadata and then deletes delete_metadata keys (delete wins for the same key). Payload, key and timestamp are untouched.

Config keys (all optional; at least one required):

  • set_metadata (object string→string): merged in, overwriting existing keys
  • delete_metadata ([]string): removed after set applies

The message is shallow-copied so concurrent lanes never share Metadata maps.

Types

type Info 

type Info struct {
	// Type is the block_type string stored in pipeline definitions.
	Type string `json:"type"`
	// Role is source | transform | sink.
	Role Role `json:"role"`
	// Summary is a one-line description for operators.
	Summary string `json:"summary"`
	// Config notes key config fields (not a full JSON Schema).
	Config string `json:"config,omitempty"`
	// ConfigSchema is a Draft-07-style JSON Schema (object form) describing the
	// block's config keys, served through the blocks-discovery API so clients can
	// render forms without hardcoding block knowledge.
	ConfigSchema json.RawMessage `json:"config_schema,omitempty"`
}

Info is static operator documentation for a built-in block type.

func InfoForTypes 

func InfoForTypes(types []string) []Info

InfoForTypes returns Info for each type, filling a minimal stub when the type is registered but has no static docs (custom constructors).

func LookupInfo 

func LookupInfo(blockType string) (Info, bool)

LookupInfo returns static operator docs for a known built-in type. Unknown types return false (registry may still have a custom constructor).

type Role 

type Role string

Role is the operator-facing position of a block in a pipeline. 

const (
	RoleSource    Role = "source"
	RoleTransform Role = "transform"
	RoleSink      Role = "sink"
)

The three Role values a block can occupy in a pipeline.

Source Files

View all Source files

Directories

Path Synopsis
Package astartesource implements the AstarteSource Flow block (issue #27, astarte_flow parity): a Source that subscribes to Astrate's existing live event bus (internal/engine/stream) and converts device events into FlowMessages, connecting device ingestion to operator-defined pipelines.
 Package astartesource implements the AstarteSource Flow block (issue #27, astarte_flow parity): a Source that subscribes to Astrate's existing live event bus (internal/engine/stream) and converts device events into FlowMessages, connecting device ingestion to operator-defined pipelines.
Package container implements the Flow "container" block (Design B / #43 PoC).
 Package container implements the Flow "container" block (Design B / #43 PoC).
Package virtualdevicepool implements the virtual_device_pool block (issue #84): it publishes pipeline messages as registered virtual devices through the engine ingest path — storage rows without MQTT.
 Package virtualdevicepool implements the virtual_device_pool block (issue #84): it publishes pipeline messages as registered virtual devices through the engine ingest path — storage rows without MQTT.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.