Странное поведение с конечной точкой SSE с использованием Redis

Question

Мне нужно передать некоторые данные клиенту, если он находится в Redis, но клиент продолжает подключаться к конечной точке SSE каждые 5 секунд.

Код бэкэнда:

@RestController
@RequestMapping("/reactive-task")
public class TaskRedisController {

    private final TaskRedisRepository taskRedisRepository;

    TaskRedisController(TaskRedisRepository taskRedisRepository){
        this.taskRedisRepository = taskRedisRepository;
    }

   @CrossOrigin
    @GetMapping(value = "/get/{id}")
    public Flux<ServerSentEvent<Task>> getSseStream2(@PathVariable("id") String id) {
        return taskRedisRepository.findByTaskId(id)
            .map(task -> ServerSentEvent.<Task>builder().data(task).build());
    }
}

@Repository
public class TaskRedisRepository {
    public Flux<Task> findByTaskId(String id) {
        return template.keys("task:" + id).flatMap(template.opsForValue()::get);
    } 
}


@AllArgsConstructor
@NoArgsConstructor
@Getter
@Setter
@EqualsAndHashCode
@Entity
public class Task {
    @Id
    @GeneratedValue(strategy=GenerationType.IDENTITY)
    private Long id;
    @Column(length = 25)
    private String result;
}

Клиент использует с помощью JS:

var evt = new EventSource("http://localhost:8080/reactive-task/get/98"); evt.onmessage = function(event) {
  console.log(event);
};

Может кто-то указать мне верное направление?
Любой совет будет принят во внимание.

Обновление: мне нужно хранить данные в течение некоторого времени (5-10 минут) в Redis. Обновление: я написал похожий код на MongoDB, и он отлично работает.

Answer 1

В этом случае, taskRedisRepository.findByTaskId(id) вероятно отправляет конечное Flux - имеется в виду несколько элементов и завершенный сигнал окончания потока.

Spring WebFlux будет интерпретировать onComplete сигнал как конец потока и закроет его. Поведение клиентов SSE в браузере по умолчанию - это повторное подключение сразу к конечной точке в случае разрыва соединения.

Если вы хотите сохранить постоянное соединение и получать уведомления только о добавляемых новых элементах, вам необходимо использовать это непосредственно как функцию вашего хранилища данных. Для Redis этот режим поддерживается функцией pub/sub ( см. Справочную документацию).

Подводя итог, я думаю, вы видите ожидаемое поведение, так как в этом случае ваше хранилище данных не будет генерировать бесконечный поток, уведомляющий вас о новых элементах, добавленных в коллекцию, а скорее конечный поток элементов, присутствующих в этой коллекции в данный момент времени.,