Large-Dataset Streaming

Juneau supports streaming serialization and parsing of arbitrarily large datasets without loading all elements into memory. The feature is built directly into the serializer and parser APIs, so every format (JSON, XML, YAML, CSV, MessagePack, …) works transparently.

Key Interfaces

Interface	Extends	Direction	Purpose
BeanSupplier<T>	Iterable<T>	Serialization	Provides beans to serializers lazily from a cursor or query
BeanConsumer<T>	ThrowingConsumer<T>	Parsing	Receives beans one-at-a-time from a parser for insertion/processing
BeanChannel<T>	BeanSupplier<T> + BeanConsumer<T>	Both	Combines both roles for round-trip marshalling on a single property
ListBeanChannel<T>	BeanChannel<T>	Both	Built-in in-memory implementation backed by an ArrayList
BeanFactory<T>	—	Factory	Creates instances of any type T, enabling DI-framework integration

Lifecycle

All three interfaces share the same three-phase lifecycle, driven by the framework:

Phase	BeanSupplier (serialization)	BeanConsumer (parsing)
Setup	begin() — open cursor, execute query	begin() — open connection, prepare statement
Transfer	iterator() — yield one bean per call	acceptThrows(T) — receive one bean per call
Error	onError(Exception) — rollback / log; rethrow to stop	onError(Exception) — rollback / log; absorb to skip-and-continue
Cleanup	complete() — close cursor, connection	complete() — final commit, close statement

complete() is always called — even when onError() rethrows — so it is safe to use for resource cleanup in all cases.

---

BeanSupplier — Serializing from a Database Cursor

The following example streams a large Employee table from a JDBC cursor directly to an HTTP response, one row at a time, without ever loading the full table into memory.

@BeanType(factory=EmployeeSupplier.Factory.class)
public class EmployeeSupplier implements BeanSupplier<Employee> {

    private final DataSource ds;
    private Connection conn;
    private ResultSet rs;

    public EmployeeSupplier(DataSource ds) {
        this.ds = ds;
    }

    @Override
    public void begin() throws Exception {
        conn = ds.getConnection();
        var stmt = conn.prepareStatement(
            "SELECT id, name, department FROM employee ORDER BY id");
        rs = stmt.executeQuery();
    }

    @Override
    public Iterator<Employee> iterator() {
        return new Iterator<>() {
            @Override public boolean hasNext() { return ResultSetIterator.hasNext(rs); }
            @Override public Employee next()   { return Employee.fromRow(rs); }
        };
    }

    @Override
    public void onError(Exception e) throws Exception {
        throw e; // propagate; complete() will still close the cursor
    }

    @Override
    public void complete() throws Exception {
        if (rs   != null) rs.close();
        if (conn != null) conn.close();
    }

    // Factory is retrieved from Spring ApplicationContext via BeanStore.
    public static class Factory implements BeanFactory<EmployeeSupplier> {
        private final DataSource ds;
        public Factory(DataSource ds) { this.ds = ds; }

        @Override
        public EmployeeSupplier create() {
            return new EmployeeSupplier(ds);
        }
    }
}

In a Spring REST resource:

@Rest
public class EmployeeResource extends BasicRestServlet {

    @Inject
    private EmployeeSupplier.Factory supplierFactory;

    @RestGet("/employees")
    public EmployeeSupplier getEmployees() {
        return supplierFactory.create(); // framework calls begin(), iterates, calls complete()
    }
}

The serializer calls begin() before iterating, serializes each Employee as it arrives, and always calls complete() afterward — whether serialization succeeded or failed.

---

BeanConsumer — Parsing into a Database Table

The following example accepts a large JSON array from an HTTP request body and bulk-inserts the rows via JDBC, committing every 500 rows.

@BeanType(factory=EmployeeConsumer.Factory.class)
public class EmployeeConsumer implements BeanConsumer<Employee> {

    private static final int BATCH_SIZE = 500;

    private final DataSource ds;
    private Connection conn;
    private PreparedStatement stmt;
    private int count;

    public EmployeeConsumer(DataSource ds) {
        this.ds = ds;
    }

