Package org.apache.juneau.commons.function

package org.apache.juneau.commons.function

Functional programming utilities including enhanced function interfaces, consumers, suppliers, tuples, exception-handling variants, and large-dataset streaming APIs.

Large-Dataset Streaming

The following interfaces enable serialization and parsing of large datasets without loading all elements into memory. They integrate directly into the Juneau marshalling framework so any serializer or parser works transparently.

• BeanSupplier - Serialize-side lifecycle interface; extends Iterable
• BeanConsumer - Parse-side lifecycle interface; extends ThrowingConsumer
• BeanChannel - Round-trip interface; extends both BeanSupplier and BeanConsumer
• ListBeanChannel - Built-in in-memory BeanChannel backed by an ArrayList
• BeanFactory - Universal factory interface for DI-framework-managed instantiation

Lifecycle

Both interfaces follow the same three-phase lifecycle driven by the framework:

Phase	BeanSupplier (serialization)	BeanConsumer (parsing)
Setup	BeanSupplier#begin() — open cursor, execute query	BeanConsumer#begin() — open connection, prepare statement
Transfer	BeanSupplier#iterator() — yield one bean per call	BeanConsumer#acceptThrows(Object) — receive one bean per call
Error	BeanSupplier#onError(Exception) — rollback / log; rethrow to stop	BeanConsumer#onError(Exception) — rollback / log; absorb to skip-and-continue
Cleanup	BeanSupplier#complete() — close cursor, connection	BeanConsumer#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 directly from the database to the HTTP response as a JSON array, without loading any rows into memory. A Spring-injected DataSource is supplied via a factory registered with BeanFactory.

Bean class with factory annotation

package com.example; @Marshalled(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 retrieved from the 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); } } }

Spring REST endpoint

@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 bean as it arrives from the cursor, and always calls complete() afterward to close the cursor and connection — regardless of whether serialization succeeded or failed.

BeanConsumer — Parsing into a database table

The following example accepts a large JSON array of Employee beans in an HTTP request body and bulk-inserts them into the database via JDBC, committing every 500 rows. On error it rolls back and rethrows; on completion it performs a final commit and closes resources.

BeanConsumer with batch commits

@Marshalled(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 parseToBeanConsumer

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

Spring REST endpoint

@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(); req.getBody().parseToBeanConsumer(consumer, Employee.class); } }

The onError() default rethrows, stopping parsing immediately. Override it to absorb the exception (log-and-skip) for fault-tolerant ingestion of partially-invalid input:

Fault-tolerant ingestion (skip bad records)

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 on a single property

BeanChannel extends both BeanSupplier and BeanConsumer, allowing the same property to drive both serialization and parsing. The implementation itself determines direction at runtime.

In-memory channel — ListBeanChannel

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

