jsonstreamer

func Read ¶

func Read[T any](
	ctx context.Context,
	filename string,
	bufferSize int,
	options ...Func[T, T],
) error

Read is like ReadTo but with one defined type, setting default processor that returns the same document.

func ReadTo ¶

func ReadTo[Source any, Target any](
	ctx context.Context,
	filename string,
	bufferSize int,
	options ...Func[Source, Target],
) (errP error)

ReadTo reads JSON array entries from a file, processes them, and outputs the results through channels or callback functions. It supports batch processing with buffering and entry skipping capabilities.

Parameters:

• ctx: Context for cancellation control
• filename: Path to the JSON file containing an array of Source type elements
• bufferSize: Number of processed entries to accumulate before triggering output
• options: Configuration options.

The function expects the input file to contain a valid JSON array. It processes entries one by one, allowing for efficient memory usage with large files. The function provides multiple output mechanisms that can be used independently or in combination:

1. Buffer channel output (BufferCh): Processed entries are sent in batches
2. Per-entry callback: Called immediately after processing each entry
3. Per-buffer callback: Called when a buffer reaches bufferSize or at end of processing

Error Handling:

• Implements panic recovery with stack trace capture
• Returns custom errors for various failure scenarios
• Maintains error threshold (maxErrorCount) to prevent infinite processing
• Sends errors through ErrorCh for asynchronous handling
• Accumulates errors for final error reporting

Recovery Mechanism:

• Tracks the last valid entry offset for potential error recovery
• Maintains processing state through currentOffset
• Supports skipping invalid entries to continue processing

Context Support:

• Honors context cancellation for graceful shutdown
• Returns appropriate error when context is cancelled

The function implements graceful error handling and ensures resources are properly managed even in error conditions. It supports early termination through context cancellation and provides detailed error information for debugging purposes.

Flowchart:

```mermaid flowchart TD

Start([Start]) --> PanicRecover[Set up panic recovery]
PanicRecover --> ApplyOptions[Apply options]
ApplyOptions --> InputValidation{Check outputs}
InputValidation -->|Invalid| ReturnError1[Return Error: Missing outputs]
InputValidation -->|Valid| ValidateAmount{Validate amount}

ValidateAmount -->|Invalid| ReturnError2[Return Error: Invalid amount]
ValidateAmount -->|Valid| OpenFile[Open JSON file]

OpenFile -->|Error| ReturnError3[Return Error: Failed to open]
OpenFile -->|Success| ValidateArray[Validate JSON array structure]

ValidateArray -->|Invalid| ReturnError4[Return Error: Invalid format]
ValidateArray -->|Valid| ProcessLoop[Enter processing loop]

ProcessLoop --> CheckContext{Check context}
CheckContext -->|Cancelled| ReturnError5[Return Error: Context done]
CheckContext -->|Active| CheckErrors{Check error count}

CheckErrors -->|Too many| ReturnError6[Return Error: Too many errors]
CheckErrors -->|OK| CheckAmount{Check amount limit}

CheckAmount -->|Reached| HandleRemaining[Handle remaining buffer]
CheckAmount -->|Not reached| ReadEntry[Read JSON entry]

ReadEntry -->|Error| HandleDecodeError[Handle decode error]
ReadEntry -->|Success| CheckSkip{Skip entry?}

CheckSkip -->|Yes| UpdateOffset[Update last valid offset]
CheckSkip -->|No| ProcessDoc[Process document]

ProcessDoc -->|Error| SendError[Send to ErrorCh]
ProcessDoc -->|Success| UpdateCounter[Update CounterCh]

UpdateCounter --> HandleCallbacks[Handle callbacks]
SendError --> ProcessLoop

HandleCallbacks --> BufferCheck{Buffer full?}
BufferCheck -->|Yes| SendBuffer[Send buffer]
BufferCheck -->|No| ProcessLoop

SendBuffer --> ProcessLoop
HandleRemaining --> End([End])

HandleDecodeError --> ProcessLoop
UpdateOffset --> ProcessLoop

classDef error fill:#ff9999
class ReturnError1,ReturnError2,ReturnError3,ReturnError4,ReturnError5,ReturnError6 error

```.