    @Override
    public void begin() throws Exception {
        conn = ds.getConnection();
        conn.setAutoCommit(false);
        stmt = conn.prepareStatement(
            "INSERT INTO employee (name, department) VALUES (?, ?)");
    }

    @Override
    public void acceptThrows(Employee emp) throws Exception {
        stmt.setString(1, emp.getName());
        stmt.setString(2, emp.getDepartment());
        stmt.executeUpdate();
        if (++count % BATCH_SIZE == 0)
            conn.commit(); // periodic batch commit
    }

    @Override
    public void onError(Exception e) throws Exception {
        conn.rollback();
        throw e; // stop parsing; complete() will still close resources
    }

    @Override
    public void complete() throws Exception {
        conn.commit(); // final commit for the last partial batch
        stmt.close();
        conn.close();
    }

    public static class Factory implements BeanFactory<EmployeeConsumer> {
        private final DataSource ds;
        public Factory(DataSource ds) { this.ds = ds; }

        @Override
        public EmployeeConsumer create() {
            return new EmployeeConsumer(ds);
        }
    }
}

Parsing via readToBeanConsumer:

// Direct API — framework calls begin(), acceptThrows() per element, complete().
var consumer = consumerFactory.create();
JsonParser.DEFAULT.getSession().readToBeanConsumer(inputStream, consumer, Employee.class);

In a REST resource:

@Rest
public class EmployeeResource extends BasicRestServlet {

    @Inject
    private EmployeeConsumer.Factory consumerFactory;

    @RestPost("/employees/bulk")
    public void importEmployees(RestRequest req) throws Exception {
        var consumer = consumerFactory.create();
        var parser = req.getContent().getParserMatch().orElseThrow().getParser();
        parser.getSession().readToBeanConsumer(req.getContent().getInputStream(), consumer, Employee.class);
    }
}

Fault-Tolerant Ingestion (Skip-and-Continue)

Override onError() to absorb exceptions instead of rethrowing. Parsing then continues to the next element, skipping only the bad record:

BeanConsumer<Employee> consumer = new BeanConsumer<>() {
    @Override
    public void acceptThrows(Employee emp) throws Exception {
        validateAndInsert(emp);
    }
    @Override
    public void onError(Exception e) {
        log.warn("Skipping invalid record: {}", e.getMessage());
        // absorb — parsing continues to the next element
    }
};

---

BeanChannel — Round-Trip Marshalling

BeanChannel<T> extends both BeanSupplier<T> and BeanConsumer<T>, allowing the same object to drive both serialization and parsing. The implementation decides which direction is active at runtime.

In-Memory: ListBeanChannel

ListBeanChannel<T> is a built-in implementation backed by an ArrayList. It requires no factory setup and is useful for testing, small datasets, or property-level round-trips:

public class EmployeeCollection {

    private final ListBeanChannel<Employee> employees = new ListBeanChannel<>();

    // No setter needed — parser calls acceptThrows() on the existing instance.
    @BeanProp(elementType=Employee.class)
    public ListBeanChannel<Employee> getEmployees() { return employees; }
}

// Serialize — channel iterated as a sequence.
String json = Json5Serializer.DEFAULT.write(collection);

// Parse — channel populated via acceptThrows().
Json5Parser.DEFAULT.read(json, EmployeeCollection.class);

Database-Backed Channel

@BeanType(factory=EmployeeChannel.Factory.class)
public class EmployeeChannel implements BeanChannel<Employee> {

    private final DataSource ds;
    private Connection conn;
    private ResultSet rs;
    private PreparedStatement insertStmt;
    private int insertCount;

    public EmployeeChannel(DataSource ds) {
        this.ds = ds;
    }

    // ---- BeanSupplier side (serialization) ----

    @Override
    public void begin() throws Exception {
        conn = ds.getConnection();
        insertStmt = conn.prepareStatement(
            "INSERT INTO employee (name, department) VALUES (?, ?)");
        conn.setAutoCommit(false);
        var qStmt = conn.prepareStatement(
            "SELECT name, department FROM employee ORDER BY id");
        rs = qStmt.executeQuery();
    }

