Risorsa di chiusura del flusso con Spring MVC

Question

Dopo aver letto this article, desidero utilizzare Spring per eseguire il flusso dei risultati della query del database direttamente su una risposta JSON per garantire l'utilizzo della memoria costante (nessun caricamento avido di un List in memoria).

Simile a quanto fatto nell'articolo con Hibernate, ho assemblato un oggetto greetingRepository che restituisce un flusso del contenuto del database basato su un JdbcTemplate. In tale implementazione, a creare un iteratore sulla interrogato ResultSet, e torno il flusso come segue:

return StreamSupport.stream(spliterator(), false).onClose(() -> { 
    log.info("Closing ResultSetIterator stream"); 
    JdbcUtils.closeResultSet(resultSet); 
}); 

cioè con un metodo onClose() garantire che il sottostante ResultSet sarà chiusa se il flusso è dichiarata in un try-with-resources costrutto :

try(Stream<Greeting> stream = greetingRepository.stream()) { 
    // operate on the stream 
} // ResultSet underlying the stream will be guaranteed to be closed 

Ma, come in questo articolo, voglio questo flusso per essere consumato da un mapper oggetto personalizzato (la maggiore MappingJackson2HttpMessageConverter definito nell'articolo). Se prendiamo il try-with-resources bisogno a parte, questo è fattibile come segue:

@RequestMapping(method = GET) 
Stream<GreetingResource> stream() { 
    return greetingRepository.stream().map(GreetingResource::new); 
} 

Tuttavia, come un collega ha commentato in fondo a questo articolo, questo non si prende cura di chiudere le risorse di base.

Nel contesto di Spring MVC, come posso eseguire lo streaming dal database fino in fondo in una risposta JSON e comunque garantire che lo ResultSet venga chiuso? Potresti fornire una soluzione di esempio concreta?

Answer 1

È possibile creare un costrutto per rinviare l'esecuzione della query al momento della serializzazione. Questo costrutto inizierà e terminerà la transazione in modo programmatico.

public class TransactionalStreamable<T> { 

    private final PlatformTransactionManager platformTransactionManager; 

    private final Callable<Stream<T>> callable; 

    public TransactionalStreamable(PlatformTransactionManager platformTransactionManager, Callable<Stream<T>> callable) { 
     this.platformTransactionManager = platformTransactionManager; 
     this.callable = callable; 
    } 

    public Stream stream() { 
     TransactionTemplate txTemplate = new TransactionTemplate(platformTransactionManager); 
     txTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); 
     txTemplate.setReadOnly(true); 

     TransactionStatus transaction = platformTransactionManager.getTransaction(txTemplate); 

     try { 
      return callable.call().onClose(() -> { 
       platformTransactionManager.commit(transaction); 
      }); 
     } catch (Exception e) { 
      platformTransactionManager.rollback(transaction); 
      throw new RuntimeException(e); 
     } 
    } 

    public void forEach(Consumer<T> c) { 
     try (Stream<T> s = stream()){ 
      s.forEach(c); 
     } 
    } 
} 

Utilizzando un JSON serializzatore dedicato:

JsonSerializer<?> transactionalStreamableSer = new StdSerializer<TransactionalStreamable<?>>(TransactionalStreamable.class, true) { 
    @Override 
    public void serialize(TransactionalStreamable<?> streamable, JsonGenerator jgen, SerializerProvider provider) throws IOException { 
     jgen.writeStartArray(); 
     streamable.forEach((CheckedConsumer) e -> { 
      provider.findValueSerializer(e.getClass(), null).serialize(e, jgen, provider); 
     }); 

     jgen.writeEndArray(); 
    } 
}; 

che potrebbe essere utilizzato in questo modo:

@RequestMapping(method = GET) 
TransactionalStreamable<GreetingResource> stream() { 
    return new TransactionalStreamable(platformTransactionManager ,() -> greetingRepository.stream().map(GreetingResource::new)); 
} 

Tutto il lavoro sarà fatto quando Jackson sarà serializzare l'oggetto. Può trattarsi o meno di un problema relativo alla gestione degli errori (ad esempio utilizzando il consiglio del controller).