0

我是响应式编程的新手，并创建了我自己的第一个 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] [5, 6] 已完成但是 Assertsubscriber 接缝没有得到任何事件，既不是 onItem 也不是 onCompleted 我在某处犯了错误吗？

4

1 回答 1

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;
    }

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

于 2021-03-13T10:09:34.203 回答