Java Collector with Lockable Resource as Battery

Question

Suppose I try to create a collector that aggregates data into a resource that should be closed after use. Is there a way to implement something similar to a block finallyin Collector? In a successful case, this can be done using the method finisher, but there seems to be no method called in the case of exceptions.

The goal is to perform an operation, such as the following, in its purest form and without the need to first collect the stream into a list in memory.

stream.collect(groupingBy(this::extractFileName, collectToFile()));

Answer 1

, , , Stream.onClose. , :

class CloseHandler implements Runnable {
    List<Runnable> children = new ArrayList<>();

    void add(Runnable ch) { children.add(ch); }

    @Override
    public void run() { children.forEach(Runnable::run); }
}

:

CloseHandler closeAll = new CloseHandler();
try (Stream<Something> stream = list.stream().onClose(closeAll)) {
    // Now collect
    stream.collect(Collectors.groupingBy(
        this::extractFileName, 
        toFile(closeAll)));
}

try-with-resources, , . , close closeAll Stream.onClose.

, // Closeable ( , closeAll):

static Collector<Something, ?, Void> toFile(CloseHandler closeAll) {

    class Acc {

        SomeResource resource; // this is your closeable resource

        Acc() {
            try {
                resource = new SomeResource(...); // create closeable resource
                closeAll.add(this::close);        // this::close is a Runnable
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        }

        void add(Something elem) {
            try {
                // TODO write/send to closeable resource here
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        }

        Acc merge(Acc another) {
            // TODO left as an exercise
        }

        // This is the close handler for this particular closeable resource
        private void close() {
            try {
                // Here we close our closeable resource
                if (resource != null) resource.close();
            } catch (IOException ignored) {
            }
        }
    }
    return Collector.of(Acc::new, Acc::add, Acc::merge, a -> null);
}

, ( Acc) add , merge Acc , ( , ).

Collector.of Acc, , null, - , Collectors.groupingBy.

, close, .

try-with-resources, CloseHandler.run , , , , , Acc .

Answer 2

. Collectors, CollectorImpl , . , ( , ):

public class CollectorUtils<T, A, R> implements Collector<T, A, R> {

    static final Set<Collector.Characteristics> CH_ID = Collections
            .unmodifiableSet(EnumSet.of(Collector.Characteristics.IDENTITY_FINISH));

    private final Supplier<A> supplier;
    private final BiConsumer<A, T> accumulator;
    private final BinaryOperator<A> combiner;
    private final Function<A, R> finisher;
    private final Set<Characteristics> characteristics;

    CollectorUtils(Supplier<A> supplier, BiConsumer<A, T> accumulator, BinaryOperator<A> combiner,
            Function<A, R> finisher, Set<Characteristics> characteristics) {
        this.supplier = supplier;
        this.accumulator = accumulator;
        this.combiner = combiner;
        this.finisher = finisher;
        this.characteristics = characteristics;
    }

    CollectorUtils(Supplier<A> supplier, BiConsumer<A, T> accumulator, BinaryOperator<A> combiner,
            Set<Characteristics> characteristics) {
        this(supplier, accumulator, combiner, castingIdentity(), characteristics);
    }

    @Override
    public BiConsumer<A, T> accumulator() {
        return accumulator;
    }

    @Override
    public Supplier<A> supplier() {
        return supplier;
    }

    @Override
    public BinaryOperator<A> combiner() {
        return combiner;
    }

    @Override
    public Function<A, R> finisher() {
        return finisher;
    }

    @Override
    public Set<Characteristics> characteristics() {
        return characteristics;
    }

    @SuppressWarnings("unchecked")
    private static <I, R> Function<I, R> castingIdentity() {
        return i -> (R) i;
    }

    public static <C extends Collection<File>> Collector<String, ?, C> toFile() {
        return new CollectorUtils<>((Supplier<List<File>>) ArrayList::new, (c, t) -> {
            c.add(toFile(t));
        }, (r1, r2) -> {
            r1.addAll(r2);
            return r1;
        }, CH_ID);
    }

    private static File toFile(String fileName) {
        try (Closeable type = () -> System.out.println("Complete! closing file " + fileName);) {
            // stuff
            System.out.println("Converting " + fileName);

            return new File(fileName);
        } catch (FileNotFoundException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        throw new RuntimeException("Failed to create file");

    }

}

, :

public static void main(String[] args) {
        Stream.of("x.txt", "y.txt","z.txt").collect(CollectorUtils.toFile());
    }

:

Convertingx.txt
closing filex.txt
Convertingy.txt
closing filey.txt
Convertingz.txt
closing filez.txt