[FIXED] Wait subscribe method to get all data into array

Issue

I have an endpoint that its purpose is to receive a csv file, make a couple of changes to its name and then send this to a method to upload all the data it contains in a single file to Google Cloud as plain text.

The file can have more than 100,000 records, so when parsing it I have to save all the data in a variable and after that save it in Google Cloud. Today I can do it, but I am overwriting all the time the same file because I don't know how to indicate in the method to wait for the complete process of the subscribe and after that upload the file, so every time data is added to the array the file is uploaded again.

Although the method complies with the idea, I want to improve the performance of this, since to upload a file of only 2 mb with 100,000 records is taking approximately 15 minutes. Any idea?

private Storage storage;

private void uploadToGoogleCloudStorage(FilePart filePart, BlobInfo blobInfo) throws IOException {
    try (ByteArrayOutputStream bos = new ByteArrayOutputStream()) {
        filePart.content()
                .subscribe(dataBuffer -> {
                    byte[] bytes = new byte[dataBuffer.readableByteCount()];
                    dataBuffer.read(bytes);
                    DataBufferUtils.release(dataBuffer);
                    try {
                        bos.write(bytes);
                        storage.createFrom(blobInfo, new ByteArrayInputStream(bos.toByteArray()));
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });

    }
}

Solution

Finally i get the solution. I change the subscribe to map, then i get the last response from the flux and with that i subscribe the response to upload the data to google cloud store with the Storage interface (package from google to use their api)

private Storage storage;

private void uploadToGoogleCloudStorage(FilePart filePart, BlobInfo blobInfo) throws IOException {
    try (ByteArrayOutputStream bos = new ByteArrayOutputStream()) {
        filePart.content()
                .map(dataBuffer -> {
                    byte[] bytes = new byte[dataBuffer.readableByteCount()];
                    dataBuffer.read(bytes);
                    DataBufferUtils.release(dataBuffer);
                    try {
                        bos.write(bytes);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    return bos;
                }).last().subscribe(data -> {
                    try {
                        storage.createFrom(blobInfo, new ByteArrayInputStream(bos.toByteArray()));
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
    }

}

Answered By - Jonathan Luke
Answer Checked By - Terry (JavaFixing Volunteer)