Как поток данных из JPA может быть передан микронавтом? - PullRequest
Новая
       37

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

0 голосов
/ 25 января 2019

В настоящее время я начинаю работать с micronaut и kotlin.У меня есть запрос 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

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

1 Ответ

0 голосов
/ 26 января 2019

В основном вам просто нужно создать 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();

})
...