Large HTTP responses and files are often processed through a simple but expensive pattern: read the complete content, deserialize it into a collection, and only then begin handling the items.

That model is convenient, but it couples three separate concerns: obtaining the byte stream, understanding the data format, and executing application logic. It also delays the first useful result and may retain far more data than the application actually needs.

The abstraction

Incremental processing means that a parser reads from an InputStream, recognizes one logical item, emits it immediately, and then continues reading. The client can process the first item before the complete input has arrived or been parsed.

The core is protocol-agnostic because protocol handling happens before its boundary. HTTP clients, file APIs, sockets, cloud-storage SDKs, archive readers, and other sources can all provide an InputStream. The library neither knows nor needs to know how that stream was obtained.

It is also format-agnostic. The stream may contain JSON objects, CSV rows, XML elements, text lines, or a custom binary representation. Those are parser responsibilities, not processor responsibilities.

Memory use and time-to-first-item

A fully buffered workflow must usually finish reading and often finish parsing before application processing begins. An incremental parser can emit the first logical item as soon as enough bytes are available to recognize it. This can reduce time-to-first-item because useful work starts before the complete payload is consumed.

The same model can keep memory usage low because the core does not retain the complete input or the complete sequence of emitted items. Instead, each item can be processed and released before the next one arrives.

These are architectural properties rather than unconditional benchmark guarantees. The benefit depends on the parser reading incrementally and on the consumer not retaining every emitted item. A parser that buffers the whole stream, or a consumer that accumulates all results, recreates the original memory and latency characteristics.

Three responsibility boundaries

inputstream-processor-core separates the flow into three roles:

  1. InputParser<T> understands the input format and emits logical items as they become available.
  2. InputStreamProcessor<T> connects the parser to the client consumer and counts successful consumer calls.
  3. Consumer<? super T> performs the application-specific action and owns the business failure policy.

The processor does not deserialize JSON, interpret CSV columns, persist entities, retry failed business operations, or decide which exceptions are recoverable. Each responsibility remains with the component that has enough context to make that decision correctly.

The core API

The parser is a functional interface: 

@FunctionalInterface
public interface InputParser<T> {
    void parse(
            InputStream input,
            Consumer<? super T> emitter
    ) throws IOException;
}

A processor is configured with one parser and then connects it to a client consumer: 

InputStreamProcessor<String> processor =
        new InputStreamProcessor<>(parser);

try (InputStream input = Files.newInputStream(path)) {
    ProcessingResult result = processor.process(
            input,
            System.out::println
    );

    System.out.println(result.getProcessedCount());
}

The Java library has no third-party runtime dependencies, supports Java 8 or later, and is published on Maven Central.

A line-oriented parser

For a plain-text stream, the parser can emit each line immediately: 

InputParser<String> parser = (input, emit) -> {
    BufferedReader reader = new BufferedReader(
            new InputStreamReader(input, StandardCharsets.UTF_8)
    );

    String line;
    while ((line = reader.readLine()) != null) {
        emit.accept(line);
    }
};

The same contract can be implemented with Jackson's streaming API, a CSV parser, an XML pull parser, or a domain-specific decoder. Those integrations remain outside the core artifact so applications can choose their own format libraries and versions.

What “incremental” actually guarantees

The processor itself never materializes the complete input or stores the complete sequence of emitted items. However, bounded-memory behavior still depends on the parser. A parser that reads the whole stream into memory before emitting anything defeats the incremental model even though it implements the same interface.

The responsibility boundary is therefore precise:

  • The core guarantees that it does not accumulate the input or emitted sequence.
  • The parser is responsible for reading and emitting incrementally.
  • The consumer is responsible for not retaining items unnecessarily.

Failure semantics

Parser and consumer exceptions propagate unchanged and terminate processing. The core does not catch, wrap, classify, or silently skip them.

The processed count is incremented only after a consumer call returns normally. If a later item fails, previously completed consumer calls remain completed; there is no rollback, and no ProcessingResult is returned from the failed invocation.

This may look minimal, but it keeps policy in the correct place. A JSON parser may know whether malformed input is recoverable. A business consumer may know whether an item should be skipped, retried, sent to a dead-letter queue, or allowed to stop the stream. The format-neutral core cannot make those decisions safely.

Stream ownership

The caller owns the supplied InputStream. Neither the processor nor the parser may close it. This makes the lifecycle explicit and prevents a nested reader or parser from unexpectedly closing a stream managed by an HTTP client, archive reader, or surrounding resource scope.

The normal usage pattern is therefore a caller-controlled try-with-resources block around process(...).

Execution and concurrency

Version 1 is synchronous and blocking. The parser emits items during its parse(...) call, and every consumer invocation completes before processing continues.

InputStreamProcessor is immutable, but safe concurrent reuse depends on the configured parser being thread-safe. Separate processing operations also require separate InputStream instances. Reactive processing, asynchronous emission, and internal parallelism are deliberately outside this core contract.

Why keep the core small?

A library that directly bundles JSON, CSV, XML, HTTP clients, retries, concurrency, and persistence would provide more features, but it would also own more policies, dependencies, and compatibility constraints.

The smaller design provides one stable mechanism: connect an incremental parser to an item consumer while preserving clear ownership and failure semantics. Format-aware adapters can be layered on top without changing that core contract.

The broader design lesson

Streaming is not only about replacing a byte array with an InputStream. A useful streaming abstraction must establish who recognizes item boundaries, who owns the input resource, when an item counts as processed, and where failure policy belongs.

The implementation can remain small when those boundaries are explicit.

Project: inputstream-processor-core on GitHub · Maven Central

LinkedIn post