@Marshalled(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 constructor scanning or reflection. Register the factories as Spring beans, then wire the SpringBeanStore into the Juneau context so the marshaller can resolve them at runtime.

Spring configuration

@Configuration public class JuneauStreamingConfig { @Marshalled public EmployeeSupplier.Factory employeeSupplierFactory(DataSource ds) { return new EmployeeSupplier.Factory(ds); } @Marshalled public EmployeeConsumer.Factory employeeConsumerFactory(DataSource ds) { return new EmployeeConsumer.Factory(ds); } @Marshalled public EmployeeChannel.Factory employeeChannelFactory(DataSource ds) { return new EmployeeChannel.Factory(ds); } }

Wiring the SpringBeanStore into a REST resource

@Rest public class EmployeeResource extends BasicSpringRestServlet { @Inject private ApplicationContext appCtx; @Override protected MarshallingContext.Builder<?> createBeanContext(RestContext.Builder rcBuilder) { return super.createBeanContext(rcBuilder) .beanStore(new SpringBeanStore(appCtx)); } }

With this wiring in place, any class annotated with @BeanType(factory=EmployeeSupplier.Factory.class) is automatically instantiated by retrieving the factory from the Spring ApplicationContext and calling BeanFactory#create(). No manual construction or injection 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 instantiating the value of a specific bean property.
@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

Supplier<T> (the standard JDK interface) 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

Related Packages

Package	Description
org.apache.juneau.commons	Foundational classes and annotations shared across the Juneau Commons libraries.

Class	Description
BeanChannel<T>	A lifecycle-aware round-trip interface that combines both BeanSupplier (for serialization) and BeanConsumer (for parsing) on the same property or class.
BeanConsumer<T>	A lifecycle-aware consumer for use as a parser target, receiving deserialized beans one at a time.
BeanFactory<T>	A universal factory interface used with @Marshalled(factory=X.class) for any bean type the framework needs to instantiate.
BeanFactory.Void	Sentinel class used as the default value for @BeanType(factory=...) and @BeanProp(factory=...) when no factory is specified.
BeanSupplier<T>	A lifecycle-aware producer for use as a serializer source, providing beans lazily one at a time.
Consumer2<A,B>	A functional interface representing an operation that accepts two arguments and returns no result.
Consumer3<A,B,C>	A functional interface representing an operation that accepts three arguments and returns no result.
Consumer4<A,B,C,D>	A functional interface representing an operation that accepts four arguments and returns no result.
Consumer5<A,B,C,D,E>	A functional interface representing an operation that accepts five arguments and returns no result.
Function2<A,B,R>	A functional interface representing a function that accepts two arguments and produces a result.
Function3<A,B,C,R>	A functional interface representing a function that accepts three arguments and produces a result.
Function4<A,B,C,D,R>	A functional interface representing a function that accepts four arguments and produces a result.
Function5<A,B,C,D,E,R>	A functional interface representing a function that accepts five arguments and produces a result.
ListBeanChannel<T>	A simple in-memory BeanChannel implementation backed by an ArrayList.
Memoizer<T>	A thread-safe supplier that caches the result of the first call and supports resetting the cache.
NullableSupplier<T>	A Supplier whose value may be null, with Optional-like null-safe query methods.
Snippet	A functional interface identical to Runnable but allows the Snippet.run() method to throw checked exceptions.
Suppliers	Supplier factories: memoizing wrappers moved from the former Utils.
ThrowingConsumer<T>	A functional interface representing an operation that accepts a single argument, returns no result, and may throw a checked exception.
ThrowingConsumer2<A,B>	A functional interface representing an operation that accepts two arguments, returns no result, and may throw a checked exception.
ThrowingConsumer3<A,B,C>	A functional interface representing an operation that accepts three arguments, returns no result, and may throw a checked exception.
ThrowingConsumer4<A,B,C,D>	A functional interface representing an operation that accepts four arguments, returns no result, and may throw a checked exception.
ThrowingConsumer5<A,B,C,D,E>	A functional interface representing an operation that accepts five arguments, returns no result, and may throw a checked exception.
ThrowingFunction<T,R>	A functional interface representing a function that accepts one argument, produces a result, and may throw a checked exception.
ThrowingFunction2<A,B,R>	A functional interface representing a function that accepts two arguments, produces a result, and may throw a checked exception.
ThrowingFunction3<A,B,C,R>	A functional interface representing a function that accepts three arguments, produces a result, and may throw a checked exception.
ThrowingFunction4<A,B,C,D,R>	A functional interface representing a function that accepts four arguments, produces a result, and may throw a checked exception.
ThrowingFunction5<A,B,C,D,E,R>	A functional interface representing a function that accepts five arguments, produces a result, and may throw a checked exception.
ThrowingSupplier<T>	A functional interface representing a supplier of results that may throw a checked exception.
Tuple1<A>	Represents an immutable tuple containing a single value.
Tuple2<A,B>	Represents an immutable tuple containing two values.
Tuple3<A,B,C>	Represents a simple tuple of 3 objects.
Tuple4<A,B,C,D>	Represents a simple tuple of 4 objects.
Tuple5<A,B,C,D,E>	Represents a simple tuple of 5 objects.