Documentation
¶
Overview ¶
Package compiler turns a parsed, validated workflow spec into the existing Mailer SDK objects (sources, operators, sinks, environment).
Compilation constructs objects only — it never opens a network connection. A Kafka consumer-group source builds a lazy reader (connects on Run, not construction); the parallel per-partition mode, which the SDK dials at construction to discover partitions, is therefore rejected here. Any liveness/connection check is a separate operation, not part of compile or validation.
Index ¶
- Constants
- func BuildPostgresMapper(table string, mapping map[string]string) (sink.RecordMapper, error)
- func CheckpointDir(workflowName, dataRoot string, rt *workflow.EnvSpec) string
- func CompileRuntime(workflowName, dataRoot string, rt *workflow.EnvSpec) (*mailer.StreamExecutionEnv, error)
- func CompileSink(spec workflow.SinkSpec) (sink.Sink, error)
- func CompileSource(spec workflow.SourceSpec) (source.Source, error)
- type CompiledWorkflow
- type Compiler
- type DeliveryGuarantee
- type GraphNode
- type PipelineGraph
Constants ¶
const DefaultDataRoot = "./data"
DefaultDataRoot is where per-workflow state and checkpoints live when no directory is configured.
Variables ¶
This section is empty.
Functions ¶
func BuildPostgresMapper ¶
BuildPostgresMapper builds a fixed-table RecordMapper from a field→column mapping. Every record maps to the SAME table (the name is a config constant, never derived from record data) and the mapped JSON fields become the columns. Column order is deterministic (sorted by column) so batched INSERTs are stable. A JSON field absent from a record yields a SQL NULL; json.Number values are coerced to int64 or float64 for correct numeric columns; nested objects/arrays are stored as JSON bytes.
func CheckpointDir ¶
CheckpointDir returns the directory a compiled workflow uses for its checkpoints, or "" when checkpointing is disabled. It mirrors the path CompileRuntime derives, so callers that must target the same storage without holding the env — e.g. the runner seeding a restored savepoint before Execute — can compute it identically.
func CompileRuntime ¶
func CompileRuntime(workflowName, dataRoot string, rt *workflow.EnvSpec) (*mailer.StreamExecutionEnv, error)
CompileRuntime builds a *mailer.StreamExecutionEnv configured from a workflow's runtime settings (buffer size, shutdown timeout, checkpointing, state backend). The returned env has no source or sink yet — the caller wires those with FromSource/…/ToSink.
State and checkpoint directories are made job-specific so two workflows can never share the same Pebble database:
<dataRoot>/<workflow-name>/state <dataRoot>/<workflow-name>/checkpoints
dataRoot defaults to DefaultDataRoot ("./data"). A directory explicitly set in the spec is honored as that resource's root, with the workflow name nested under it — so isolation holds either way.
func CompileSink ¶
CompileSink builds a sink.Sink from a sink spec. Kafka, transactional Kafka, stdout, and blackhole sinks construct without connecting; constructing a Postgres sink opens its connection pool (SDK behavior), so build it last, after the rest of a workflow validates.
func CompileSource ¶
func CompileSource(spec workflow.SourceSpec) (source.Source, error)
CompileSource builds a source.Source from a source spec. It returns an error for unsupported or invalid configuration but never connects.
Types ¶
type CompiledWorkflow ¶
type CompiledWorkflow struct {
Env *mailer.StreamExecutionEnv
Name string
Graph PipelineGraph
Delivery DeliveryGuarantee
// CheckpointDir is the on-disk directory holding this workflow's
// checkpoints, or "" when checkpointing is disabled. Exposed so the
// runner can seed a restored savepoint into the same storage.
CheckpointDir string
}
CompiledWorkflow bundles the executable environment with a static description of what was compiled.
type Compiler ¶
type Compiler struct {
// Secrets resolves ${VAR} references in sensitive connection fields.
// Defaults to secrets.Environment.
Secrets secrets.SecretResolver
// BaseDataDir is the root for per-workflow state/checkpoint
// directories. Defaults to DefaultDataRoot ("./data").
BaseDataDir string
}
Compiler turns a declarative workflow into an executable Mailer environment. It does not start the pipeline and, apart from creating job directories and (for a Postgres sink) opening a connection pool, establishes no connections.
func (*Compiler) Compile ¶
func (c *Compiler) Compile(spec *workflow.WorkflowSpec) (*mailer.StreamExecutionEnv, error)
Compile validates and compiles a workflow into an executable Mailer environment, ready to Execute. It returns an error without producing an environment if the workflow is invalid or references anything the declarative compiler cannot build.
func (*Compiler) CompileWorkflow ¶
func (c *Compiler) CompileWorkflow(spec *workflow.WorkflowSpec) (*CompiledWorkflow, error)
CompileWorkflow is Compile with the full compiled description.
Order: validate → resolve secrets → create source → create env (runtime config) → apply operators → create sink → return.
type DeliveryGuarantee ¶
type DeliveryGuarantee string
DeliveryGuarantee is the end-to-end guarantee a compiled workflow provides.
const ( AtMostOnce DeliveryGuarantee = "at-most-once" AtLeastOnce DeliveryGuarantee = "at-least-once" ExactlyOnce DeliveryGuarantee = "exactly-once" )
type PipelineGraph ¶
PipelineGraph is a static description of the compiled pipeline.
Source Files