Flux <Object> à InputStream

Question

Comment puis-je convertir le type Stream<Object> en InputStream? À l'heure actuelle, je reçois le iterator et boucle à travers toutes les données de conversion à un byteArray et l'ajouter à un fluxEntrée:

ByteArrayOutputStream bos = new ByteArrayOutputStream(); 
ObjectOutputStream oos = new ObjectOutputStream(bos); 

Iterator<MyType> myItr = MyObject.getStream().iterator(); 

while (myItr.hasNext()) { 

     oos.writeObject(myItr.next().toString() 
     .getBytes(StandardCharsets.UTF_8)); 
    } 
    oos.flush(); 
    oos.close(); 

    InputStream is = new ByteArrayInputStream(bao.toByteArray()); 

Quelle est la tête de le faire bien? Si mon flux contient un téraoctet de données, est-ce que je ne sucerais pas un téraoctet de données en mémoire? Y a-t-il un meilleur moyen d'y parvenir?

Answer 1

Vous devriez être en mesure de convertir le OutputStream en un InputStream en utilisant un tuyau:

PipedOutputStream pos = new PipedOutputStream(); 
InputStream is = new PipedInputStream(pos); 

new Thread(() -> { 
    try (ObjectOutputStream oos = new ObjectOutputStream(pos)) { 
     Iterator<MyType> myItr = MyObject.getStream().iterator(); 
     while (myItr.hasNext()) { 
      oos.writeObject(myItr.next().toString() 
       .getBytes(StandardCharsets.UTF_8)); 
     } 
    } catch (IOException e) { 
     // handle closed pipe etc. 
    } 
}).start(); 

Inspirée par this answer.

Answer 2

Cela fonctionnerait-il pour vous?

https://gist.github.com/stephenhand/292cdd8bba7a452d83c51c00d9ef113c

Il est une implémentation InputStream qui prend Stream<byte[]> comme données d'entrée. Vous avez juste besoin de .map() vos objets abitrary à des tableaux d'octets, mais vous voulez que chaque objet soit représenté par des octets.

Il appelle seulement une opération de terminal sur le Stream lorsque le InputStream est lu, tirer des objets hors Stream que le consommateur lit plus de InputStream il ne charge jamais l'ensemble en mémoire