[FIXED] Cannot create hot stream from Queue

Issue

I have the following rest controller, which receives requests, transforms them into JSON strings and puts them into a concurrent queue. I would like to make a Flux out of this queue and subscribe to it.
Unfortunately, it doesn't work.
What am I doing wrong here?

@RestController
public class EventController {
    private final ObjectMapper mapper = new ObjectMapper();
    private final FirehosePutService firehosePutService;
    private ConcurrentLinkedQueue<String> events = new ConcurrentLinkedQueue<>();
    private int batchSize = 10;

    @Autowired
    public EventController(FirehosePutService firehosePutService) {
        this.firehosePutService = firehosePutService;

        Flux<String> eventFlux = Flux.create((FluxSink<String> sink) -> {
            String next;
            while (( next = events.poll()) != null) {
                sink.next(next);
            }
        });

        eventFlux.publish().autoConnect().subscribe(new BaseSubscriber<String>() {
            int consumed;
            List<String> batchOfEvents = new ArrayList<>(batchSize);

            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                request(batchSize);
            }

            @Override
            protected void hookOnNext(String value) {
                batchOfEvents.add(value);
                consumed++;

                if (consumed == batchSize) {
                    batchOfEvents.addAll(events);
                    log.info("Consume {} elements. Size of batchOfEvents={}", consumed, batchOfEvents.size());
                    firehosePutService.saveBulk(batchOfEvents);
                    consumed = 0;
                    batchOfEvents.clear();
                    events.clear();

                    request(batchSize);
                }
            }

        });
    }

    @GetMapping(value = "/saveMany", produces = "text/html")
    public ResponseEntity<Void> saveMany(@RequestParam MultiValueMap<String, String> allRequestParams) throws JsonProcessingException {
        Map<String, String> paramValues = allRequestParams.toSingleValueMap();
        String reignnEvent = mapper.writeValueAsString(paramValues);
        events.add(reignnEvent);

        return new ResponseEntity<>(HttpStatus.OK);
    }
}

Solution

First of all, you use poll method. It is not blocking and returns null if queue is empty. You loop collection until first null (i.e. while (next != null), so your code exits loop almost immediately because queue is empty on start. You must replace poll with take which is blocking and will wait until element is available.

Secondly, hookOnNext is invoked when the event is removed from the queue. However, you are trying to read events again using batchOfEvents.addAll(events);. Moreover, you also clear all pending events events.clear();

I advise you to remove all direct access to events collection from hookOnNext method.

---

Why do you use Flux here at all? Seems overcomplicated. You can use plain thread here

@Autowired
public EventController(FirehosePutService firehosePutService) {
    this.firehosePutService = firehosePutService;

    Thread persister = new Thread(() -> {
        List<String> batchOfEvents = new ArrayList<>(batchSize);

        String next;
        while (( next = events.take()) != null) {
            batchOfEvents.add(value);

            if (batchOfEvents.size() == batchSize) {
                log.info("Consume {} elements. Size of batchOfEvents={}", consumed, batchOfEvents.size());
                firehosePutService.saveBulk(batchOfEvents);
                batchOfEvents.clear();
            }
        }
    });
    persister.start();

}

Answered By - Alexander Pavlov