file

func DrainPatchAdapter ¶ added in v0.12.0

func DrainPatchAdapter[T any](
	f ports.File[T],
	varsFor func(map[string]any) map[string]string,
	opts DrainPatchAdapterOptions,
) ports.SinkAdapter[map[string]any]

DrainPatchAdapter returns a ports.SinkAdapter that applies each item as a partial update (JSON Merge Patch semantics) to an existing typed file via ports.File.Patch, instead of overwriting the whole file like DrainWriteFileAdapter. f is built by hand (not via ports.FilePattern): the stream's item type (map[string]any) is deliberately different from f's own type T, the same way ReadEachAdapter takes an independent content type. Use with ports.SinkPort.Bind:

domain.ConfigUpdates.Bind(ctx, file.DrainPatchAdapter(configFile,
    func(_ map[string]any) map[string]string { return nil },
    file.DrainPatchAdapterOptions{}))

f's format must be map-based (JSON, YAML, TOML, or format.New); otherwise every item fails with ports.FilePatchNotSupportedError (passed through to Options.OnError unchanged — see below).

When the bound ports.SinkPort declares Params, each varsFor result is validated with ports.ValidateParams before the patch; a validation failure is reported to Options.OnError as WriteError wrapping codex.ValidationErrors and the item is otherwise skipped (not patched). A ports.File.Patch failure (including ports.FilePatchNotSupportedError) is passed to Options.OnError unchanged, mirroring DrainWriteFileAdapter's treatment of ports.File.Write failures.

func DrainPatchEncodedAdapter ¶ added in v0.12.0

func DrainPatchEncodedAdapter[T, P any](
	f ports.File[T],
	patchCodec codex.Codec[P],
	varsFor func(P) map[string]string,
	opts DrainPatchEncodedAdapterOptions,
) ports.SinkAdapter[P]

DrainPatchEncodedAdapter returns a ports.SinkAdapter that applies each item as a typed partial update to an existing typed file via ports.PatchEncoded. Unlike DrainPatchAdapter's untyped map[string]any patches, patchCodec fields not present in f's own codec are still persisted — the right choice for intentionally adding new fields to a file. f is built by hand (not via ports.FilePattern): the stream's item type P is deliberately different from f's own type T. Use with ports.SinkPort.Bind:

domain.ConfigUpdates.Bind(ctx, file.DrainPatchEncodedAdapter(configFile, configPatchCodec,
    func(_ AppConfigPatch) map[string]string { return nil },
    file.DrainPatchEncodedAdapterOptions{}))

f's format must be map-based (JSON, YAML, TOML, or format.New); otherwise every item fails with ports.FilePatchNotSupportedError (passed through to Options.OnError unchanged — see below).

When the bound ports.SinkPort declares Params, each varsFor result is validated with ports.ValidateParams before the patch; a validation failure is reported to Options.OnError as WriteError wrapping codex.ValidationErrors and the item is otherwise skipped (not patched). A ports.PatchEncoded failure (including ports.FilePatchNotSupportedError) is passed to Options.OnError unchanged, mirroring DrainWriteFileAdapter's treatment of ports.File.Write failures.

func DrainWriteAdapter ¶

func DrainWriteAdapter[T any](
	w io.Writer,
	fmt format.Format[T],
	opts DrainWriteAdapterOptions,
) ports.SinkAdapter[T]

DrainWriteAdapter returns a ports.SinkAdapter that encodes each item and writes it as a line to w. Use with ports.SinkPort.Bind:

f, _ := os.Create("results.ndjson")
domain.OEEResults.Bind(ctx, file.DrainWriteAdapter(f, format.JSON(oeeCodec),
    file.DrainWriteAdapterOptions{Path: "results.ndjson"}))

func DrainWriteFileAdapter ¶

func DrainWriteFileAdapter[T any](
	f ports.File[T],
	varsFor func(T) map[string]string,
	opts DrainWriteFileAdapterOptions,
) ports.SinkAdapter[T]

DrainWriteFileAdapter returns a ports.SinkAdapter that writes each item as a complete typed file (whole-file overwrite). Use with ports.SinkPort.Bind:

