Как записать java .util.stream.Stream в поток вывода StreamingResponseBody

Question

Я работаю над созданием API REST, в котором большой объем данных из базы данных Oracle можно отправлять порциями через потоковую передачу в клиентское приложение (например, загрузку файла или прямой поток).

I я получаю поток из JpaRepository, как указано ниже -

@Query("select u from UsersEntity u")
Stream<UsersEntity> findAllByCustomQueryAndStream();

Но теперь возникает проблема, записать этот поток в StreamingResponseBody Выходной поток

Я пытался многими способами, но безуспешно -

Первый подход -

Stream<UsersEntity> usersResultStream = usersRepository.findAllByCustomQueryAndStream();

        StreamingResponseBody stream = outputStream -> {
            Iterator<UsersEntity> iterator = usersResultStream.iterator();

            try (ObjectOutputStream oos = new ObjectOutputStream(outputStream)) {

                while (iterator.hasNext()) {
                    oos.write(iterator.next().toString().getBytes());
                }
            }
        };

Ошибка -

java.sql.SQLException: Closed Resultset: next
    at oracle.jdbc.driver.InsensitiveScrollableResultSet.next(InsensitiveScrollableResultSet.java:565) ~[ojdbc7-12.1.0.2.jar:12.1.0.2.0]

Второй подход -

StreamingResponseBody stream = new StreamingResponseBody() {

            @Transactional(readOnly = true)
            @Override
            public void writeTo(OutputStream outputStream) throws IOException {

                Stream<UsersEntity> usersResultStream = usersRepository.findAllByCustomQueryAndStream();

                try (ObjectOutputStream oos = new ObjectOutputStream(outputStream)) {

                    usersResultStream.forEach(user->{
                        try {
                            oos.write(user.toString().getBytes());
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    });
                }
            }
        }; 

Получена ошибка -

org.springframework.dao.InvalidDataAccessApiUsageException: You're trying to execute a streaming query method without a surrounding transaction that keeps the connection open so that the Stream can actually be consumed. Make sure the code consuming the stream uses @Transactional or any other way of declaring a (read-only) transaction.

Я загрузил тренировочный код по приведенной ниже ссылке - Образец заказа на поставку C Ссылка

У меня нет опыта выполнения задачи, связанной с потоковой передачей, поэтому, пожалуйста, помогите мне с этим.

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

Answer 1

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

Класс контроллера -

@RestController
@RequestMapping("/api")
public class UsersController {
    @Autowired
    private UserService service;

    @GetMapping(value = "/userstream")
    public ResponseEntity<StreamingResponseBody> fetchUsersStream() {

        StreamingResponseBody stream = this::writeTo;

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

    private void writeTo(OutputStream outputStream) {
        service.writeToOutputStream(outputStream);
    }
}

Класс обслуживания -

@Service
public class UserService {

    @Autowired
    private UsersRepository usersRepository;

    @Transactional(readOnly = true)
    public void writeToOutputStream(final OutputStream outputStream) {
        try (Stream<UsersEntity> usersResultStream = usersRepository.findAllByCustomQueryAndStream()) {
            try (ObjectOutputStream oos = new ObjectOutputStream(outputStream)) {

                usersResultStream.forEach(emp -> {
                    try {
                        oos.write(emp.toString().getBytes());
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

Полный код доступен на github - https://github.com/bagesh2050/HttpResponseStreamingDemo

Тем не менее, я желаю предложений, касающихся потоковой передачи по протоколу Http. Пожалуйста, предоставьте, если у вас есть лучшие идеи.

Answer 2

Ни в одном из примеров не показано "такое сложное" использование StreamingResponseBody, и я боюсь, что это "невозможно" (по крайней мере, я не смог справиться / исправить это с StreamingResponseBody и Потоковый запрос)

... но, что было возможно:

1. Использовать findAll() (обычный метод репозитория без потоков) в StreamingResponseBody.

   (Но Я понимаю необходимость выполнения асинхронного веб-запроса ... и запрос базы данных "передается" ...)

2. Использование Callable (асин c веб-запрос) и @Async CompletableFuture<..> (asyn c db request):

   @RestController
   @RequestMapping("/api")
   public class UsersController {
   
      @Autowired
      private UsersRepository usersRepository;
   
      @GetMapping(value = "/async/users")
      public Callable<List<UsersEntity>> fetchUsersAsync() {
          Callable callable = () -> {
              return usersRepository.readAllBy().get();
          };
          return callable;
      }
   }
   

   .. и репозиторий, такой как:

   @Repository
   public interface UsersRepository extends JpaRepository<UsersEntity, Integer> {
   
       @Async
       CompletableFuture<List<UsersEntity>> readAllBy();
   }
   

   (см. spring-samples ) .. не забудьте @EnableAsync в вашем приложении / конфигурации:

   @org.springframework.scheduling.annotation.EnableAsync
   @SpringBootApplication
   public class Application { ... }
   

Извините, это даже не ответ, но мои выводы - слишком долго для комментария.

Асинхронный веб-запрос может быть выполнен различными способами. (см. https://spring.io/blog/2012/05/10/spring-mvc-3-2-preview-making-a-controller-method-asynchronous/, https://niels.nu/blog/2016/spring-async-rest.html и даже не упомянутые «реактивные» API)