blob: 569a30ac17167afe19aa41d3a48793ba464a3945 [file] [log] [blame]
/**
* 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 "MessageImpl.h"
namespace pulsar {
MessageImpl::MessageImpl() : metadata(), payload(), messageId(), cnx_(0) {}
const Message::StringMap& MessageImpl::properties() {
if (properties_.size() == 0) {
for (int i = 0; i < metadata.properties_size(); i++) {
const std::string& key = metadata.properties(i).key();
const std::string& value = metadata.properties(i).value();
properties_.insert(std::make_pair(key, value));
}
}
return properties_;
}
const std::string& MessageImpl::getPartitionKey() const { return metadata.partition_key(); }
bool MessageImpl::hasPartitionKey() const { return metadata.has_partition_key(); }
uint64_t MessageImpl::getPublishTimestamp() const {
if (metadata.has_publish_time()) {
return metadata.publish_time();
} else {
return 0ull;
}
}
uint64_t MessageImpl::getEventTimestamp() const {
if (metadata.has_event_time()) {
return metadata.event_time();
} else {
return 0ull;
}
}
void MessageImpl::setReplicationClusters(const std::vector<std::string>& clusters) {
google::protobuf::RepeatedPtrField<std::string> r(clusters.begin(), clusters.end());
r.Swap(metadata.mutable_replicate_to());
}
void MessageImpl::disableReplication(bool flag) {
google::protobuf::RepeatedPtrField<std::string> r;
if (flag) {
r.AddAllocated(new std::string("__local__"));
}
r.Swap(metadata.mutable_replicate_to());
}
void MessageImpl::setProperty(const std::string& name, const std::string& value) {
proto::KeyValue* keyValue = proto::KeyValue().New();
keyValue->set_key(name);
keyValue->set_value(value);
metadata.mutable_properties()->AddAllocated(keyValue);
}
void MessageImpl::setPartitionKey(const std::string& partitionKey) {
metadata.set_partition_key(partitionKey);
}
void MessageImpl::setEventTimestamp(uint64_t eventTimestamp) { metadata.set_event_time(eventTimestamp); }
} // namespace pulsar