PipeLine发布资源lambda

Question

我有以下问题。我已经实现了一个管道处理，有时我必须释放资源，例如文件。同时，我的管道是异步的，所以它立即释放控制，我在非lamda实现中有一个关闭方法。我无法在使用兰巴达时插上它。我的当前代码如下：

@FunctionalInterface
public interface Stage<T,Q>  {

    public Q process(T toProcess);

}


@FunctionalInterface
public interface IntermediatStage<T,Q>  extends Stage<T,Q> {


    public default <P> IntermediatStage<T,P> nextStage(Stage<Q,P> nextStage) {
        return  (T t) -> {return nextStage.process(this.process(t)); }; 
    }



}

随后，我需要在不同的阶段调用一个关闭方法。问题是，用默认方法将它们链接起来，我在它们上没有可见性。当我添加阶段也能够添加一个关闭钩子时，基于管道是异步的事实，我可以在以后独立调用这个钩子，这是否可能呢？

谢谢

Answer 1

您不能只使用接口和lambda表达式来解决所有问题。对于具有如下模式的普通类来说，这看起来是一个简单的任务：

public final class Stage<T,R> {
    static final Runnable NO_OP = () -> {};
    public static <I,O> Stage<I,O> create(Function<I,O> f) {
        return new Stage<>(f, NO_OP);
    }
    public static <I,O> Stage<I,O> create(Function<I,O> f, Runnable cleanup) {
        return new Stage<>(f, cleanup);
    }
    private final Function<T,R> actualAction;
    private final Runnable cleanup;

    private Stage(Function<T,R> f, Runnable r) {
        actualAction=f;
        cleanup=r;
    }

    public <P> Stage<T,P> nextStage(Function<R,P> nextStage) {
        return new Stage<>(actualAction.andThen(nextStage), cleanup);
    }
    public <P> Stage<T,P> nextStage(Function<R,P> nextStage, Runnable nextCleanup) {
        return new Stage<>(actualAction.andThen(nextStage),
            cleanup==NO_OP? nextCleanup: () -> { cleanup.run(); nextCleanup.run(); });
    }
    public R process(T t) {
        return actualAction.apply(t);
    }
    public Function<T, R> getActualAction() {
        return actualAction;
    }
    public void cleanup() {
        cleanup.run();
    }
    public Runnable getCleanup() {
        return cleanup;
    }
}

Stage类是简单和不变的，但是它的实际行为是由Function和Runnable实例决定的，如果您愿意的话，这些实例可以通过lambda表达式创建。您可以使用纯函数创建和链接，也可以通过提供函数和清理操作来创建和链接。您将为函数和清理操作获得两个不同的链，因此可以独立执行它们。

Answer 2

如果没有更多关于如何清理资源的信息，人们就可以想象出这样的情况

@FunctionalInterface
public interface Stage<T,Q>  {

    Q process(T toProcess);

    static <T,Q> Stage<T,Q> of(Stage<T,Q> stage){
        return stage;
    }

    default Stage<T,Q> withRelease(Consumer<T> releaser){
        return t -> {
            Q q = process(t);
            releaser.accept(t);
            return q;
        };
    }
}

你可以调用如下

Stage.of(Thing::process).withRelease(Thing::close)