* The blob will be compressed and checksum will be written if required. * @@ -131,20 +130,12 @@ public T readBlob(BlobContainer blobContainer, String blobName) throws IOExcepti * @param name blob name */ public void writeAtomic(T obj, BlobContainer blobContainer, String name) throws IOException { - String blobName = blobName(name); - String tempBlobName = tempBlobName(name); - writeBlob(obj, blobContainer, tempBlobName); - try { - blobContainer.move(tempBlobName, blobName); - } catch (IOException ex) { - // Move failed - try cleaning up - try { - blobContainer.deleteBlob(tempBlobName); - } catch (Exception e) { - ex.addSuppressed(e); + final String blobName = blobName(name); + writeTo(obj, blobName, bytesArray -> { + try (InputStream stream = bytesArray.streamInput()) { + blobContainer.writeBlobAtomic(blobName, stream, bytesArray.length()); } - throw ex; - } + }); } /** @@ -157,51 +148,35 @@ public void writeAtomic(T obj, BlobContainer blobContainer, String name) throws * @param name blob name */ public void write(T obj, BlobContainer blobContainer, String name) throws IOException { - String blobName = blobName(name); - writeBlob(obj, blobContainer, blobName); + final String blobName = blobName(name); + writeTo(obj, blobName, bytesArray -> { + try (InputStream stream = bytesArray.streamInput()) { + blobContainer.writeBlob(blobName, stream, bytesArray.length()); + } + }); } - /** - * Writes blob in atomic manner without resolving the blobName using using {@link #blobName} method. - *
- * The blob will be compressed and checksum will be written if required.
- *
- * @param obj object to be serialized
- * @param blobContainer blob container
- * @param blobName blob name
- */
- protected void writeBlob(T obj, BlobContainer blobContainer, String blobName) throws IOException {
- BytesReference bytes = write(obj);
- try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) {
+ private void writeTo(final T obj, final String blobName, final CheckedConsumer