/*

* Licensed to the Apache Software Foundation (ASF) under one or more

* contributor license agreements. See the NOTICE file distributed with

* this work for additional information regarding copyright ownership.

* The ASF licenses this file to You under the Apache License, Version 2.0

* (the "License"); you may not use this file except in compliance with

* the License. You may obtain a copy of the License at

*

* http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/

#include "MessageExtImpl.h"

#include <sstream> // std::stringstream

#include "MessageClientIDSetter.h"

#include "MessageSysFlag.h"

#include "SocketUtil.h"

#include "UtilAll.h"

namespace rocketmq {

// ============================

// MessageExtImpl

// ============================

MessageExtImpl::MessageExtImpl() : MessageExtImpl(0, 0, nullptr, 0, nullptr, null) {}

MessageExtImpl::MessageExtImpl(int queueId,

int64_t bornTimestamp,

const struct sockaddr* bornHost,

int64_t storeTimestamp,

const struct sockaddr* storeHost,

const std::string& msgId)

: store_size_(0),

body_crc_(0),

queue_id_(queueId),

queue_offset_(0),

commit_log_offset_(0),

sys_flag_(0),

born_timestamp_(bornTimestamp),

born_host_(SockaddrToStorage(bornHost)),

store_timestamp_(storeTimestamp),

store_host_(SockaddrToStorage(storeHost)),

reconsume_times_(3),

prepared_transaction_offset_(0),

msg_id_(msgId) {}

MessageExtImpl::~MessageExtImpl() = default;

TopicFilterType MessageExtImpl::parseTopicFilterType(int32_t sysFlag) {

if ((sysFlag & MessageSysFlag::MULTI_TAGS_FLAG) == MessageSysFlag::MULTI_TAGS_FLAG) {

return MULTI_TAG;

}

return SINGLE_TAG;

}

int32_t MessageExtImpl::store_size() const {

return store_size_;

}

void MessageExtImpl::set_store_size(int32_t storeSize) {

store_size_ = storeSize;

}

int32_t MessageExtImpl::body_crc() const {

return body_crc_;

}

void MessageExtImpl::set_body_crc(int32_t bodyCRC) {

body_crc_ = bodyCRC;

}

int32_t MessageExtImpl::queue_id() const {

return queue_id_;

}

void MessageExtImpl::set_queue_id(int32_t queueId) {

queue_id_ = queueId;

}

int64_t MessageExtImpl::queue_offset() const {

return queue_offset_;

}

void MessageExtImpl::set_queue_offset(int64_t queueOffset) {

queue_offset_ = queueOffset;

}

int64_t MessageExtImpl::commit_log_offset() const {

return commit_log_offset_;

}

void MessageExtImpl::set_commit_log_offset(int64_t physicOffset) {

commit_log_offset_ = physicOffset;

}

int32_t MessageExtImpl::sys_flag() const {

return sys_flag_;

}

void MessageExtImpl::set_sys_flag(int32_t sysFlag) {

sys_flag_ = sysFlag;

}

int64_t MessageExtImpl::born_timestamp() const {

return born_timestamp_;

}

void MessageExtImpl::set_born_timestamp(int64_t bornTimestamp) {

born_timestamp_ = bornTimestamp;

}

const struct sockaddr* MessageExtImpl::born_host() const {

return reinterpret_cast<sockaddr*>(born_host_.get());

}

std::string MessageExtImpl::born_host_string() const {

return SockaddrToString(born_host());

}

void MessageExtImpl::set_born_host(const struct sockaddr* bornHost) {

born_host_ = SockaddrToStorage(bornHost);

}

int64_t MessageExtImpl::store_timestamp() const {

return store_timestamp_;

}

void MessageExtImpl::set_store_timestamp(int64_t storeTimestamp) {

store_timestamp_ = storeTimestamp;

}

const struct sockaddr* MessageExtImpl::store_host() const {

return reinterpret_cast<sockaddr*>(store_host_.get());

}

std::string MessageExtImpl::store_host_string() const {

return SockaddrToString(store_host());

}

void MessageExtImpl::set_store_host(const struct sockaddr* storeHost) {

store_host_ = SockaddrToStorage(storeHost);

}

const std::string& MessageExtImpl::msg_id() const {

return msg_id_;

}

void MessageExtImpl::set_msg_id(const std::string& msgId) {

msg_id_ = msgId;

}

int32_t MessageExtImpl::reconsume_times() const {

return reconsume_times_;

}

void MessageExtImpl::set_reconsume_times(int32_t reconsumeTimes) {

reconsume_times_ = reconsumeTimes;

}

int64_t MessageExtImpl::prepared_transaction_offset() const {

return prepared_transaction_offset_;

}

void MessageExtImpl::set_prepared_transaction_offset(int64_t preparedTransactionOffset) {

prepared_transaction_offset_ = preparedTransactionOffset;

}

std::string MessageExtImpl::toString() const {

std::stringstream ss;

ss << "MessageExt [queueId=" << queue_id_ << ", storeSize=" << store_size_ << ", queueOffset=" << queue_offset_

<< ", sysFlag=" << sys_flag_ << ", bornTimestamp=" << born_timestamp_ << ", bornHost=" << born_host_string()

<< ", storeTimestamp=" << store_timestamp_ << ", storeHost=" << store_host_string() << ", msgId=" << msg_id()

<< ", commitLogOffset=" << commit_log_offset_ << ", bodyCRC=" << body_crc_

<< ", reconsumeTimes=" << reconsume_times_ << ", preparedTransactionOffset=" << prepared_transaction_offset_

<< ", toString()=" << MessageImpl::toString() << "]";

return ss.str();

}

// ============================

// MessageClientExtImpl

// ============================

const std::string& MessageClientExtImpl::msg_id() const {

const auto& unique_id = MessageClientIDSetter::getUniqID(*this);

return unique_id.empty() ? offset_msg_id() : unique_id;

}

void MessageClientExtImpl::set_msg_id(const std::string& msgId) {

// DO NOTHING

// MessageClientIDSetter::setUniqID(*this);

}

const std::string& MessageClientExtImpl::offset_msg_id() const {

return MessageExtImpl::msg_id();

}

void MessageClientExtImpl::set_offset_msg_id(const std::string& offsetMsgId) {

return MessageExtImpl::set_msg_id(offsetMsgId);

}

} // namespace rocketmq