body_stream.hpp

Go to the documentation of this file.

// Copyright (c) Microsoft Corporation. All rights reserved.
// SPDX-License-Identifier: MIT
 
#pragma once
 
#include "azure/core/platform.hpp"
 
#if defined(AZ_PLATFORM_POSIX)
#include <unistd.h>
#endif
 
#include "azure/core/context.hpp"
 
#include <algorithm>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <functional>
#include <memory>
#include <vector>
 
namespace Azure { namespace Core { namespace IO {
 
  class BodyStream {
  private:
    virtual size_t OnRead(uint8_t* buffer, size_t count, Azure::Core::Context const& context) = 0;
 
  public:
    virtual ~BodyStream() = default;
 
    virtual int64_t Length() const = 0;
 
    /*
     * @brief Resets the stream back to the beginning (for retries).
     * @remark Derived classes that send data in an HTTP request MUST override this and implement
     * it properly.
     */
    virtual void Rewind()
    {
      AZURE_ASSERT_MSG(
          false,
          "The specified BodyStream doesn't support Rewind which is required to guarantee fault "
          "tolerance when retrying any operation. Consider creating a MemoryBodyStream or "
          "FileBodyStream, which are rewindable.");
    }
 
    size_t Read(
        uint8_t* buffer,
        size_t count,
        Azure::Core::Context const& context = Azure::Core::Context())
    {
      AZURE_ASSERT(buffer || count == 0);
 
      context.ThrowIfCancelled();
      return OnRead(buffer, count, context);
    }
 
    size_t ReadToCount(
        uint8_t* buffer,
        size_t count,
        Azure::Core::Context const& context = Azure::Core::Context());
 
    std::vector<uint8_t> ReadToEnd(Azure::Core::Context const& context = Azure::Core::Context());
  };
 
  class MemoryBodyStream final : public BodyStream {
  private:
    const uint8_t* m_data;
    size_t m_length;
    size_t m_offset = 0;
 
    size_t OnRead(uint8_t* buffer, size_t count, Azure::Core::Context const& context) override;
 
  public:
    // Forbid constructor for rval so we don't end up storing dangling ptr
    MemoryBodyStream(std::vector<uint8_t> const&&) = delete;
 
    MemoryBodyStream(std::vector<uint8_t> const& buffer)
        : MemoryBodyStream(buffer.data(), buffer.size())
    {
    }
 
    explicit MemoryBodyStream(const uint8_t* data, size_t length) : m_data(data), m_length(length)
    {
      AZURE_ASSERT(data || length == 0);
    }
 
    int64_t Length() const override { return this->m_length; }
 
    void Rewind() override { m_offset = 0; }
  };
 
  namespace _internal {
    class RandomAccessFileBodyStream final : public BodyStream {
    private:
      // immutable
#if defined(AZ_PLATFORM_POSIX)
      int m_fileDescriptor;
#elif defined(AZ_PLATFORM_WINDOWS)
      void* m_filehandle;
#endif
      int64_t m_baseOffset;
      int64_t m_length;
      // mutable
      int64_t m_offset;
 
      size_t OnRead(uint8_t* buffer, size_t count, Azure::Core::Context const& context) override;
 
    public:
#if defined(AZ_PLATFORM_POSIX)
 
      RandomAccessFileBodyStream(int fileDescriptor, int64_t offset, int64_t length)
          : m_fileDescriptor(fileDescriptor), m_baseOffset(offset), m_length(length), m_offset(0)
      {
        AZURE_ASSERT(fileDescriptor >= 0 && offset >= 0 && length >= 0);
      }
 
      RandomAccessFileBodyStream() : m_fileDescriptor(0), m_baseOffset(0), m_length(0), m_offset(0)
      {
      }
 
#elif defined(AZ_PLATFORM_WINDOWS)
 
      RandomAccessFileBodyStream(void* fileHandle, int64_t offset, int64_t length)
          : m_filehandle(fileHandle), m_baseOffset(offset), m_length(length), m_offset(0)
      {
        AZURE_ASSERT(fileHandle && offset >= 0 && length >= 0);
      }
 
      RandomAccessFileBodyStream() : m_filehandle(NULL), m_baseOffset(0), m_length(0), m_offset(0)
      {
      }
#endif
 
      // Rewind seeks back to 0
      void Rewind() override { this->m_offset = 0; }
 
      int64_t Length() const override { return this->m_length; }
    };
 
  } // namespace _internal
 
  class FileBodyStream final : public BodyStream {
  private:
    // immutable
#if defined(AZ_PLATFORM_WINDOWS)
    void* m_filehandle;
#elif defined(AZ_PLATFORM_POSIX)
    int m_fileDescriptor;
#endif
    // mutable
    std::unique_ptr<_internal::RandomAccessFileBodyStream> m_randomAccessFileBodyStream;
 
    size_t OnRead(uint8_t* buffer, size_t count, Azure::Core::Context const& context) override;
 
  public:
    FileBodyStream(const std::string& filename);
 
    ~FileBodyStream();
 
    // Rewind seeks back to 0
    void Rewind() override;
 
    int64_t Length() const override;
  };
 
  class ProgressBodyStream : public BodyStream {
  private:
    BodyStream* m_bodyStream;
    int64_t m_bytesTransferred;
    std::function<void(int64_t bytesTransferred)> m_callback;
 
  private:
    size_t OnRead(uint8_t* buffer, size_t count, Azure::Core::Context const& context) override;
 
  public:
    ProgressBodyStream(
        BodyStream& bodyStream,
        std::function<void(int64_t bytesTransferred)> callback);
 
    void Rewind() override;
 
    int64_t Length() const override;
  };
}}} // namespace Azure::Core::IO