测试自己的Mutiny操作符

Question

我是反应式编程的新手，并且创建了我自己的第一个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, 2, 3, 4), Arrays.asList(5, 6));
}

从我的第一个订阅者，我得到: 1，2，3，4已完成，但Assertsubscriber看起来没有得到任何事件，无论是onItem还是onCompleted，我是不是在哪里弄错了？

Answer 1

过了一段时间，我发现我犯了哪一个错误：

我没有正确使用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;
    }

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