domain.OEEResults.Bind(ctx, file.DrainWriteFileAdapter(resultFile,
    func(oee OEE) map[string]string { return map[string]string{"machineID": oee.MachineID} },
    file.DrainWriteFileAdapterOptions{}))

varsFor may be nil when f declares merge-capable path params (via ports.NewFilePathParam): vars are then derived PER-ITEM from each item's own merge fields automatically via ports.WriteHandle — the same "one struct, one call" convenience [mqtt5.PublishHandle] provides. Pass a non-nil varsFor to keep building the map yourself (e.g. no merge fields declared, or vars come from a field the file's own type doesn't have).

When the bound ports.SinkPort declares Params, each varsFor result is validated with ports.ValidateParams before the file write; a validation failure is reported to Options.OnError as WriteError wrapping codex.ValidationErrors and the item is otherwise skipped (not written).

func ReadAdapter ¶

func ReadAdapter[In, Resp any](
	f ports.File[Resp],
	varsFor func(In) map[string]string,
	opts ReadEachAdapterOptions,
) ports.IOAdapter[In, Resp]

ReadAdapter returns a ports.IOAdapter that reads a complete typed file for each In item and emits the file content directly as the response — the 2-type complement of ReadEachAdapter (whose independent file-content type and combine func serve enrichment). Pairs with a ports.FilePattern declared on a ports.IOPort[In, Resp], where the file content IS the port's response type:

calibFile, _ := ports.FileHandle[CalibrationData](domain.Calibration)
domain.Calibration.Bind(ctx, file.ReadAdapter(calibFile,
    func(r SensorReading) map[string]string {
        return map[string]string{"sensorID": r.SensorID}
    },
    file.ReadEachAdapterOptions{}))

When the bound ports.IOPort declares Params, each varsFor result is validated with ports.ValidateParams before the file read; a validation failure is delivered as ReadError wrapping codex.ValidationErrors.

Reads via ports.File.ReadMerged (inherited from ReadEachAdapter) — merge-capable path params declared via ports.NewFilePathParam are merged into the returned Resp automatically.

func ReadEachAdapter ¶

func ReadEachAdapter[In, T, Resp any](
	f ports.File[T],
	varsFor func(In) map[string]string,
	combine func(In, T) Resp,
	opts ReadEachAdapterOptions,
) ports.IOAdapter[In, Resp]

ReadEachAdapter returns a ports.IOAdapter that reads a complete typed file for each In item, combining the result. Use with ports.IOPort.Bind:

domain.Calibration.Bind(ctx, file.ReadEachAdapter(calibrationFile,
    func(r SensorReading) map[string]string { return map[string]string{"id": r.SensorID} },
    func(r SensorReading, c CalibrationData) CalibratedReading { return ... },
    file.ReadEachAdapterOptions{}))

When the bound ports.IOPort declares Params, each varsFor result is validated with ports.ValidateParams before the file read; a validation failure is delivered as ReadError wrapping codex.ValidationErrors.

The file is read via ports.File.ReadMerged — when f declares merge-capable path params (via ports.NewFilePathParam), the vars derived from varsFor(In) are ADDITIONALLY merged into the decoded T (e.g. a path-derived sensorID is populated onto T automatically). Identical to a bare ports.File.Read when f declares no merge fields.

func ScanAdapter ¶

func ScanAdapter[T any](path string, fmt format.Format[T], opts ScanAdapterOptions) ports.SourceAdapter[T]

ScanAdapter returns a ports.SourceAdapter that reads a file line-by-line, decoding each line. When the file is fully read the adapter exits. Use with ports.SourcePort.Bind:

domain.Readings.Bind(ctx, file.ScanAdapter("readings.ndjson", format.JSON(readingCodec),
    file.ScanAdapterOptions{}))

func WatchAdapter ¶

func WatchAdapter(dir string, interval time.Duration, opts WatchAdapterOptions) ports.SourceAdapter[string]

WatchAdapter returns a ports.SourceAdapter that emits file paths for new files created in dir. Runs until ctx is cancelled. Use with ports.SourcePort.Bind:

domain.NewFiles.Bind(ctx, file.WatchAdapter("/data/incoming", 5*time.Second,
    file.WatchAdapterOptions{}))