    @Override
    public Iterator<Employee> iterator() {
        return new Iterator<>() {
            @Override public boolean hasNext() { return ResultSetIterator.hasNext(rs); }
            @Override public Employee next()   { return Employee.fromRow(rs); }
        };
    }

    // ---- BeanConsumer side (parsing) ----

    @Override
    public void acceptThrows(Employee emp) throws Exception {
        insertStmt.setString(1, emp.getName());
        insertStmt.setString(2, emp.getDepartment());
        insertStmt.executeUpdate();
        if (++insertCount % 500 == 0) conn.commit();
    }

    // ---- Shared lifecycle ----

    @Override
    public void onError(Exception e) throws Exception {
        if (conn != null) conn.rollback();
        throw e;
    }

    @Override
    public void complete() throws Exception {
        if (conn != null) {
            conn.commit();
            if (rs         != null) rs.close();
            if (insertStmt != null) insertStmt.close();
            conn.close();
        }
    }

    public static class Factory implements BeanFactory<EmployeeChannel> {
        private final DataSource ds;
        public Factory(DataSource ds) { this.ds = ds; }

        @Override
        public EmployeeChannel create() { return new EmployeeChannel(ds); }
    }
}

---

Spring Integration via BeanFactory and BeanStore

Spring-managed beans (e.g. a DataSource) can be injected into streaming implementations without reflection-based construction scanning. Register the factories as Spring beans, then wire a SpringBeanStore into the Juneau context.

Spring Configuration

@Configuration
public class JuneauStreamingConfig {

    @Bean
    public EmployeeSupplier.Factory employeeSupplierFactory(DataSource ds) {
        return new EmployeeSupplier.Factory(ds);
    }

    @Bean
    public EmployeeConsumer.Factory employeeConsumerFactory(DataSource ds) {
        return new EmployeeConsumer.Factory(ds);
    }

    @Bean
    public EmployeeChannel.Factory employeeChannelFactory(DataSource ds) {
        return new EmployeeChannel.Factory(ds);
    }
}

Wiring SpringBeanStore

import org.apache.juneau.commons.inject.Bean;

@Rest
public class EmployeeResource extends BasicSpringRestServlet {

    @Bean(name="beanStore")
    public static BeanStore beanStore(ApplicationContext appCtx) {
        return new SpringBeanStore(appCtx);
    }
}

With this wiring in place, any class annotated with @BeanType(factory=X.class) is automatically instantiated by retrieving the factory from the Spring ApplicationContext and calling BeanFactory.create(). No manual construction is needed in individual REST methods.

---

Supporting Annotations

Annotation	Target	Purpose
@BeanType(factory=X.class)	Class	Specifies the BeanFactory class used to instantiate this type during parsing. The factory is resolved from the BeanStore.
@BeanProp(factory=X.class)	Bean property	Specifies a property-level factory for a specific bean property value.
@BeanProp(elementType=Y.class)	Bean property	Declares the element type for generic streaming properties (Stream<Y>, BeanSupplier<Y>, etc.), overcoming Java type erasure. Also supports narrowing to concrete implementation types.

---

Supplier<T> Single-Value Unwrapping

The standard JDK Supplier<T> (not BeanSupplier) is treated as a single-value lazy wrapper. Serializers call get() and serialize the result transparently. Nested Supplier chains are unwrapped recursively up to a depth of 10.

note

BeanSupplier is not unwrapped — it is treated as an Iterable sequence, not a single-value wrapper.

// Serialized as the string "hello" — Supplier is unwrapped.
Supplier<String> lazy = () -> "hello";
String json = Json5Serializer.DEFAULT.write(lazy); // 'hello'

// Nested Supplier chains are also unwrapped.
Supplier<Supplier<Integer>> nested = () -> () -> 42;
json = Json5Serializer.DEFAULT.write(nested); // 42

---

Direction Validation

Using a BeanConsumer during serialization or a BeanSupplier during parsing throws an appropriate exception with a message recommending BeanChannel for bidirectional use cases. BeanChannel is accepted in both directions.