Update error handling on upload

src/IO/IOManagerAWS.cpp

@@ -240,6 +240,7 @@ namespace OpenVDS
   bool UploadRequestAWS::isSuccess(Error& error) const
   {
     std::unique_lock<std::mutex> lock(m_context->mutex);
+    error = m_error;
     return m_error.code == 0;
   }
   void UploadRequestAWS::cancel()

src/VDS/VolumeDataAccessManagerImpl.cpp

@@ -216,6 +216,7 @@ VolumeDataAccessManagerImpl::VolumeDataAccessManagerImpl(VDSHandle* handle)
   : m_layout(handle->volumeDataLayout.get())
   , m_ioManager(handle->ioManager.get())
   , m_layerMetadataContainer(&handle->layerMetadataContainer)
+  , m_currentErrorIndex(0)
 {
 }
 
@@ -553,10 +554,20 @@ void VolumeDataAccessManagerImpl::getCurrentUploadError(const char** objectId, i
 
   const auto &error = m_uploadErrors[m_currentErrorIndex];
   m_currentErrorIndex++;
+  if (objectId)
+    *objectId = error->urlObject.c_str();
+  if (errorCode)
+    *errorCode = error->error.code;
+  if (errorString)
+    *errorString = error->error.string.c_str();
   lock.unlock();
-  *objectId = error->urlObject.c_str();
-  *errorCode = error->error.code;
-  *errorString = error->error.string.c_str();
+}
+
+void VolumeDataAccessManagerImpl::addUploadError(const Error& error, VolumeDataLayer* layer, uint64_t chunk)
+{
+  std::string urlString = createUrlForChunk(createBaseUrl(layer), chunk);
+  std::unique_lock<std::mutex> lock(m_mutex);
+  m_uploadErrors.emplace_back(new UploadError(error, urlString));
 }
 
 }

src/VDS/VolumeDataAccessManagerImpl.h

@@ -209,6 +209,7 @@ public:
   int32_t uploadErrorCount() override;
   void getCurrentUploadError(const char **objectId, int32_t *errorCode, const char **errorString) override;
 
+  void addUploadError(const Error &error, VolumeDataLayer *layer, uint64_t chunk);
 private:
   VolumeDataLayout *m_layout;
   IOManager *m_ioManager;

src/VDS/VolumeDataPageImpl.cpp

@@ -20,6 +20,7 @@
 #include "VolumeDataChunk.h"
 #include "VolumeDataLayer.h"
 #include "VolumeDataAccessManagerImpl.h"
+#include "VolumeDataStore.h"
 #include <OpenVDS/VolumeDataChannelDescriptor.h>
 
 #include <algorithm>
@@ -176,12 +177,15 @@ void VolumeDataPageImpl::setBufferData(std::vector<uint8_t>&& blob, const int(&p
 void VolumeDataPageImpl::writeBack(VolumeDataLayer* volumeDataLayer, std::unique_lock<std::mutex>& pageListMutexLock)
 {
   assert(m_isDirty);
+  Error error;
   std::shared_ptr<std::vector<uint8_t>> to_write = std::make_shared<std::vector<uint8_t>>();
-  *to_write = m_blob;
+  if (!VolumeDataStore::serialize({volumeDataLayer, m_chunk}, m_blob, CompressionMethod::None, *to_write, error))
+  {
+    m_volumeDataPageAccessor->getManager()->addUploadError(error, volumeDataLayer, m_chunk);
+    return;
+  }
   m_volumeDataPageAccessor->requestWritePage(m_chunk, to_write);
-  ///IOManager *iomanager = m_volumeDataPageAccessor->wri
-
-  //iomanager->
+  m_isDirty = false;
 }
 
 

src/VDS/VolumeDataStore.cpp

@@ -414,4 +414,46 @@ bool VolumeDataStore::deserializeVolumeData(const VolumeDataChunk &volumeDataChu
 
   return true;
 }
+
+bool VolumeDataStore::serialize(const VolumeDataChunk& chunk, const std::vector<uint8_t>& chunkData, CompressionMethod compressionMethod, std::vector<uint8_t>& destinationBuffer, Error& error)
+{
+  destinationBuffer = chunkData;
+//    DataBlockDescriptor dataBlockDescriptor;
+//
+//    if(!dataBlockDescriptor->isValid())
+//    {
+//      error.code = -1;
+//      error.string = "Failed to decode DataBlockDescriptor";
+//      return false;
+//    }
+//
+//    if (!initializeDataBlock(*dataBlockDescriptor, dataBlock, error))
+//      return false;
+//
+//    void * source = dataBlockDescriptor + 1;
+//
+//    int32_t byteSize = getByteSize(*dataBlockDescriptor);
+//    std::unique_ptr<uint8_t[]>buffer(new uint8_t[byteSize]);
+//
+//    int32_t decompressedSize = rle_Decompress((uint8_t *)buffer.get(), byteSize, (uint8_t *)source);
+//    assert(decompressedSize == byteSize);
+//
+//    int allocatedSize = getAllocatedByteSize(dataBlock);
+//    destination.resize(allocatedSize);
+//    copyLinearBufferIntoDataBlock(buffer.get(), dataBlock, destination);
+//  switch(compressionMethod)
+//  {
+//  case CompressionMethod::None:
+//    destinationBuffer = layerData;
+//    return true;
+//  case CompressionMethod::Zip:
+//    return true;
+//  default:
+//    error.code = -20;
+//    error.string = "Not implemented compresssion algorithm";
+//    return false;
+//  }
+  return true;
+}
+
 }

src/VDS/VolumeDataStore.h

@@ -36,6 +36,7 @@ struct VolumeDataStore
   static bool verify(const VolumeDataChunk& volumeDataChunk, const std::vector<uint8_t>& serializedData, CompressionMethod compressionMethod, bool isFullyRead);
   static bool deserializeVolumeData(const VolumeDataChunk& chunk, const std::vector<uint8_t>& serializedData, const std::vector<uint8_t>& metadata, CompressionMethod compressionMethod, int32_t adaptiveLevel, VolumeDataChannelDescriptor::Format loadFormat, DataBlock &dataBlock, std::vector<uint8_t>& target, Error& error);
   static bool createConstantValueDataBlock(VolumeDataChunk const &volumeDataChunk, VolumeDataChannelDescriptor::Format format, float noValue, VolumeDataChannelDescriptor::Components components, VolumeDataHash const &constantValueVolumeDataHash, DataBlock &dataBlock, std::vector<uint8_t> &buffer, Error &error);
+  static bool serialize(const VolumeDataChunk &chunk, const std::vector<uint8_t> &chunkData, CompressionMethod compressionMethod, std::vector<uint8_t> &destinationBuffer, Error &error);
 };
 
 }