This is an automated email from the ASF dual-hosted git repository. mmerli pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push: new 6645341 Renamed destination to 'topic' in C++ code (#1281) 6645341 is described below commit 6645341c4bda82655d2fbbc05938bd6d77820226 Author: Matteo Merli <mme...@apache.org> AuthorDate: Sun Feb 25 20:57:54 2018 -0800 Renamed destination to 'topic' in C++ code (#1281) --- pulsar-client-cpp/include/pulsar/Message.h | 2 +- pulsar-client-cpp/include/pulsar/MessageBuilder.h | 4 +- pulsar-client-cpp/lib/BinaryProtoLookupService.cc | 63 ++++++----- pulsar-client-cpp/lib/BinaryProtoLookupService.h | 14 +-- pulsar-client-cpp/lib/ClientConnection.cc | 9 +- pulsar-client-cpp/lib/ClientConnection.h | 4 +- pulsar-client-cpp/lib/ClientImpl.cc | 51 ++++----- pulsar-client-cpp/lib/ClientImpl.h | 8 +- pulsar-client-cpp/lib/ConsumerImpl.cc | 2 +- pulsar-client-cpp/lib/HTTPLookupService.cc | 21 ++-- pulsar-client-cpp/lib/HTTPLookupService.h | 2 +- pulsar-client-cpp/lib/LookupService.h | 12 +-- pulsar-client-cpp/lib/PartitionedConsumerImpl.cc | 20 ++-- pulsar-client-cpp/lib/PartitionedConsumerImpl.h | 6 +- pulsar-client-cpp/lib/PartitionedProducerImpl.cc | 11 +- pulsar-client-cpp/lib/PartitionedProducerImpl.h | 8 +- pulsar-client-cpp/lib/ProducerImpl.cc | 2 +- .../lib/{DestinationName.cc => TopicName.cc} | 80 +++++++------- .../lib/{DestinationName.h => TopicName.h} | 24 ++--- pulsar-client-cpp/perf/PerfConsumer.cc | 2 +- pulsar-client-cpp/python/pulsar.py | 8 +- pulsar-client-cpp/tests/BasicEndToEndTest.cc | 2 +- pulsar-client-cpp/tests/BatchMessageTest.cc | 2 +- pulsar-client-cpp/tests/BinaryLookupServiceTest.cc | 8 +- pulsar-client-cpp/tests/ConsumerStatsTest.cc | 2 +- pulsar-client-cpp/tests/DestinationNameTest.cc | 117 --------------------- pulsar-client-cpp/tests/TopicNameTest.cc | 110 +++++++++++++++++++ 27 files changed, 292 insertions(+), 302 deletions(-) diff --git a/pulsar-client-cpp/include/pulsar/Message.h b/pulsar-client-cpp/include/pulsar/Message.h index 5462a68..b98b48b 100644 --- a/pulsar-client-cpp/include/pulsar/Message.h +++ b/pulsar-client-cpp/include/pulsar/Message.h @@ -107,7 +107,7 @@ class Message { /** * Get the partition key for this message - * @return key string that is hashed to determine message's destination partition + * @return key string that is hashed to determine message's topic partition */ const std::string& getPartitionKey() const; bool hasPartitionKey() const; diff --git a/pulsar-client-cpp/include/pulsar/MessageBuilder.h b/pulsar-client-cpp/include/pulsar/MessageBuilder.h index 7a08817..9738afe 100644 --- a/pulsar-client-cpp/include/pulsar/MessageBuilder.h +++ b/pulsar-client-cpp/include/pulsar/MessageBuilder.h @@ -65,7 +65,7 @@ class MessageBuilder { /* * set partition key for the message routing - * @param hash of this key is used to determine message's destination partition + * @param hash of this key is used to determine message's topic partition */ MessageBuilder& setPartitionKey(const std::string& partitionKey); @@ -95,7 +95,7 @@ class MessageBuilder { /** * override namespace replication clusters. note that it is the * caller's responsibility to provide valid cluster names, and that - * all clusters have been previously configured as destinations. + * all clusters have been previously configured as topics. * * given an empty list, the message will replicate per the namespace * configuration. diff --git a/pulsar-client-cpp/lib/BinaryProtoLookupService.cc b/pulsar-client-cpp/lib/BinaryProtoLookupService.cc index e5b8d42..c4bef30 100644 --- a/pulsar-client-cpp/lib/BinaryProtoLookupService.cc +++ b/pulsar-client-cpp/lib/BinaryProtoLookupService.cc @@ -16,11 +16,12 @@ * specific language governing permissions and limitations * under the License. */ -#include "DestinationName.h" #include "BinaryProtoLookupService.h" #include "SharedBuffer.h" #include <boost/shared_ptr.hpp> +#include <lib/TopicName.h> + #include <boost/bind.hpp> #include "ConnectionPool.h" @@ -38,20 +39,19 @@ BinaryProtoLookupService::BinaryProtoLookupService(ConnectionPool& cnxPool, cons : cnxPool_(cnxPool), serviceUrl_(lookupUrl), mutex_(), requestIdGenerator_(0) {} /* - * @param destination_name topic name to get broker for + * @param topicName topic name to get broker for * - * Looks up the owner broker for the given destination name + * Looks up the owner broker for the given topic name */ -Future<Result, LookupDataResultPtr> BinaryProtoLookupService::lookupAsync( - const std::string& destinationName) { - DestinationNamePtr dn = DestinationName::get(destinationName); - if (!dn) { - LOG_ERROR("Unable to parse destination - " << destinationName); +Future<Result, LookupDataResultPtr> BinaryProtoLookupService::lookupAsync(const std::string& topic) { + TopicNamePtr topicName = TopicName::get(topic); + if (!topicName) { + LOG_ERROR("Unable to parse topic - " << topic); LookupDataResultPromisePtr promise = boost::make_shared<LookupDataResultPromise>(); promise->setFailed(ResultInvalidTopicName); return promise->getFuture(); } - std::string lookupName = dn->toString(); + std::string lookupName = topicName->toString(); LookupDataResultPromisePtr promise = boost::make_shared<LookupDataResultPromise>(); Future<Result, ClientConnectionWeakPtr> future = cnxPool_.getConnectionAsync(serviceUrl_, serviceUrl_); future.addListener(boost::bind(&BinaryProtoLookupService::sendTopicLookupRequest, this, lookupName, false, @@ -60,24 +60,24 @@ Future<Result, LookupDataResultPtr> BinaryProtoLookupService::lookupAsync( } /* - * @param destination_name topic to get number of partitions. + * @param topicName topic to get number of partitions. * */ Future<Result, LookupDataResultPtr> BinaryProtoLookupService::getPartitionMetadataAsync( - const DestinationNamePtr& dn) { + const TopicNamePtr& topicName) { LookupDataResultPromisePtr promise = boost::make_shared<LookupDataResultPromise>(); - if (!dn) { + if (!topicName) { promise->setFailed(ResultInvalidTopicName); return promise->getFuture(); } - std::string lookupName = dn->toString(); + std::string lookupName = topicName->toString(); Future<Result, ClientConnectionWeakPtr> future = cnxPool_.getConnectionAsync(serviceUrl_, serviceUrl_); future.addListener(boost::bind(&BinaryProtoLookupService::sendPartitionMetadataLookupRequest, this, lookupName, _1, _2, promise)); return promise->getFuture(); } -void BinaryProtoLookupService::sendTopicLookupRequest(const std::string& destinationName, bool authoritative, +void BinaryProtoLookupService::sendTopicLookupRequest(const std::string& topicName, bool authoritative, Result result, const ClientConnectionWeakPtr& clientCnx, LookupDataResultPromisePtr promise) { if (result != ResultOk) { @@ -87,40 +87,37 @@ void BinaryProtoLookupService::sendTopicLookupRequest(const std::string& destina LookupDataResultPromisePtr lookupPromise = boost::make_shared<LookupDataResultPromise>(); ClientConnectionPtr conn = clientCnx.lock(); uint64_t requestId = newRequestId(); - conn->newTopicLookup(destinationName, authoritative, requestId, lookupPromise); - lookupPromise->getFuture().addListener(boost::bind(&BinaryProtoLookupService::handleLookup, this, - destinationName, _1, _2, clientCnx, promise)); + conn->newTopicLookup(topicName, authoritative, requestId, lookupPromise); + lookupPromise->getFuture().addListener( + boost::bind(&BinaryProtoLookupService::handleLookup, this, topicName, _1, _2, clientCnx, promise)); } -void BinaryProtoLookupService::handleLookup(const std::string& destinationName, Result result, +void BinaryProtoLookupService::handleLookup(const std::string& topicName, Result result, LookupDataResultPtr data, const ClientConnectionWeakPtr& clientCnx, LookupDataResultPromisePtr promise) { if (data) { if (data->isRedirect()) { - LOG_DEBUG("Lookup request is for " << destinationName << " redirected to " - << data->getBrokerUrl()); + LOG_DEBUG("Lookup request is for " << topicName << " redirected to " << data->getBrokerUrl()); const std::string& logicalAddress = data->getBrokerUrl(); const std::string& physicalAddress = data->shouldProxyThroughServiceUrl() ? serviceUrl_ : logicalAddress; Future<Result, ClientConnectionWeakPtr> future = cnxPool_.getConnectionAsync(logicalAddress, physicalAddress); - future.addListener(boost::bind(&BinaryProtoLookupService::sendTopicLookupRequest, this, - destinationName, data->isAuthoritative(), _1, _2, promise)); + future.addListener(boost::bind(&BinaryProtoLookupService::sendTopicLookupRequest, this, topicName, + data->isAuthoritative(), _1, _2, promise)); } else { - LOG_DEBUG("Lookup response for " << destinationName << ", lookup-broker-url " - << data->getBrokerUrl()); + LOG_DEBUG("Lookup response for " << topicName << ", lookup-broker-url " << data->getBrokerUrl()); promise->setValue(data); } } else { - LOG_DEBUG("Lookup failed for " << destinationName << ", result " << result); + LOG_DEBUG("Lookup failed for " << topicName << ", result " << result); promise->setFailed(result); } } -void BinaryProtoLookupService::sendPartitionMetadataLookupRequest(const std::string& destinationName, - Result result, +void BinaryProtoLookupService::sendPartitionMetadataLookupRequest(const std::string& topicName, Result result, const ClientConnectionWeakPtr& clientCnx, LookupDataResultPromisePtr promise) { if (result != ResultOk) { @@ -131,22 +128,22 @@ void BinaryProtoLookupService::sendPartitionMetadataLookupRequest(const std::str LookupDataResultPromisePtr lookupPromise = boost::make_shared<LookupDataResultPromise>(); ClientConnectionPtr conn = clientCnx.lock(); uint64_t requestId = newRequestId(); - conn->newPartitionedMetadataLookup(destinationName, requestId, lookupPromise); + conn->newPartitionedMetadataLookup(topicName, requestId, lookupPromise); lookupPromise->getFuture().addListener( - boost::bind(&BinaryProtoLookupService::handlePartitionMetadataLookup, this, destinationName, _1, _2, + boost::bind(&BinaryProtoLookupService::handlePartitionMetadataLookup, this, topicName, _1, _2, clientCnx, promise)); } -void BinaryProtoLookupService::handlePartitionMetadataLookup(const std::string& destinationName, - Result result, LookupDataResultPtr data, +void BinaryProtoLookupService::handlePartitionMetadataLookup(const std::string& topicName, Result result, + LookupDataResultPtr data, const ClientConnectionWeakPtr& clientCnx, LookupDataResultPromisePtr promise) { if (data) { - LOG_DEBUG("PartitionMetadataLookup response for " << destinationName << ", lookup-broker-url " + LOG_DEBUG("PartitionMetadataLookup response for " << topicName << ", lookup-broker-url " << data->getBrokerUrl()); promise->setValue(data); } else { - LOG_DEBUG("PartitionMetadataLookup failed for " << destinationName << ", result " << result); + LOG_DEBUG("PartitionMetadataLookup failed for " << topicName << ", result " << result); promise->setFailed(result); } } diff --git a/pulsar-client-cpp/lib/BinaryProtoLookupService.h b/pulsar-client-cpp/lib/BinaryProtoLookupService.h index 26dd9a0..f647c62 100644 --- a/pulsar-client-cpp/lib/BinaryProtoLookupService.h +++ b/pulsar-client-cpp/lib/BinaryProtoLookupService.h @@ -36,9 +36,9 @@ class BinaryProtoLookupService : public LookupService { */ BinaryProtoLookupService(ConnectionPool& cnxPool, const std::string& serviceUrl); - Future<Result, LookupDataResultPtr> lookupAsync(const std::string& destinationName); + Future<Result, LookupDataResultPtr> lookupAsync(const std::string& topicName); - Future<Result, LookupDataResultPtr> getPartitionMetadataAsync(const DestinationNamePtr& dn); + Future<Result, LookupDataResultPtr> getPartitionMetadataAsync(const TopicNamePtr& topicName); private: boost::mutex mutex_; @@ -47,18 +47,18 @@ class BinaryProtoLookupService : public LookupService { std::string serviceUrl_; ConnectionPool& cnxPool_; - void sendTopicLookupRequest(const std::string& destinationName, bool authoritative, Result result, + void sendTopicLookupRequest(const std::string& topicName, bool authoritative, Result result, const ClientConnectionWeakPtr& clientCnx, LookupDataResultPromisePtr promise); - void handleLookup(const std::string& destinationName, Result result, LookupDataResultPtr data, + void handleLookup(const std::string& topicName, Result result, LookupDataResultPtr data, const ClientConnectionWeakPtr& clientCnx, LookupDataResultPromisePtr promise); - void sendPartitionMetadataLookupRequest(const std::string& destinationName, Result result, + void sendPartitionMetadataLookupRequest(const std::string& topicName, Result result, const ClientConnectionWeakPtr& clientCnx, LookupDataResultPromisePtr promise); - void handlePartitionMetadataLookup(const std::string& destinationName, Result result, - LookupDataResultPtr data, const ClientConnectionWeakPtr& clientCnx, + void handlePartitionMetadataLookup(const std::string& topicName, Result result, LookupDataResultPtr data, + const ClientConnectionWeakPtr& clientCnx, LookupDataResultPromisePtr promise); uint64_t newRequestId(); diff --git a/pulsar-client-cpp/lib/ClientConnection.cc b/pulsar-client-cpp/lib/ClientConnection.cc index 018f4ab..fc0f611 100644 --- a/pulsar-client-cpp/lib/ClientConnection.cc +++ b/pulsar-client-cpp/lib/ClientConnection.cc @@ -953,15 +953,14 @@ Future<Result, BrokerConsumerStatsImpl> ClientConnection::newConsumerStats(uint6 return promise.getFuture(); } -void ClientConnection::newTopicLookup(const std::string& destinationName, bool authoritative, +void ClientConnection::newTopicLookup(const std::string& topicName, bool authoritative, const uint64_t requestId, LookupDataResultPromisePtr promise) { - newLookup(Commands::newLookup(destinationName, authoritative, requestId), requestId, promise); + newLookup(Commands::newLookup(topicName, authoritative, requestId), requestId, promise); } -void ClientConnection::newPartitionedMetadataLookup(const std::string& destinationName, - const uint64_t requestId, +void ClientConnection::newPartitionedMetadataLookup(const std::string& topicName, const uint64_t requestId, LookupDataResultPromisePtr promise) { - newLookup(Commands::newPartitionMetadataRequest(destinationName, requestId), requestId, promise); + newLookup(Commands::newPartitionMetadataRequest(topicName, requestId), requestId, promise); } void ClientConnection::newLookup(const SharedBuffer& cmd, const uint64_t requestId, diff --git a/pulsar-client-cpp/lib/ClientConnection.h b/pulsar-client-cpp/lib/ClientConnection.h index 47f4994..2176e27 100644 --- a/pulsar-client-cpp/lib/ClientConnection.h +++ b/pulsar-client-cpp/lib/ClientConnection.h @@ -111,10 +111,10 @@ class ClientConnection : public boost::enable_shared_from_this<ClientConnection> Future<Result, ClientConnectionWeakPtr> getCloseFuture(); - void newTopicLookup(const std::string& destinationName, bool authoritative, const uint64_t requestId, + void newTopicLookup(const std::string& topicName, bool authoritative, const uint64_t requestId, LookupDataResultPromisePtr promise); - void newPartitionedMetadataLookup(const std::string& destinationName, const uint64_t requestId, + void newPartitionedMetadataLookup(const std::string& topicName, const uint64_t requestId, LookupDataResultPromisePtr promise); void sendCommand(const SharedBuffer& cmd); diff --git a/pulsar-client-cpp/lib/ClientImpl.cc b/pulsar-client-cpp/lib/ClientImpl.cc index b760135..68d366a 100644 --- a/pulsar-client-cpp/lib/ClientImpl.cc +++ b/pulsar-client-cpp/lib/ClientImpl.cc @@ -22,7 +22,6 @@ #include "ConsumerImpl.h" #include "ProducerImpl.h" #include "ReaderImpl.h" -#include "DestinationName.h" #include "PartitionedProducerImpl.h" #include "PartitionedConsumerImpl.h" #include <boost/bind.hpp> @@ -31,6 +30,7 @@ #include <openssl/sha.h> #include "boost/date_time/posix_time/posix_time.hpp" #include <lib/HTTPLookupService.h> +#include <lib/TopicName.h> DECLARE_LOG_OBJECT() @@ -99,33 +99,33 @@ ExecutorServiceProviderPtr ClientImpl::getPartitionListenerExecutorProvider() { } void ClientImpl::createProducerAsync(const std::string& topic, ProducerConfiguration conf, CreateProducerCallback callback) { - DestinationNamePtr dn; + TopicNamePtr topicName; { Lock lock(mutex_); if (state_ != Open) { lock.unlock(); callback(ResultAlreadyClosed, Producer()); return; - } else if (!(dn = DestinationName::get(topic))) { + } else if (!(topicName = TopicName::get(topic))) { lock.unlock(); callback(ResultInvalidTopicName, Producer()); return; } } - lookupServicePtr_->getPartitionMetadataAsync(dn).addListener( - boost::bind(&ClientImpl::handleCreateProducer, shared_from_this(), _1, _2, dn, conf, callback)); + lookupServicePtr_->getPartitionMetadataAsync(topicName).addListener(boost::bind( + &ClientImpl::handleCreateProducer, shared_from_this(), _1, _2, topicName, conf, callback)); } void ClientImpl::handleCreateProducer(const Result result, const LookupDataResultPtr partitionMetadata, - DestinationNamePtr dn, ProducerConfiguration conf, + TopicNamePtr topicName, ProducerConfiguration conf, CreateProducerCallback callback) { if (!result) { ProducerImplBasePtr producer; if (partitionMetadata->getPartitions() > 1) { - producer = boost::make_shared<PartitionedProducerImpl>(shared_from_this(), dn, + producer = boost::make_shared<PartitionedProducerImpl>(shared_from_this(), topicName, partitionMetadata->getPartitions(), conf); } else { - producer = boost::make_shared<ProducerImpl>(shared_from_this(), dn->toString(), conf); + producer = boost::make_shared<ProducerImpl>(shared_from_this(), topicName->toString(), conf); } producer->getProducerCreatedFuture().addListener( boost::bind(&ClientImpl::handleProducerCreated, shared_from_this(), _1, _2, callback, producer)); @@ -135,7 +135,7 @@ void ClientImpl::handleCreateProducer(const Result result, const LookupDataResul producer->start(); } else { LOG_ERROR("Error Checking/Getting Partition Metadata while creating producer on " - << dn->toString() << " -- " << result); + << topicName->toString() << " -- " << result); callback(result, Producer()); } } @@ -147,14 +147,14 @@ void ClientImpl::handleProducerCreated(Result result, ProducerImplBaseWeakPtr pr void ClientImpl::createReaderAsync(const std::string& topic, const MessageId& startMessageId, const ReaderConfiguration& conf, ReaderCallback callback) { - DestinationNamePtr dn; + TopicNamePtr topicName; { Lock lock(mutex_); if (state_ != Open) { lock.unlock(); callback(ResultAlreadyClosed, Reader()); return; - } else if (!(dn = DestinationName::get(topic))) { + } else if (!(topicName = TopicName::get(topic))) { lock.unlock(); callback(ResultInvalidTopicName, Reader()); return; @@ -162,12 +162,13 @@ void ClientImpl::createReaderAsync(const std::string& topic, const MessageId& st } BatchMessageId msgId(startMessageId); - lookupServicePtr_->getPartitionMetadataAsync(dn).addListener(boost::bind( - &ClientImpl::handleReaderMetadataLookup, shared_from_this(), _1, _2, dn, msgId, conf, callback)); + lookupServicePtr_->getPartitionMetadataAsync(topicName).addListener( + boost::bind(&ClientImpl::handleReaderMetadataLookup, shared_from_this(), _1, _2, topicName, msgId, + conf, callback)); } void ClientImpl::handleReaderMetadataLookup(const Result result, const LookupDataResultPtr partitionMetadata, - DestinationNamePtr dn, BatchMessageId startMessageId, + TopicNamePtr topicName, BatchMessageId startMessageId, ReaderConfiguration conf, ReaderCallback callback) { if (result != ResultOk) { LOG_ERROR("Error Checking/Getting Partition Metadata while creating reader: " << result); @@ -176,12 +177,12 @@ void ClientImpl::handleReaderMetadataLookup(const Result result, const LookupDat } if (partitionMetadata->getPartitions() > 1) { - LOG_ERROR("Topic reader cannot be created on a partitioned topic: " << dn->toString()); + LOG_ERROR("Topic reader cannot be created on a partitioned topic: " << topicName->toString()); callback(ResultOperationNotSupported, Reader()); return; } - ReaderImplPtr reader = boost::make_shared<ReaderImpl>(shared_from_this(), dn->toString(), conf, + ReaderImplPtr reader = boost::make_shared<ReaderImpl>(shared_from_this(), topicName->toString(), conf, getListenerExecutorProvider()->get(), callback); reader->start(startMessageId); @@ -191,26 +192,26 @@ void ClientImpl::handleReaderMetadataLookup(const Result result, const LookupDat void ClientImpl::subscribeAsync(const std::string& topic, const std::string& consumerName, const ConsumerConfiguration& conf, SubscribeCallback callback) { - DestinationNamePtr dn; + TopicNamePtr topicName; { Lock lock(mutex_); if (state_ != Open) { lock.unlock(); callback(ResultAlreadyClosed, Consumer()); return; - } else if (!(dn = DestinationName::get(topic))) { + } else if (!(topicName = TopicName::get(topic))) { lock.unlock(); callback(ResultInvalidTopicName, Consumer()); return; } } - lookupServicePtr_->getPartitionMetadataAsync(dn).addListener(boost::bind( - &ClientImpl::handleSubscribe, shared_from_this(), _1, _2, dn, consumerName, conf, callback)); + lookupServicePtr_->getPartitionMetadataAsync(topicName).addListener(boost::bind( + &ClientImpl::handleSubscribe, shared_from_this(), _1, _2, topicName, consumerName, conf, callback)); } void ClientImpl::handleSubscribe(const Result result, const LookupDataResultPtr partitionMetadata, - DestinationNamePtr dn, const std::string& consumerName, + TopicNamePtr topicName, const std::string& consumerName, ConsumerConfiguration conf, SubscribeCallback callback) { if (result == ResultOk) { // generate random name if not supplied by the customer. @@ -224,11 +225,11 @@ void ClientImpl::handleSubscribe(const Result result, const LookupDataResultPtr callback(ResultInvalidConfiguration, Consumer()); return; } - consumer = boost::make_shared<PartitionedConsumerImpl>(shared_from_this(), consumerName, dn, - partitionMetadata->getPartitions(), conf); + consumer = boost::make_shared<PartitionedConsumerImpl>( + shared_from_this(), consumerName, topicName, partitionMetadata->getPartitions(), conf); } else { - consumer = - boost::make_shared<ConsumerImpl>(shared_from_this(), dn->toString(), consumerName, conf); + consumer = boost::make_shared<ConsumerImpl>(shared_from_this(), topicName->toString(), + consumerName, conf); } consumer->getConsumerCreatedFuture().addListener( boost::bind(&ClientImpl::handleConsumerCreated, shared_from_this(), _1, _2, callback, consumer)); diff --git a/pulsar-client-cpp/lib/ClientImpl.h b/pulsar-client-cpp/lib/ClientImpl.h index 679e2de..105df8a 100644 --- a/pulsar-client-cpp/lib/ClientImpl.h +++ b/pulsar-client-cpp/lib/ClientImpl.h @@ -24,9 +24,9 @@ #include "BinaryProtoLookupService.h" #include "ConnectionPool.h" #include "LookupDataResult.h" -#include "DestinationName.h" #include <boost/shared_ptr.hpp> #include <boost/thread/mutex.hpp> +#include <lib/TopicName.h> #include "ProducerImplBase.h" #include "ConsumerImplBase.h" @@ -83,15 +83,15 @@ class ClientImpl : public boost::enable_shared_from_this<ClientImpl> { private: void handleCreateProducer(const Result result, const LookupDataResultPtr partitionMetadata, - DestinationNamePtr dn, ProducerConfiguration conf, + TopicNamePtr topicName, ProducerConfiguration conf, CreateProducerCallback callback); void handleSubscribe(const Result result, const LookupDataResultPtr partitionMetadata, - DestinationNamePtr dn, const std::string& consumerName, ConsumerConfiguration conf, + TopicNamePtr topicName, const std::string& consumerName, ConsumerConfiguration conf, SubscribeCallback callback); void handleReaderMetadataLookup(const Result result, const LookupDataResultPtr partitionMetadata, - DestinationNamePtr dn, BatchMessageId startMessageId, + TopicNamePtr topicName, BatchMessageId startMessageId, ReaderConfiguration conf, ReaderCallback callback); void handleProducerCreated(Result result, ProducerImplBaseWeakPtr producerWeakPtr, diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc index 1c62023..55e4e6c 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.cc +++ b/pulsar-client-cpp/lib/ConsumerImpl.cc @@ -21,11 +21,11 @@ #include "Commands.h" #include "LogUtils.h" #include <boost/bind.hpp> +#include <lib/TopicName.h> #include "pulsar/Result.h" #include "pulsar/MessageId.h" #include "Utils.h" #include <exception> -#include "DestinationName.h" #include <algorithm> namespace pulsar { diff --git a/pulsar-client-cpp/lib/HTTPLookupService.cc b/pulsar-client-cpp/lib/HTTPLookupService.cc index 1a98f60..30b2af6 100644 --- a/pulsar-client-cpp/lib/HTTPLookupService.cc +++ b/pulsar-client-cpp/lib/HTTPLookupService.cc @@ -43,30 +43,31 @@ HTTPLookupService::HTTPLookupService(const std::string &lookupUrl, } } -Future<Result, LookupDataResultPtr> HTTPLookupService::lookupAsync(const std::string &destinationName) { +Future<Result, LookupDataResultPtr> HTTPLookupService::lookupAsync(const std::string &topic) { LookupPromise promise; - boost::shared_ptr<DestinationName> dn = DestinationName::get(destinationName); - if (!dn) { - LOG_ERROR("Unable to parse destination - " << destinationName); + boost::shared_ptr<TopicName> topicName = TopicName::get(topic); + if (!topicName) { + LOG_ERROR("Unable to parse topic - " << topic); promise.setFailed(ResultInvalidTopicName); return promise.getFuture(); } std::stringstream completeUrlStream; - completeUrlStream << adminUrl_ << V2_PATH << "persistent/" << dn->getProperty() << '/' << dn->getCluster() - << '/' << dn->getNamespacePortion() << '/' << dn->getEncodedLocalName(); + completeUrlStream << adminUrl_ << V2_PATH << "persistent/" << topicName->getProperty() << '/' + << topicName->getCluster() << '/' << topicName->getNamespacePortion() << '/' + << topicName->getEncodedLocalName(); executorProvider_->get()->postWork(boost::bind(&HTTPLookupService::sendHTTPRequest, shared_from_this(), promise, completeUrlStream.str(), Lookup)); return promise.getFuture(); } Future<Result, LookupDataResultPtr> HTTPLookupService::getPartitionMetadataAsync( - const DestinationNamePtr &dn) { + const TopicNamePtr &topicName) { LookupPromise promise; std::stringstream completeUrlStream; - completeUrlStream << adminUrl_ << PARTITION_PATH << dn->getProperty() << '/' << dn->getCluster() << '/' - << dn->getNamespacePortion() << '/' << dn->getEncodedLocalName() << '/' - << PARTITION_METHOD_NAME; + completeUrlStream << adminUrl_ << PARTITION_PATH << topicName->getProperty() << '/' + << topicName->getCluster() << '/' << topicName->getNamespacePortion() << '/' + << topicName->getEncodedLocalName() << '/' << PARTITION_METHOD_NAME; executorProvider_->get()->postWork(boost::bind(&HTTPLookupService::sendHTTPRequest, shared_from_this(), promise, completeUrlStream.str(), PartitionMetaData)); return promise.getFuture(); diff --git a/pulsar-client-cpp/lib/HTTPLookupService.h b/pulsar-client-cpp/lib/HTTPLookupService.h index b8c06e8..d7fff33 100644 --- a/pulsar-client-cpp/lib/HTTPLookupService.h +++ b/pulsar-client-cpp/lib/HTTPLookupService.h @@ -59,7 +59,7 @@ class HTTPLookupService : public LookupService, public boost::enable_shared_from Future<Result, LookupDataResultPtr> lookupAsync(const std::string&); - Future<Result, LookupDataResultPtr> getPartitionMetadataAsync(const DestinationNamePtr&); + Future<Result, LookupDataResultPtr> getPartitionMetadataAsync(const TopicNamePtr&); }; } // namespace pulsar diff --git a/pulsar-client-cpp/lib/LookupService.h b/pulsar-client-cpp/lib/LookupService.h index 36ba800..48d5595 100644 --- a/pulsar-client-cpp/lib/LookupService.h +++ b/pulsar-client-cpp/lib/LookupService.h @@ -22,25 +22,25 @@ #include <lib/LookupDataResult.h> #include <pulsar/Result.h> #include <lib/Future.h> -#include <lib/DestinationName.h> #include <lib/LogUtils.h> +#include <lib/TopicName.h> namespace pulsar { class LookupService { public: /* - * @param destinationName - topic name + * @param topicName - topic name * - * Looks up the owner broker for the given destination name + * Looks up the owner broker for the given topic name */ - virtual Future<Result, LookupDataResultPtr> lookupAsync(const std::string& destinationName) = 0; + virtual Future<Result, LookupDataResultPtr> lookupAsync(const std::string& topicName) = 0; /* - * @param dn - pointer to destination (topic) name + * @param topicName - pointer to topic name * * Gets Partition metadata */ - virtual Future<Result, LookupDataResultPtr> getPartitionMetadataAsync(const DestinationNamePtr& dn) = 0; + virtual Future<Result, LookupDataResultPtr> getPartitionMetadataAsync(const TopicNamePtr& topicName) = 0; virtual ~LookupService() {} }; diff --git a/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc b/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc index 3d22de9..f9c02b3 100644 --- a/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc +++ b/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc @@ -23,12 +23,12 @@ DECLARE_LOG_OBJECT() namespace pulsar { PartitionedConsumerImpl::PartitionedConsumerImpl(ClientImplPtr client, const std::string& subscriptionName, - const DestinationNamePtr destinationName, + const TopicNamePtr topicName, const unsigned int numPartitions, const ConsumerConfiguration& conf) : client_(client), subscriptionName_(subscriptionName), - destinationName_(destinationName), + topicName_(topicName), numPartitions_(numPartitions), numConsumersCreated_(0), conf_(conf), @@ -37,7 +37,7 @@ PartitionedConsumerImpl::PartitionedConsumerImpl(ClientImplPtr client, const std messages_(1000), listenerExecutor_(client->getListenerExecutorProvider()->get()), messageListener_(conf.getMessageListener()), - topic_(destinationName->toString()) { + topic_(topicName->toString()) { std::stringstream consumerStrStream; consumerStrStream << "[Partitioned Consumer: " << topic_ << "," << subscriptionName << "," << numPartitions << "]"; @@ -96,7 +96,7 @@ Result PartitionedConsumerImpl::receive(Message& msg, int timeout) { } void PartitionedConsumerImpl::unsubscribeAsync(ResultCallback callback) { - LOG_INFO("[" << destinationName_->toString() << "," << subscriptionName_ << "] Unsubscribing"); + LOG_INFO("[" << topicName_->toString() << "," << subscriptionName_ << "] Unsubscribing"); // change state to Closing, so that no Ready state operation is permitted during unsubscribe setState(Closing); // do not accept un subscribe until we have subscribe to all of the partitions of a topic @@ -108,7 +108,7 @@ void PartitionedConsumerImpl::unsubscribeAsync(ResultCallback callback) { for (ConsumerList::const_iterator consumer = consumers_.begin(); consumer != consumers_.end(); consumer++) { LOG_DEBUG("Unsubcribing Consumer - " << index << " for Subscription - " << subscriptionName_ - << " for Topic - " << destinationName_->toString()); + << " for Topic - " << topicName_->toString()); (*consumer)->unsubscribeAsync(boost::bind(&PartitionedConsumerImpl::handleUnsubscribeAsync, shared_from_this(), _1, index++, callback)); } @@ -124,7 +124,7 @@ void PartitionedConsumerImpl::handleUnsubscribeAsync(Result result, unsigned int // or do we still go ahead and check how many could we close successfully? LOG_DEBUG("handleUnsubscribeAsync callback received in Failed State for consumerIndex - " << consumerIndex << "with Result - " << result << " for Subscription - " - << subscriptionName_ << " for Topic - " << destinationName_->toString()); + << subscriptionName_ << " for Topic - " << topicName_->toString()); return; } lock.unlock(); @@ -139,7 +139,7 @@ void PartitionedConsumerImpl::handleUnsubscribeAsync(Result result, unsigned int // this means we have successfully closed this partition consumer and no unsubscribe has failed so far LOG_INFO("Successfully Unsubscribed Consumer - " << consumerIndex << " for Subscription - " << subscriptionName_ << " for Topic - " - << destinationName_->toString()); + << topicName_->toString()); unsubscribedSoFar_++; if (unsubscribedSoFar_ == numPartitions_) { LOG_DEBUG("Unsubscribed all of the partition consumer for subscription - " << subscriptionName_); @@ -178,7 +178,7 @@ void PartitionedConsumerImpl::start() { // create consumer on each partition for (unsigned int i = 0; i < numPartitions_; i++) { - std::string topicPartitionName = destinationName_->getTopicPartitionName(i); + std::string topicPartitionName = topicName_->getTopicPartitionName(i); consumer = boost::make_shared<ConsumerImpl>(client_, topicPartitionName, subscriptionName_, config, internalListenerExecutor, Partitioned); consumer->getConsumerCreatedFuture().addListener(boost::bind( @@ -218,8 +218,8 @@ void PartitionedConsumerImpl::handleSinglePartitionConsumerCreated( assert(partitionIndex < numPartitions_ && partitionIndex >= 0); numConsumersCreated_++; if (numConsumersCreated_ == numPartitions_) { - LOG_INFO("Successfully Subscribed to Partitioned Topic - " - << destinationName_->toString() << " with - " << numPartitions_ << " Partitions."); + LOG_INFO("Successfully Subscribed to Partitioned Topic - " << topicName_->toString() << " with - " + << numPartitions_ << " Partitions."); state_ = Ready; lock.unlock(); receiveMessages(); diff --git a/pulsar-client-cpp/lib/PartitionedConsumerImpl.h b/pulsar-client-cpp/lib/PartitionedConsumerImpl.h index 3d1bca1..56d04205 100644 --- a/pulsar-client-cpp/lib/PartitionedConsumerImpl.h +++ b/pulsar-client-cpp/lib/PartitionedConsumerImpl.h @@ -20,7 +20,6 @@ #define PULSAR_PARTITIONED_CONSUMER_HEADER #include "ConsumerImpl.h" #include "ClientImpl.h" -#include "DestinationName.h" #include <vector> #include <boost/shared_ptr.hpp> #include <boost/thread/mutex.hpp> @@ -29,6 +28,7 @@ #include "lib/UnAckedMessageTrackerDisabled.h" #include <lib/Latch.h> #include <lib/PartitionedBrokerConsumerStatsImpl.h> +#include <lib/TopicName.h> namespace pulsar { class PartitionedConsumerImpl; @@ -44,7 +44,7 @@ class PartitionedConsumerImpl : public ConsumerImplBase, Failed }; PartitionedConsumerImpl(ClientImplPtr client, const std::string& subscriptionName, - const DestinationNamePtr destinationName, const unsigned int numPartitions, + const TopicNamePtr topicName, const unsigned int numPartitions, const ConsumerConfiguration& conf); virtual ~PartitionedConsumerImpl(); virtual Future<Result, ConsumerImplBaseWeakPtr> getConsumerCreatedFuture(); @@ -72,7 +72,7 @@ class PartitionedConsumerImpl : public ConsumerImplBase, private: const ClientImplPtr client_; const std::string subscriptionName_; - const DestinationNamePtr destinationName_; + const TopicNamePtr topicName_; unsigned int numPartitions_; unsigned int numConsumersCreated_; const ConsumerConfiguration conf_; diff --git a/pulsar-client-cpp/lib/PartitionedProducerImpl.cc b/pulsar-client-cpp/lib/PartitionedProducerImpl.cc index 2c97ab4..728573a 100644 --- a/pulsar-client-cpp/lib/PartitionedProducerImpl.cc +++ b/pulsar-client-cpp/lib/PartitionedProducerImpl.cc @@ -20,11 +20,11 @@ #include "PartitionedProducerImpl.h" #include "LogUtils.h" #include <boost/bind.hpp> +#include <lib/TopicName.h> #include <sstream> #include "RoundRobinMessageRouter.h" #include "SinglePartitionMessageRouter.h" #include "TopicMetadataImpl.h" -#include "DestinationName.h" #include "MessageImpl.h" DECLARE_LOG_OBJECT() @@ -33,13 +33,12 @@ namespace pulsar { const std::string PartitionedProducerImpl::PARTITION_NAME_SUFFIX = "-partition-"; -PartitionedProducerImpl::PartitionedProducerImpl(ClientImplPtr client, - const DestinationNamePtr destinationName, +PartitionedProducerImpl::PartitionedProducerImpl(ClientImplPtr client, const TopicNamePtr topicName, const unsigned int numPartitions, const ProducerConfiguration& config) : client_(client), - destinationName_(destinationName), - topic_(destinationName_->toString()), + topicName_(topicName), + topic_(topicName_->toString()), conf_(config), state_(Pending), topicMetadata_(new TopicMetadataImpl(numPartitions)) { @@ -76,7 +75,7 @@ void PartitionedProducerImpl::start() { boost::shared_ptr<ProducerImpl> producer; // create producer per partition for (unsigned int i = 0; i < topicMetadata_->getNumPartitions(); i++) { - std::string topicPartitionName = destinationName_->getTopicPartitionName(i); + std::string topicPartitionName = topicName_->getTopicPartitionName(i); producer = boost::make_shared<ProducerImpl>(client_, topicPartitionName, conf_); producer->getProducerCreatedFuture().addListener(boost::bind( &PartitionedProducerImpl::handleSinglePartitionProducerCreated, shared_from_this(), _1, _2, i)); diff --git a/pulsar-client-cpp/lib/PartitionedProducerImpl.h b/pulsar-client-cpp/lib/PartitionedProducerImpl.h index b06c592..be73ddc 100644 --- a/pulsar-client-cpp/lib/PartitionedProducerImpl.h +++ b/pulsar-client-cpp/lib/PartitionedProducerImpl.h @@ -18,13 +18,13 @@ */ #include "ProducerImpl.h" #include "ClientImpl.h" -#include "DestinationName.h" #include <vector> #include <boost/shared_ptr.hpp> #include <boost/scoped_ptr.hpp> #include <boost/thread/mutex.hpp> #include <pulsar/MessageRoutingPolicy.h> #include <pulsar/TopicMetadata.h> +#include <lib/TopicName.h> namespace pulsar { @@ -43,8 +43,8 @@ class PartitionedProducerImpl : public ProducerImplBase, typedef boost::unique_lock<boost::mutex> Lock; - PartitionedProducerImpl(ClientImplPtr ptr, const DestinationNamePtr destinationName, - const unsigned int numPartitions, const ProducerConfiguration& config); + PartitionedProducerImpl(ClientImplPtr ptr, const TopicNamePtr topicName, const unsigned int numPartitions, + const ProducerConfiguration& config); virtual ~PartitionedProducerImpl(); virtual void sendAsync(const Message& msg, SendCallback callback); @@ -84,7 +84,7 @@ class PartitionedProducerImpl : public ProducerImplBase, private: const ClientImplPtr client_; - const DestinationNamePtr destinationName_; + const TopicNamePtr topicName_; const std::string topic_; boost::scoped_ptr<TopicMetadata> topicMetadata_; diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc index b2ed33f..7f09873 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.cc +++ b/pulsar-client-cpp/lib/ProducerImpl.cc @@ -21,10 +21,10 @@ #include "MessageImpl.h" #include "PulsarApi.pb.h" #include "Commands.h" -#include "DestinationName.h" #include "BatchMessageContainer.h" #include <boost/bind.hpp> #include <boost/date_time/local_time/local_time.hpp> +#include <lib/TopicName.h> namespace pulsar { DECLARE_LOG_OBJECT() diff --git a/pulsar-client-cpp/lib/DestinationName.cc b/pulsar-client-cpp/lib/TopicName.cc similarity index 60% rename from pulsar-client-cpp/lib/DestinationName.cc rename to pulsar-client-cpp/lib/TopicName.cc index 5c8eb3e..1a79a5d 100644 --- a/pulsar-client-cpp/lib/DestinationName.cc +++ b/pulsar-client-cpp/lib/TopicName.cc @@ -16,7 +16,6 @@ * specific language governing permissions and limitations * under the License. */ -#include "DestinationName.h" #include "NamedEntity.h" #include "LogUtils.h" #include "PartitionedProducerImpl.h" @@ -25,6 +24,7 @@ #include <boost/algorithm/string.hpp> #include <boost/algorithm/string/find.hpp> #include <boost/make_shared.hpp> +#include <lib/TopicName.h> #include <vector> #include <iostream> #include <sstream> @@ -36,10 +36,10 @@ namespace pulsar { typedef boost::unique_lock<boost::mutex> Lock; // static members -CURL* DestinationName::curl = NULL; -boost::mutex DestinationName::curlHandleMutex; +CURL* TopicName::curl = NULL; +boost::mutex TopicName::curlHandleMutex; -CURL* DestinationName::getCurlHandle() { +CURL* TopicName::getCurlHandle() { if (curl == NULL) { // this handle can not be shared across threads, so had to get here everytime curl = curl_easy_init(); @@ -47,30 +47,30 @@ CURL* DestinationName::getCurlHandle() { return curl; } //******************************************************************** -DestinationName::DestinationName() {} +TopicName::TopicName() {} -bool DestinationName::init(const std::string& destinationName) { - destination_ = destinationName; - if (destinationName.find("://") == std::string::npos) { - LOG_ERROR("Destination Name Invalid, domain not present - " << destinationName); +bool TopicName::init(const std::string& topicName) { + topicName_ = topicName; + if (topicName.find("://") == std::string::npos) { + LOG_ERROR("Topic name is not valid, domain not present - " << topicName); return false; } - parse(destination_, domain_, property_, cluster_, namespacePortion_, localName_); + parse(topicName_, domain_, property_, cluster_, namespacePortion_, localName_); if (localName_.empty()) { - LOG_ERROR("Destination Name is not valid, topic name is empty - " << destination_); + LOG_ERROR("Topic name is not valid, topic name is empty - " << topicName_); return false; } namespaceName_ = NamespaceName::get(property_, cluster_, namespacePortion_); return true; } -void DestinationName::parse(const std::string& destinationName, std::string& domain, std::string& property, - std::string& cluster, std::string& namespacePortion, std::string& localName) { - std::string destinationNameCopy = destinationName; - boost::replace_first(destinationNameCopy, "://", "/"); +void TopicName::parse(const std::string& topicName, std::string& domain, std::string& property, + std::string& cluster, std::string& namespacePortion, std::string& localName) { + std::string topicNameCopy = topicName; + boost::replace_first(topicNameCopy, "://", "/"); std::vector<std::string> pathTokens; - boost::algorithm::split(pathTokens, destinationNameCopy, boost::algorithm::is_any_of("/")); + boost::algorithm::split(pathTokens, topicNameCopy, boost::algorithm::is_any_of("/")); if (pathTokens.size() < 5) { - LOG_ERROR("Destination Name Invalid, does not have enough parts - " << destinationName); + LOG_ERROR("Topic name is not valid, does not have enough parts - " << topicName); return; } domain = pathTokens[0]; @@ -80,13 +80,13 @@ void DestinationName::parse(const std::string& destinationName, std::string& dom size_t slashIndex = -1; // find four '/', whatever is left is topic local name for (int i = 0; i < 4; i++) { - slashIndex = destinationNameCopy.find('/', slashIndex + 1); + slashIndex = topicNameCopy.find('/', slashIndex + 1); } // get index to next char to '/' slashIndex++; - localName = destinationNameCopy.substr(slashIndex, (destinationNameCopy.size() - slashIndex)); + localName = topicNameCopy.substr(slashIndex, (topicNameCopy.size() - slashIndex)); } -std::string DestinationName::getEncodedName(const std::string& nameBeforeEncoding) { +std::string TopicName::getEncodedName(const std::string& nameBeforeEncoding) { Lock lock(curlHandleMutex); std::string nameAfterEncoding; if (getCurlHandle()) { @@ -104,23 +104,23 @@ std::string DestinationName::getEncodedName(const std::string& nameBeforeEncodin return nameAfterEncoding; } -std::string DestinationName::getDomain() { return domain_; } +std::string TopicName::getDomain() { return domain_; } -std::string DestinationName::getProperty() { return property_; } +std::string TopicName::getProperty() { return property_; } -std::string DestinationName::getCluster() { return cluster_; } +std::string TopicName::getCluster() { return cluster_; } -std::string DestinationName::getNamespacePortion() { return namespacePortion_; } +std::string TopicName::getNamespacePortion() { return namespacePortion_; } -std::string DestinationName::getLocalName() { return localName_; } +std::string TopicName::getLocalName() { return localName_; } -std::string DestinationName::getEncodedLocalName() { return getEncodedName(localName_); } +std::string TopicName::getEncodedLocalName() { return getEncodedName(localName_); } -bool DestinationName::operator==(const DestinationName& other) { - return (this->destination_.compare(other.destination_) == 0); +bool TopicName::operator==(const TopicName& other) { + return (this->topicName_.compare(other.topicName_) == 0); } -bool DestinationName::validateDestination() { +bool TopicName::validate() { // check domain matches to "persistent", in future check "memory" when server is ready if (domain_.compare("persistent") != 0) { return false; @@ -133,22 +133,22 @@ bool DestinationName::validateDestination() { } } -boost::shared_ptr<DestinationName> DestinationName::get(const std::string& destination) { - boost::shared_ptr<DestinationName> ptr(new DestinationName()); - if (!ptr->init(destination)) { - LOG_ERROR("Destination Name Initialization failed"); - return boost::shared_ptr<DestinationName>(); +boost::shared_ptr<TopicName> TopicName::get(const std::string& topicName) { + boost::shared_ptr<TopicName> ptr(new TopicName()); + if (!ptr->init(topicName)) { + LOG_ERROR("Topic name initialization failed"); + return boost::shared_ptr<TopicName>(); } - if (ptr->validateDestination()) { + if (ptr->validate()) { return ptr; } else { - LOG_ERROR("Destination Name Validation Failed"); - return boost::shared_ptr<DestinationName>(); + LOG_ERROR("Topic name validation Failed"); + return boost::shared_ptr<TopicName>(); } } // TODO - for now return empty string if there's any error in format, later think about better error handling -std::string DestinationName::getLookupName() { +std::string TopicName::getLookupName() { std::stringstream ss; std::string seperator("/"); ss << domain_ << seperator << property_ << seperator << cluster_ << seperator << namespacePortion_ @@ -156,7 +156,7 @@ std::string DestinationName::getLookupName() { return ss.str(); } -std::string DestinationName::toString() { +std::string TopicName::toString() { std::stringstream ss; std::string seperator("/"); ss << domain_ << "://" << property_ << seperator << cluster_ << seperator << namespacePortion_ @@ -164,7 +164,7 @@ std::string DestinationName::toString() { return ss.str(); } -const std::string DestinationName::getTopicPartitionName(unsigned int partition) { +const std::string TopicName::getTopicPartitionName(unsigned int partition) { std::stringstream topicPartitionName; // make this topic name as well topicPartitionName << toString() << PartitionedProducerImpl::PARTITION_NAME_SUFFIX << partition; diff --git a/pulsar-client-cpp/lib/DestinationName.h b/pulsar-client-cpp/lib/TopicName.h similarity index 76% rename from pulsar-client-cpp/lib/DestinationName.h rename to pulsar-client-cpp/lib/TopicName.h index 251a01a..b8d610c 100644 --- a/pulsar-client-cpp/lib/DestinationName.h +++ b/pulsar-client-cpp/lib/TopicName.h @@ -16,8 +16,8 @@ * specific language governing permissions and limitations * under the License. */ -#ifndef _PULSAR_DESTINATION_NAME_HEADER_ -#define _PULSAR_DESTINATION_NAME_HEADER_ +#ifndef _PULSAR_TOPIC_NAME_HEADER_ +#define _PULSAR_TOPIC_NAME_HEADER_ #include "NamespaceName.h" #include "ServiceUnitId.h" @@ -30,9 +30,9 @@ #pragma GCC visibility push(default) namespace pulsar { -class DestinationName : public ServiceUnitId { +class TopicName : public ServiceUnitId { private: - std::string destination_; + std::string topicName_; std::string domain_; std::string property_; std::string cluster_; @@ -49,8 +49,8 @@ class DestinationName : public ServiceUnitId { std::string getLocalName(); std::string getEncodedLocalName(); std::string toString(); - static boost::shared_ptr<DestinationName> get(const std::string& destination); - bool operator==(const DestinationName& other); + static boost::shared_ptr<TopicName> get(const std::string& topicName); + bool operator==(const TopicName& other); static std::string getEncodedName(const std::string& nameBeforeEncoding); const std::string getTopicPartitionName(unsigned int partition); @@ -58,16 +58,16 @@ class DestinationName : public ServiceUnitId { static CURL* getCurlHandle(); static CURL* curl; static boost::mutex curlHandleMutex; - static void parse(const std::string& destinationName, std::string& domain, std::string& property, + static void parse(const std::string& topicName, std::string& domain, std::string& property, std::string& cluster, std::string& namespacePortion, std::string& localName); - DestinationName(); - bool validateDestination(); - bool init(const std::string& destinationName); + TopicName(); + bool validate(); + bool init(const std::string& topicName); }; -typedef boost::shared_ptr<DestinationName> DestinationNamePtr; +typedef boost::shared_ptr<TopicName> TopicNamePtr; } // namespace pulsar // end of namespace pulsar #pragma GCC visibility pop -#endif //_PULSAR_DESTINATION_NAME_HEADER_ +#endif //_PULSAR_TOPIC_NAME_HEADER_ diff --git a/pulsar-client-cpp/perf/PerfConsumer.cc b/pulsar-client-cpp/perf/PerfConsumer.cc index 97aa8db..337035e 100644 --- a/pulsar-client-cpp/perf/PerfConsumer.cc +++ b/pulsar-client-cpp/perf/PerfConsumer.cc @@ -212,7 +212,7 @@ void startPerfConsumer(const Arguments& args) { latch.wait(); LOG_INFO( - "Start receiving from " << args.numConsumers << " consumers on " << args.numTopics << " destinations"); + "Start receiving from " << args.numConsumers << " consumers on " << args.numTopics << " topics"); while (true) { std::this_thread::sleep_for(seconds(10)); diff --git a/pulsar-client-cpp/python/pulsar.py b/pulsar-client-cpp/python/pulsar.py index 060e611..ac38b25 100644 --- a/pulsar-client-cpp/python/pulsar.py +++ b/pulsar-client-cpp/python/pulsar.py @@ -553,13 +553,13 @@ class Producer: A dict of application-defined string properties. * `partition_key`: Sets the partition key for message routing. A hash of this key is used - to determine the message's destination partition. + to determine the message's topic partition. * `sequence_id`: Specify a custom sequence id for the message being published. * `replication_clusters`: Override namespace replication clusters. Note that it is the caller's responsibility to provide valid cluster names and that all clusters - have been previously configured as destinations. Given an empty list, + have been previously configured as topics. Given an empty list, the message will replicate according to the namespace configuration. * `disable_replication`: Do not replicate this message. @@ -603,12 +603,12 @@ class Producer: A dict of application0-defined string properties. * `partition_key`: Sets the partition key for the message routing. A hash of this key is - used to determine the message's destination partition. + used to determine the message's topic partition. * `sequence_id`: Specify a custom sequence id for the message being published. * `replication_clusters`: Override namespace replication clusters. Note that it is the caller's responsibility to provide valid cluster names - and that all clusters have been previously configured as destinations. + and that all clusters have been previously configured as topics. Given an empty list, the message will replicate per the namespace configuration. * `disable_replication`: diff --git a/pulsar-client-cpp/tests/BasicEndToEndTest.cc b/pulsar-client-cpp/tests/BasicEndToEndTest.cc index 0c3f2b4..dfbf44e 100644 --- a/pulsar-client-cpp/tests/BasicEndToEndTest.cc +++ b/pulsar-client-cpp/tests/BasicEndToEndTest.cc @@ -21,13 +21,13 @@ #include <boost/lexical_cast.hpp> #include <lib/LogUtils.h> #include <pulsar/MessageBuilder.h> -#include "DestinationName.h" #include <lib/Commands.h> #include <sstream> #include "boost/date_time/posix_time/posix_time.hpp" #include "CustomRoutingPolicy.h" #include <boost/thread.hpp> #include <boost/thread/mutex.hpp> +#include <lib/TopicName.h> #include "PulsarFriend.h" #include "HttpHelper.h" #include <set> diff --git a/pulsar-client-cpp/tests/BatchMessageTest.cc b/pulsar-client-cpp/tests/BatchMessageTest.cc index 7ba682f..dda8f79 100644 --- a/pulsar-client-cpp/tests/BatchMessageTest.cc +++ b/pulsar-client-cpp/tests/BatchMessageTest.cc @@ -21,8 +21,8 @@ #include <boost/lexical_cast.hpp> #include <lib/LogUtils.h> #include <pulsar/MessageBuilder.h> -#include "DestinationName.h" #include <lib/Commands.h> +#include <lib/TopicName.h> #include <sstream> #include "boost/date_time/posix_time/posix_time.hpp" #include "CustomRoutingPolicy.h" diff --git a/pulsar-client-cpp/tests/BinaryLookupServiceTest.cc b/pulsar-client-cpp/tests/BinaryLookupServiceTest.cc index b5baeb0..67e6af5 100644 --- a/pulsar-client-cpp/tests/BinaryLookupServiceTest.cc +++ b/pulsar-client-cpp/tests/BinaryLookupServiceTest.cc @@ -39,17 +39,17 @@ TEST(BinaryLookupServiceTest, basicLookup) { ConnectionPool pool_(conf, ioExecutorProvider_, authData, true); BinaryProtoLookupService lookupService(pool_, url); - std::string topic = "persistent://prop/unit/ns1/destination"; - DestinationNamePtr dn = DestinationName::get(topic); + std::string topic = "persistent://prop/unit/ns1/topic"; + TopicNamePtr topicName = TopicName::get(topic); - Future<Result, LookupDataResultPtr> partitionFuture = lookupService.getPartitionMetadataAsync(dn); + Future<Result, LookupDataResultPtr> partitionFuture = lookupService.getPartitionMetadataAsync(topicName); LookupDataResultPtr lookupData; Result result = partitionFuture.get(lookupData); ASSERT_TRUE(lookupData != NULL); ASSERT_EQ(0, lookupData->getPartitions()); Future<Result, LookupDataResultPtr> future = - lookupService.lookupAsync("persistent://prop/unit/ns1/destination"); + lookupService.lookupAsync("persistent://prop/unit/ns1/topic"); result = future.get(lookupData); ASSERT_EQ(ResultOk, result); diff --git a/pulsar-client-cpp/tests/ConsumerStatsTest.cc b/pulsar-client-cpp/tests/ConsumerStatsTest.cc index 292c47e..09c4b75 100644 --- a/pulsar-client-cpp/tests/ConsumerStatsTest.cc +++ b/pulsar-client-cpp/tests/ConsumerStatsTest.cc @@ -20,7 +20,6 @@ #include <pulsar/Client.h> #include <boost/lexical_cast.hpp> #include <lib/LogUtils.h> -#include <lib/DestinationName.h> #include <lib/Commands.h> #include "boost/date_time/posix_time/posix_time.hpp" #include "CustomRoutingPolicy.h" @@ -32,6 +31,7 @@ #include "HttpHelper.h" #include <lib/Latch.h> #include <lib/PartitionedConsumerImpl.h> +#include <lib/TopicName.h> DECLARE_LOG_OBJECT(); using namespace pulsar; diff --git a/pulsar-client-cpp/tests/DestinationNameTest.cc b/pulsar-client-cpp/tests/DestinationNameTest.cc deleted file mode 100644 index b88a1d8..0000000 --- a/pulsar-client-cpp/tests/DestinationNameTest.cc +++ /dev/null @@ -1,117 +0,0 @@ -/** - * 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 <DestinationName.h> - -#include <gtest/gtest.h> - -using namespace pulsar; - -TEST(DestinationNameTest, testLookup) { - boost::shared_ptr<DestinationName> dn = - DestinationName::get("persistent://pulsar/bf1/TESTNS.0/curveballapps"); - std::string lookup_name = dn->getLookupName(); - ASSERT_EQ(lookup_name, "persistent/pulsar/bf1/TESTNS.0/curveballapps"); -} - -TEST(DestinationNameTest, testDestinationName) { - // Compare getters and setters - boost::shared_ptr<DestinationName> dn = - DestinationName::get("persistent://property/cluster/namespace/destination"); - ASSERT_EQ("property", dn->getProperty()); - ASSERT_EQ("cluster", dn->getCluster()); - ASSERT_EQ("namespace", dn->getNamespacePortion()); - ASSERT_EQ("persistent", dn->getDomain()); - ASSERT_EQ(DestinationName::getEncodedName("destination"), dn->getLocalName()); - - // Compare == operator - boost::shared_ptr<DestinationName> dn1 = DestinationName::get("persistent://p/c/n/d"); - boost::shared_ptr<DestinationName> dn2 = DestinationName::get("persistent://p/c/n/d"); - ASSERT_TRUE(*dn1 == *dn2); -} - -TEST(DestinationNameTest, testDestinationNameWithSlashes) { - // Compare getters and setters - boost::shared_ptr<DestinationName> dn = - DestinationName::get("persistent://property/cluster/namespace/destination/name/with/slash"); - ASSERT_EQ("property", dn->getProperty()); - ASSERT_EQ("cluster", dn->getCluster()); - ASSERT_EQ("namespace", dn->getNamespacePortion()); - ASSERT_EQ("persistent", dn->getDomain()); - ASSERT_EQ("destination/name/with/slash", dn->getLocalName()); - - dn = DestinationName::get("persistent://property/cluster/namespace/destination/ends/with/slash/"); - ASSERT_TRUE(dn != NULL); - ASSERT_EQ(DestinationName::getEncodedName("destination/ends/with/slash/"), dn->getEncodedLocalName()); - - dn = DestinationName::get("persistent://property/cluster/namespace/`~!@#$%^&*()-_+=[]{}|\\;:'\"<>,./?"); - ASSERT_TRUE(dn != NULL); - ASSERT_EQ(DestinationName::getEncodedName("`~!@#$%^&*()-_+=[]{}|\\;:'\"<>,./?"), - dn->getEncodedLocalName()); - - dn = DestinationName::get("persistent://property/cluster/namespace/topic@%*)(&!%$#@#$><?"); - ASSERT_TRUE(dn != NULL); - ASSERT_EQ(DestinationName::getEncodedName("topic@%*)(&!%$#@#$><?"), dn->getEncodedLocalName()); - - dn = DestinationName::get("persistent://property/cluster/namespace/destination//with//double//slash//"); - ASSERT_TRUE(dn != NULL); - ASSERT_EQ(DestinationName::getEncodedName("destination//with//double//slash//"), - dn->getEncodedLocalName()); - - dn = DestinationName::get("persistent://property/cluster/namespace//destination/starts/with/slash/"); - ASSERT_TRUE(dn != NULL); - ASSERT_EQ(DestinationName::getEncodedName("/destination/starts/with/slash/"), dn->getEncodedLocalName()); -} -TEST(DestinationNameTest, testEmptyClusterName) { - // Compare getters and setters - boost::shared_ptr<DestinationName> dn = - DestinationName::get("persistent://property//namespace/destination"); - - ASSERT_FALSE(dn); -} - -TEST(DestinationNameTest, testExtraSlashes) { - boost::shared_ptr<DestinationName> dn = - DestinationName::get("persistent://property/cluster//namespace/destination"); - ASSERT_FALSE(dn); - dn = DestinationName::get("persistent://property//cluster//namespace//destination"); - ASSERT_FALSE(dn); -} - -TEST(DestinationNameTest, testIllegalCharacters) { - boost::shared_ptr<DestinationName> dn = - DestinationName::get("persistent://prop!!!erty/cluster&)&Name/name%%%space/destination"); - ASSERT_FALSE(dn); -} - -TEST(DestinationNameTest, testIllegalUrl) { - boost::shared_ptr<DestinationName> dn = - DestinationName::get("persistent:::/property/cluster/namespace/destination"); - ASSERT_FALSE(dn); -} - -TEST(DestinationNameTest, testEmptyString) { - boost::shared_ptr<DestinationName> dn = DestinationName::get(""); - ASSERT_FALSE(dn); -} - -TEST(DestinationNameTest, testExtraArguments) { - boost::shared_ptr<DestinationName> dn = - DestinationName::get("persistent:::/property/cluster/namespace/destination/some/extra/args"); - ASSERT_FALSE(dn); -} diff --git a/pulsar-client-cpp/tests/TopicNameTest.cc b/pulsar-client-cpp/tests/TopicNameTest.cc new file mode 100644 index 0000000..373bbef --- /dev/null +++ b/pulsar-client-cpp/tests/TopicNameTest.cc @@ -0,0 +1,110 @@ +/** + * 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 <gtest/gtest.h> +#include <lib/TopicName.h> + +using namespace pulsar; + +TEST(TopicNameTest, testLookup) { + boost::shared_ptr<TopicName> topicName = TopicName::get("persistent://pulsar/bf1/TESTNS.0/curveballapps"); + std::string lookup_name = topicName->getLookupName(); + ASSERT_EQ(lookup_name, "persistent/pulsar/bf1/TESTNS.0/curveballapps"); +} + +TEST(TopicNameTest, testTopicName) { + // Compare getters and setters + boost::shared_ptr<TopicName> topicName = TopicName::get("persistent://property/cluster/namespace/topic"); + ASSERT_EQ("property", topicName->getProperty()); + ASSERT_EQ("cluster", topicName->getCluster()); + ASSERT_EQ("namespace", topicName->getNamespacePortion()); + ASSERT_EQ("persistent", topicName->getDomain()); + ASSERT_EQ(TopicName::getEncodedName("topic"), topicName->getLocalName()); + + // Compare == operator + boost::shared_ptr<TopicName> topicName1 = TopicName::get("persistent://p/c/n/d"); + boost::shared_ptr<TopicName> topicName2 = TopicName::get("persistent://p/c/n/d"); + ASSERT_TRUE(*topicName1 == *topicName2); +} + +TEST(TopicNameTest, testTopicNameWithSlashes) { + // Compare getters and setters + boost::shared_ptr<TopicName> topicName = + TopicName::get("persistent://property/cluster/namespace/topic/name/with/slash"); + ASSERT_EQ("property", topicName->getProperty()); + ASSERT_EQ("cluster", topicName->getCluster()); + ASSERT_EQ("namespace", topicName->getNamespacePortion()); + ASSERT_EQ("persistent", topicName->getDomain()); + ASSERT_EQ("topic/name/with/slash", topicName->getLocalName()); + + topicName = TopicName::get("persistent://property/cluster/namespace/topic/ends/with/slash/"); + ASSERT_TRUE(topicName != NULL); + ASSERT_EQ(TopicName::getEncodedName("topic/ends/with/slash/"), topicName->getEncodedLocalName()); + + topicName = TopicName::get("persistent://property/cluster/namespace/`~!@#$%^&*()-_+=[]{}|\\;:'\"<>,./?"); + ASSERT_TRUE(topicName != NULL); + ASSERT_EQ(TopicName::getEncodedName("`~!@#$%^&*()-_+=[]{}|\\;:'\"<>,./?"), + topicName->getEncodedLocalName()); + + topicName = TopicName::get("persistent://property/cluster/namespace/topic@%*)(&!%$#@#$><?"); + ASSERT_TRUE(topicName != NULL); + ASSERT_EQ(TopicName::getEncodedName("topic@%*)(&!%$#@#$><?"), topicName->getEncodedLocalName()); + + topicName = TopicName::get("persistent://property/cluster/namespace/topic//with//double//slash//"); + ASSERT_TRUE(topicName != NULL); + ASSERT_EQ(TopicName::getEncodedName("topic//with//double//slash//"), topicName->getEncodedLocalName()); + + topicName = TopicName::get("persistent://property/cluster/namespace//topic/starts/with/slash/"); + ASSERT_TRUE(topicName != NULL); + ASSERT_EQ(TopicName::getEncodedName("/topic/starts/with/slash/"), topicName->getEncodedLocalName()); +} +TEST(TopicNameTest, testEmptyClusterName) { + // Compare getters and setters + boost::shared_ptr<TopicName> topicName = TopicName::get("persistent://property//namespace/topic"); + + ASSERT_FALSE(topicName); +} + +TEST(TopicNameTest, testExtraSlashes) { + boost::shared_ptr<TopicName> topicName = TopicName::get("persistent://property/cluster//namespace/topic"); + ASSERT_FALSE(topicName); + topicName = TopicName::get("persistent://property//cluster//namespace//topic"); + ASSERT_FALSE(topicName); +} + +TEST(TopicNameTest, testIllegalCharacters) { + boost::shared_ptr<TopicName> topicName = + TopicName::get("persistent://prop!!!erty/cluster&)&Name/name%%%space/topic"); + ASSERT_FALSE(topicName); +} + +TEST(TopicNameTest, testIllegalUrl) { + boost::shared_ptr<TopicName> topicName = TopicName::get("persistent:::/property/cluster/namespace/topic"); + ASSERT_FALSE(topicName); +} + +TEST(TopicNameTest, testEmptyString) { + boost::shared_ptr<TopicName> topicName = TopicName::get(""); + ASSERT_FALSE(topicName); +} + +TEST(TopicNameTest, testExtraArguments) { + boost::shared_ptr<TopicName> topicName = + TopicName::get("persistent:::/property/cluster/namespace/topic/some/extra/args"); + ASSERT_FALSE(topicName); +} -- To stop receiving notification emails like this one, please contact mme...@apache.org.