file

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 format.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{}))

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 format.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].

func ReadEachAdapter ¶

func ReadEachAdapter[In, T, Resp any](
	f format.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].

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{}))