1 package de.juplo.kafka.chat.backend.implementation.kafka;
3 import lombok.RequiredArgsConstructor;
4 import lombok.extern.slf4j.Slf4j;
7 @RequiredArgsConstructor
9 public class ConsumerTaskRunner
11 private final ConsumerTaskExecutor infoChannelConsumerTaskExecutor;
12 private final ConsumerTaskExecutor dataChannelConsumerTaskExecutor;
13 private final InfoChannel infoChannel;
15 public void executeConsumerTasks()
17 infoChannelConsumerTaskExecutor.executeConsumerTask();
18 dataChannelConsumerTaskExecutor.executeConsumerTask();
21 public void joinConsumerTasks() throws InterruptedException
23 dataChannelConsumerTaskExecutor.joinConsumerTaskJob();
24 while (infoChannel.isLoadInProgress())
26 log.info("Waiting for {} to finish loading...", infoChannel);
29 infoChannelConsumerTaskExecutor.joinConsumerTaskJob();