package io.quarkus.smallrye.reactivemessaging.blocking.beans;

import java.util.ArrayList;
import java.util.List;

import javax.enterprise.context.ApplicationScoped;

import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams;
import org.eclipse.microprofile.reactive.streams.operators.SubscriberBuilder;

@ApplicationScoped
public class InfiniteSubscriber {

    private final List<Integer> payloads = new ArrayList<>();
    private final List<Integer> messages = new ArrayList<>();

    @Incoming("infinite-producer-payload")
    public SubscriberBuilder<Integer, Void> consumeFourItems() {
        return ReactiveStreams
                .<Integer> builder()
                .limit(4)
                .forEach(payloads::add);
    }

    @Incoming("infinite-producer-msg")
    public SubscriberBuilder<Integer, Void> consumeFourMessages() {
        return ReactiveStreams
                .<Integer> builder()
                .limit(4)
                .forEach(messages::add);
    }

    public List<Integer> payloads() {
        return payloads;
    }

    public List<Integer> messages() {
        return messages;
    }

}

@@ -73,21 +73,16 @@
 va.util.
-Array
 List;%0Aim
@@ -92,24 +92,51 @@
 t java.util.
+concurrent.CopyOnWriteArray
 List;%0A%0Aimpor
@@ -491,32 +491,43 @@
  payloads = new 
+CopyOnWrite
 ArrayList%3C%3E();%0A 
@@ -572,16 +572,27 @@
 s = new 
+CopyOnWrite
 ArrayLis

Fix concurrency modification exception that could happen in the reactive messaging tests.