测试自己的 Mutiny 操作员

问题描述

我是响应式编程的新手，并创建了我自己的第一个 Mutiny 运算符，如 https://smallrye.io/smallrye-mutiny/guides/plug 中所述。

我的操作员是一个收集器，它收集列表中的 x 个项目，并在到达 x 或流完成时发送这些项目。

public class ListCollectorUntil<T> extends AbstractMultiOperator<T,List<T>> {

ListCollectorUntil(Multi<? extends T> upstream) {
    super(upstream);
}

public ListCollectorUntil(Multi<? extends T> upstream,int count) {
    this(upstream);
    this.count = count;
}

private int count;
private final List<T> bufferList = new ArrayList<>();
private final List<T> tempBufferList = new ArrayList<>();

@Override
public void subscribe(MultiSubscriber<? super List<T>> downstream) {
    upstream.subscribe().withSubscriber(new ListCollectorProcessor(downstream));
}

private class ListCollectorProcessor extends MultiOperatorProcessor<T,List<T>> {

    public ListCollectorProcessor(MultiSubscriber<? super List<T>> downstream) {
        super(downstream);
    }

    @Override
    public void onItem(T item) {
        bufferList.add(item);
        if (bufferList.size() >= count) {
            this.downstream.onItem(getUni());
        }
    }

    @Override
    public void onCompletion() {
        Subscription subscription = this.upstream.getAndSet(Subscriptions.CANCELLED);
        if (subscription != Subscriptions.CANCELLED) {
            this.downstream.onItem(getUni());
            this.downstream.onCompletion();
        }
    }

    private List<T> getUni() {
        tempBufferList.clear();
        tempBufferList.addAll(bufferList);
        bufferList.clear();
        return tempBufferList;
    }
}

当我开始使用以下代码进行测试时：

void buffer_after5Inputs_uniShouldBeReturned() {
    Multi<List<Integer>> multi = Multi.createFrom().items(1,2,3,4,5,6)
            .plug( integerMulti -> new Buffer<>(integerMulti,4));
    multi.subscribe().with(
            integerList -> System.out.println(integerList),failur -> System.out.println(failur),() -> System.out.println("completed")
    );
    AssertSubscriber<List<Integer>> assertSubscriber =
            multi.subscribe().withSubscriber(AssertSubscriber.create(2))
                    .assertCompleted()
            .assertItems(Arrays.asList(1,4),Arrays.asList(5,6));
}

从我的第一个订阅者那里我得到： [1,4] [5,6] 完全的但是 Assertsubscriber 接缝没有事件，无论是 onItem 还是 onCompleted 我是不是哪里弄错了？

解决方法

过了一会儿，我弄清楚自己犯了哪个错误：

我没有正确使用 AssertSubscriber： create(int) 描述了从发射器请求的事件数量。就我而言，我需要 6 个事件

我没有制作我的列表的深层副本，而是使用我的新值更改为以前的列表。这看起来像是事件没有正确记录，但是事件指向的列表的值发生了变化。因此，显示了正确的值。我改变了我的代码：

 private List<T> getUni() {
    tempBufferList.clear();
    tempBufferList.addAll(bufferList);
    bufferList.clear();
    return tempBufferList;

到

private List<T> getUni() {
        List<T> tempBufferList = new ArrayList<>(bufferList);
        bufferList = new ArrayList<>();
        return tempBufferList;
    }

并且我的列表缓冲区工作正常并且测试正确运行。

java java mutiny