// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.

#include "azure/messaging/eventhubs/event_data_batch.hpp"

#include "private/eventhubs_constants.hpp"
#include "private/eventhubs_utilities.hpp"

#include <azure/core/diagnostics/logger.hpp>
#include <azure/core/internal/diagnostics/log.hpp>

using namespace Azure::Core::Diagnostics::_internal;
using namespace Azure::Core::Diagnostics;

namespace Azure { namespace Messaging { namespace EventHubs {

  bool EventDataBatch::TryAddMessage(Azure::Messaging::EventHubs::Models::EventData const& message)
  {
    return TryAddAmqpMessage(message.GetRawAmqpMessage());
  }

  Azure::Core::Amqp::Models::AmqpMessage EventDataBatch::ToAmqpMessage() const
  {
    Azure::Core::Amqp::Models::AmqpMessage returnValue{m_batchEnvelope};
    if (m_marshalledMessages.size() == 0)
    {
      throw std::runtime_error("No messages added to the batch.");
    }

    // Make sure that the partition key in the message is the current partition key.
    if (!m_partitionKey.empty())
    {
      returnValue.DeliveryAnnotations.emplace(
          _detail::PartitionKeyAnnotation, Azure::Core::Amqp::Models::AmqpValue(m_partitionKey));
    }

    std::vector<Azure::Core::Amqp::Models::AmqpBinaryData> messageList;
    for (auto const& marshalledMessage : m_marshalledMessages)
    {
      Azure::Core::Amqp::Models::AmqpBinaryData data(marshalledMessage);
      messageList.push_back(data);
    }

    returnValue.SetBody(messageList);
    return returnValue;
  }

  bool EventDataBatch::TryAddAmqpMessage(
      std::shared_ptr<Azure::Core::Amqp::Models::AmqpMessage const> const& message)
  {
    Azure::Core::Amqp::Models::AmqpMessage messageToSend{*message};

    // Fix up some properties in the message to send if they have not been already set.
    if (!message->Properties.MessageId.HasValue())
    {
      messageToSend.Properties.MessageId
          = Azure::Core::Amqp::Models::AmqpValue(Azure::Core::Uuid::CreateUuid().ToString());
    }

    if (!m_partitionKey.empty())
    {
      messageToSend.MessageAnnotations.emplace(
          _detail::PartitionKeyAnnotation, Azure::Core::Amqp::Models::AmqpValue(m_partitionKey));
    }

    auto serializedMessage = Azure::Core::Amqp::Models::AmqpMessage::Serialize(messageToSend);

    std::lock_guard<std::mutex> lock(m_rwMutex);

    if (m_marshalledMessages.size() == 0)
    {
      // The first message is special - we use its properties and annotations on the envelope for
      // the batch message.
      m_batchEnvelope = CreateBatchEnvelope(message);
      m_currentSize = serializedMessage.size();
    }
    auto actualPayloadSize = CalculateActualSizeForPayload(serializedMessage);
    if (m_currentSize + actualPayloadSize > m_maxBytes.Value())
    {
      m_currentSize = 0;
      m_batchEnvelope = nullptr;
      return false;
    }

    m_currentSize += actualPayloadSize;
    m_marshalledMessages.push_back(serializedMessage);
    return true;
  }

}}} // namespace Azure::Messaging::EventHubs

namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail {
  EventDataBatch EventDataBatchFactory::CreateEventDataBatch(EventDataBatchOptions const& options)
  {
    return EventDataBatch{options};
  }
}}}} // namespace Azure::Messaging::EventHubs::_detail
