Как поток данных из JPA может быть передан с микронавтом?

Question

Я сейчас начинаю работать с микронавтом и котлиным. У меня есть запрос JPA, который дает около 1 миллиона результатов. Эти результаты я хочу транслировать из одного сервиса микронавтов в другой.

Мой запрос возвращает allQuery.resultStream типа java.util.stream,

Контроллер службы отправки:

@Get("/test{value1,value2,value3}")
fun getTestObjects(
    value1: String,
    value2: String,
    value3: String
): Stream<TestObject> {
    val entries = testRepository.findAllWhere(value1, value2, value3)

    return entries
}

Клиент получающей услуги:

@Get("/data/test{value1,value2,value3}")
override fun getTestObjects(alue1: String,
    value2: String,
    value3: String) : Stream<TestObject>

И запрос JPA выглядит так:

    val cb = entityManager.criteriaBuilder
    val cq = cb.createQuery(TestObject::class.java)

    val rootEntry = cq.from(TestObject::class.java)

    val predicates = mutableListOf<Predicate>()

    predicates.add(cb.like(rootEntry.get<String>("value1"), value1))
    predicates.add(cb.equal(rootEntry.get<String>("value2"), value2))
    predicates.add(cb.equal(rootEntry.get<Int>("value3"), value3))

    val cqAllWhere = cq.select(rootEntry)
        .where(cb.or(*predicates.toTypedArray()))

    val allQuery = entityManager.createQuery(cqAllWhere)
    val entries = allQuery.resultStream

    return entries

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

Answer 1

В основном вам просто нужно создать Flowable и испускать предметы по мере их поступления.

return Flowable.create(emitter -> {

  //loop through result set
  //for each item
  emitter.onNext(item);

  //If you encounter an error
  emitter.onError(...);

  //When you're done
  emitter.onComplete();

})