| /** |
| * Copyright 2016 Yahoo Inc. |
| * |
| * Licensed 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 "ClientConnection.h" |
| |
| #include "PulsarApi.pb.h" |
| |
| #include <boost/shared_ptr.hpp> |
| #include <boost/array.hpp> |
| #include <iostream> |
| #include <algorithm> |
| #include <boost/date_time/posix_time/posix_time.hpp> |
| #include <boost/date_time/gregorian/gregorian.hpp> |
| #include <boost/filesystem.hpp> |
| #include <climits> |
| |
| #include "ExecutorService.h" |
| #include "Commands.h" |
| #include "LogUtils.h" |
| #include "Url.h" |
| |
| #include <boost/bind.hpp> |
| |
| #include <string> |
| |
| #include "ProducerImpl.h" |
| #include "ConsumerImpl.h" |
| #include "checksum/ChecksumProvider.h" |
| |
| DECLARE_LOG_OBJECT() |
| |
| using namespace pulsar::proto; |
| using namespace boost::asio::ip; |
| |
| namespace pulsar { |
| |
| static const uint32_t DefaultBufferSize = 64 * 1024; |
| |
| static const int KeepAliveIntervalInSeconds = 30; |
| |
| // Convert error codes from protobuf to client API Result |
| static Result getResult(ServerError serverError) { |
| switch (serverError) { |
| case UnknownError: |
| return ResultUnknownError; |
| |
| case MetadataError: |
| return ResultBrokerMetadataError; |
| |
| case ChecksumError: |
| return ResultChecksumError; |
| |
| case PersistenceError: |
| return ResultBrokerPersistenceError; |
| |
| case AuthenticationError: |
| return ResultAuthenticationError; |
| |
| case AuthorizationError: |
| return ResultAuthorizationError; |
| |
| case ConsumerBusy: |
| return ResultConsumerBusy; |
| |
| case ServiceNotReady: |
| return ResultServiceUnitNotReady; |
| |
| case ProducerBlockedQuotaExceededError: |
| return ResultProducerBlockedQuotaExceededError; |
| |
| case ProducerBlockedQuotaExceededException: |
| return ResultProducerBlockedQuotaExceededException; |
| |
| case TopicNotFound: |
| return ResultTopicNotFound; |
| |
| case SubscriptionNotFound: |
| return ResultSubscriptionNotFound; |
| |
| case ConsumerNotFound: |
| return ResultConsumerNotFound; |
| |
| case UnsupportedVersionError: |
| return ResultUnsupportedVersionError; |
| } |
| // NOTE : Do not add default case in the switch above. In future if we get new cases for |
| // ServerError and miss them in the switch above we would like to get notified. Adding |
| // return here to make the compiler happy. |
| return ResultUnknownError; |
| } |
| |
| ClientConnection::ClientConnection(const std::string& endpoint, ExecutorServicePtr executor, |
| const ClientConfiguration& clientConfiguration, const AuthenticationPtr& authentication) |
| : state_(Pending), |
| operationsTimeout_(seconds(clientConfiguration.getOperationTimeoutSeconds())), |
| authentication_(authentication), |
| serverProtocolVersion_(ProtocolVersion_MIN), |
| executor_(executor), |
| resolver_(executor->createTcpResolver()), |
| socket_(executor->createSocket()), |
| address_(endpoint), |
| cnxString_("[<none> -> " + endpoint + "] "), |
| error_(boost::system::error_code()), |
| incomingBuffer_(SharedBuffer::allocate(DefaultBufferSize)), |
| incomingCmd_(), |
| pendingWriteBuffers_(), |
| pendingWriteOperations_(0), |
| outgoingBuffer_(SharedBuffer::allocate(DefaultBufferSize)), |
| outgoingCmd_(), |
| havePendingPingRequest_(false), |
| keepAliveTimer_(), |
| maxPendingLookupRequest_(clientConfiguration.getConcurrentLookupRequest()), |
| consumerStatsRequestTimer_(executor_->createDeadlineTimer()), |
| consumerStatsTTLMs_(30 * 1000), |
| numOfPendingLookupRequest_(0), |
| isTlsAllowInsecureConnection_(false) { |
| if (clientConfiguration.isUseTls()) { |
| using namespace boost::filesystem; |
| |
| #if BOOST_VERSION >= 105400 |
| boost::asio::ssl::context ctx(executor_->io_service_, boost::asio::ssl::context::tlsv12_client); |
| #else |
| boost::asio::ssl::context ctx(executor_->io_service_, boost::asio::ssl::context::tlsv1_client); |
| #endif |
| if (clientConfiguration.isTlsAllowInsecureConnection()) { |
| ctx.set_verify_mode(boost::asio::ssl::context::verify_none); |
| isTlsAllowInsecureConnection_ = true; |
| } else { |
| ctx.set_verify_mode(boost::asio::ssl::context::verify_peer); |
| std::string trustCertFilePath = clientConfiguration.getTlsTrustCertsFilePath(); |
| if (exists(path(trustCertFilePath))) { |
| ctx.load_verify_file(trustCertFilePath); |
| } else { |
| LOG_ERROR(trustCertFilePath << ": No such trustCertFile"); |
| close(); |
| return; |
| } |
| } |
| |
| AuthenticationDataPtr authData; |
| if (authentication_->getAuthData(authData) == ResultOk && authData->hasDataForTls()){ |
| std::string tlsCertificates = authData->getTlsCertificates(); |
| std::string tlsPrivateKey = authData->getTlsPrivateKey(); |
| |
| if (exists(path(tlsCertificates))) { |
| ctx.use_certificate_file(tlsCertificates, boost::asio::ssl::context::pem); |
| } else { |
| LOG_ERROR(tlsCertificates << ": No such tlsCertificates"); |
| close(); |
| return; |
| } |
| |
| if (exists(path(tlsPrivateKey))) { |
| ctx.use_private_key_file(tlsPrivateKey, boost::asio::ssl::context::pem); |
| } else { |
| LOG_ERROR(tlsPrivateKey << ": No such tlsPrivateKey"); |
| close(); |
| return; |
| } |
| } |
| |
| tlsSocket_ = executor->createTlsSocket(socket_, ctx); |
| } |
| } |
| |
| ClientConnection::~ClientConnection() { |
| LOG_INFO(cnxString_ << "Destroyed connection"); |
| } |
| |
| void ClientConnection::handlePulsarConnected(const CommandConnected& cmdConnected) { |
| if (!cmdConnected.has_server_version()) { |
| LOG_ERROR(cnxString_ << "Server version is not set"); |
| close(); |
| return; |
| } |
| |
| state_ = Ready; |
| serverProtocolVersion_ = cmdConnected.protocol_version(); |
| connectPromise_.setValue(shared_from_this()); |
| |
| if (serverProtocolVersion_ >= v1) { |
| // Only send keep-alive probes if the broker supports it |
| keepAliveTimer_ = executor_->createDeadlineTimer(); |
| keepAliveTimer_->expires_from_now(boost::posix_time::seconds(KeepAliveIntervalInSeconds)); |
| keepAliveTimer_->async_wait( |
| boost::bind(&ClientConnection::handleKeepAliveTimeout, shared_from_this())); |
| } |
| |
| if (serverProtocolVersion_ >= v7) { |
| startConsumerStatsTimer(std::vector<uint64_t>()); |
| } |
| } |
| |
| void ClientConnection::startConsumerStatsTimer(std::vector<uint64_t> consumerStatsRequests) { |
| std::vector<Promise<Result, BrokerConsumerStats> > consumerStatsPromises; |
| Lock lock(mutex_); |
| |
| for (int i = 0; i < consumerStatsRequests.size(); i++) { |
| PendingConsumerStatsMap::iterator it = pendingConsumerStatsMap_.find(consumerStatsRequests[i]); |
| if (it != pendingConsumerStatsMap_.end()) { |
| LOG_DEBUG(cnxString_ << " removing request_id " << it->first << " from the pendingConsumerStatsMap_"); |
| consumerStatsPromises.push_back(it->second); |
| pendingConsumerStatsMap_.erase(it); |
| } else { |
| LOG_DEBUG(cnxString_ << "request_id " << it->first << " already fulfilled - not removing it"); |
| } |
| } |
| |
| consumerStatsRequests.clear(); |
| for (PendingConsumerStatsMap::iterator it = pendingConsumerStatsMap_.begin(); |
| it != pendingConsumerStatsMap_.end(); ++it) { |
| consumerStatsRequests.push_back(it->first); |
| } |
| consumerStatsRequestTimer_->expires_from_now(operationsTimeout_); |
| consumerStatsRequestTimer_->async_wait( |
| boost::bind(&ClientConnection::handleConsumerStatsTimeout, shared_from_this(), |
| boost::asio::placeholders::error, consumerStatsRequests)); |
| lock.unlock(); |
| // Complex logic since promises need to be fulfilled outside the lock |
| for (int i = 0; i < consumerStatsPromises.size(); i++) { |
| consumerStatsPromises[i].setFailed(ResultTimeout); |
| LOG_WARN(cnxString_ << " Operation timedout, didn't get response from broker"); |
| } |
| } |
| |
| /// The number of unacknowledged probes to send before considering the connection dead and notifying the application layer |
| typedef boost::asio::detail::socket_option::integer<IPPROTO_TCP, TCP_KEEPCNT> tcp_keep_alive_count; |
| |
| /// The interval between subsequential keepalive probes, regardless of what the connection has exchanged in the meantime |
| typedef boost::asio::detail::socket_option::integer<IPPROTO_TCP, TCP_KEEPINTVL> tcp_keep_alive_interval; |
| |
| /// The interval between the last data packet sent (simple ACKs are not considered data) and the first keepalive |
| /// probe; after the connection is marked to need keepalive, this counter is not used any further |
| #ifdef __APPLE__ |
| typedef boost::asio::detail::socket_option::integer<IPPROTO_TCP, TCP_KEEPALIVE> tcp_keep_alive_idle; |
| #else |
| typedef boost::asio::detail::socket_option::integer<IPPROTO_TCP, TCP_KEEPIDLE> tcp_keep_alive_idle; |
| #endif |
| |
| /* |
| * TCP Connect handler |
| * |
| * if async_connect without any error, connected_ would be set to true |
| * at this point the connection is deemed valid to be used by clients of this class |
| */ |
| void ClientConnection::handleTcpConnected(const boost::system::error_code& err, |
| tcp::resolver::iterator endpointIterator) { |
| if (!err) { |
| std::stringstream cnxStringStream; |
| cnxStringStream << "[" << socket_->local_endpoint() << " -> " << socket_->remote_endpoint() << "] "; |
| cnxString_ = cnxStringStream.str(); |
| |
| LOG_INFO(cnxString_ << "Connected to broker"); |
| state_ = TcpConnected; |
| socket_->set_option(tcp::no_delay(true)); |
| |
| socket_->set_option(tcp::socket::keep_alive(true)); |
| |
| // Start TCP keep-alive probes after connection has been idle after 1 minute. Ideally this |
| // should never happen, given that we're sending our own keep-alive probes (within the TCP |
| // connection) every 30 seconds |
| socket_->set_option(tcp_keep_alive_idle(1 * 60)); |
| |
| // Send up to 10 probes before declaring the connection broken |
| socket_->set_option(tcp_keep_alive_count(10)); |
| |
| // Interval between probes: 6 seconds |
| socket_->set_option(tcp_keep_alive_interval(6)); |
| |
| if (tlsSocket_) { |
| if (!isTlsAllowInsecureConnection_) { |
| boost::system::error_code err; |
| Url service_url; |
| if (!Url::parse(address_, service_url)) { |
| LOG_ERROR(cnxString_ << "Invalid Url, unable to parse: " << err << " " << err.message()); |
| close(); |
| return; |
| } |
| } |
| tlsSocket_->async_handshake(boost::asio::ssl::stream<tcp::socket>::client, boost::bind(&ClientConnection::handleHandshake, shared_from_this(), boost::asio::placeholders::error)); |
| } else { |
| handleHandshake(boost::system::errc::make_error_code(boost::system::errc::success)); |
| } |
| } else if (endpointIterator != tcp::resolver::iterator()) { |
| // The connection failed. Try the next endpoint in the list. |
| socket_->close(); |
| tcp::endpoint endpoint = *endpointIterator; |
| socket_->async_connect( |
| endpoint, |
| boost::bind(&ClientConnection::handleTcpConnected, shared_from_this(), |
| boost::asio::placeholders::error, ++endpointIterator)); |
| } else { |
| LOG_ERROR(cnxString_ << "Failed to establish connection: " << err.message()); |
| close(); |
| return; |
| |
| } |
| |
| } |
| |
| void ClientConnection::handleHandshake(const boost::system::error_code& err) { |
| SharedBuffer buffer = Commands::newConnect(authentication_); |
| // Send CONNECT command to broker |
| asyncWrite( |
| buffer.const_asio_buffer(), |
| boost::bind(&ClientConnection::handleSentPulsarConnect, shared_from_this(), |
| boost::asio::placeholders::error, buffer)); |
| } |
| |
| void ClientConnection::handleSentPulsarConnect(const boost::system::error_code& err, |
| const SharedBuffer& buffer) { |
| if (err) { |
| LOG_ERROR(cnxString_ << "Failed to establish connection: " << err.message()); |
| close(); |
| return; |
| } |
| |
| // Schedule the reading of CONNECTED command from broker |
| readNextCommand(); |
| } |
| |
| /* |
| * Async method to establish TCP connection with broker |
| * |
| * tcpConnectCompletionHandler is notified when the result of this call is available. |
| * |
| */ |
| void ClientConnection::tcpConnectAsync() { |
| boost::system::error_code err; |
| Url service_url; |
| if (!Url::parse(address_, service_url)) { |
| LOG_ERROR(cnxString_ << "Invalid Url, unable to parse: " << err << " " << err.message()); |
| close(); |
| return; |
| } |
| |
| if (service_url.protocol() != "pulsar" && service_url.protocol() != "pulsar+ssl") { |
| LOG_ERROR(cnxString_ << "Invalid Url protocol '" << service_url.protocol() << "'. Valid values are 'pulsar' and 'pulsar+ssl'"); |
| close(); |
| return; |
| } |
| |
| LOG_DEBUG(cnxString_ << "Connecting to " << service_url.host() << ":" << service_url.port()); |
| tcp::resolver::query query(service_url.host(), |
| boost::lexical_cast<std::string>(service_url.port())); |
| resolver_->async_resolve( |
| query, |
| boost::bind(&ClientConnection::handleResolve, shared_from_this(), |
| boost::asio::placeholders::error, boost::asio::placeholders::iterator)); |
| } |
| |
| void ClientConnection::handleResolve(const boost::system::error_code& err, |
| tcp::resolver::iterator endpointIterator) { |
| if (err) { |
| LOG_ERROR(cnxString_ << "Resolve error: " << err << " : " << err.message()); |
| close(); |
| return; |
| } |
| |
| if (endpointIterator != tcp::resolver::iterator()) { |
| LOG_DEBUG(cnxString_ << "Resolved hostname " << endpointIterator->host_name() // |
| << " to " << endpointIterator->endpoint()); |
| socket_->async_connect( |
| *endpointIterator++, |
| boost::bind(&ClientConnection::handleTcpConnected, shared_from_this(), |
| boost::asio::placeholders::error, endpointIterator)); |
| } else { |
| LOG_WARN(cnxString_ << "No IP address found"); |
| close(); |
| return; |
| } |
| } |
| |
| void ClientConnection::readNextCommand() { |
| const static uint32_t minReadSize = sizeof(uint32_t); |
| asyncReceive( |
| incomingBuffer_.asio_buffer(), |
| customAllocReadHandler( |
| boost::bind(&ClientConnection::handleRead, shared_from_this(), _1, _2, |
| minReadSize))); |
| } |
| |
| void ClientConnection::handleRead(const boost::system::error_code& err, size_t bytesTransferred, |
| uint32_t minReadSize) { |
| // Update buffer write idx with new data |
| incomingBuffer_.bytesWritten(bytesTransferred); |
| |
| if (err || bytesTransferred == 0) { |
| close(); |
| } else if (bytesTransferred < minReadSize) { |
| // Read the remaining part, use a slice of buffer to write on the next |
| // region |
| SharedBuffer buffer = incomingBuffer_.slice(bytesTransferred); |
| asyncReceive( |
| buffer.asio_buffer(), |
| customAllocReadHandler( |
| boost::bind(&ClientConnection::handleRead, shared_from_this(), _1, _2, |
| minReadSize - bytesTransferred))); |
| } else { |
| processIncomingBuffer(); |
| } |
| } |
| |
| void ClientConnection::processIncomingBuffer() { |
| // Process all the available frames from the incoming buffer |
| while (incomingBuffer_.readableBytes() >= sizeof(uint32_t)) { |
| |
| // Extract message frames from incoming buffer |
| // At this point we have at least 4 bytes in the buffer |
| uint32_t frameSize = incomingBuffer_.readUnsignedInt(); |
| |
| if (frameSize > incomingBuffer_.readableBytes()) { |
| // We don't have the entire frame yet |
| const uint32_t bytesToReceive = frameSize - incomingBuffer_.readableBytes(); |
| |
| // Rollback the reading of frameSize (when the frame will be complete, |
| // we'll read it again |
| incomingBuffer_.rollback(sizeof(uint32_t)); |
| |
| if (bytesToReceive <= incomingBuffer_.writableBytes()) { |
| // The rest of the frame still fits in the current buffer |
| asyncReceive( |
| incomingBuffer_.asio_buffer(), |
| customAllocReadHandler( |
| boost::bind(&ClientConnection::handleRead, shared_from_this(), _1, |
| _2, bytesToReceive))); |
| return; |
| } else { |
| // Need to allocate a buffer big enough for the frame |
| uint32_t newBufferSize = std::max<uint32_t>(DefaultBufferSize, |
| frameSize + sizeof(uint32_t)); |
| incomingBuffer_ = SharedBuffer::copyFrom(incomingBuffer_, newBufferSize); |
| |
| asyncReceive( |
| incomingBuffer_.asio_buffer(), |
| customAllocReadHandler( |
| boost::bind(&ClientConnection::handleRead, shared_from_this(), _1, |
| _2, bytesToReceive))); |
| return; |
| } |
| } |
| |
| // At this point, we have at least one complete frame available in the buffer |
| uint32_t cmdSize = incomingBuffer_.readUnsignedInt(); |
| if (!incomingCmd_.ParseFromArray(incomingBuffer_.data(), cmdSize)) { |
| LOG_ERROR(cnxString_ << "Error parsing protocol buffer command"); |
| close(); |
| return; |
| } |
| |
| incomingBuffer_.consume(cmdSize); |
| |
| if (incomingCmd_.type() == BaseCommand::MESSAGE) { |
| // Parse message metadata and extract payload |
| MessageMetadata msgMetadata; |
| |
| //read checksum |
| bool isChecksumValid = verifyChecksum(incomingBuffer_, incomingCmd_); |
| |
| uint32_t metadataSize = incomingBuffer_.readUnsignedInt(); |
| if (!msgMetadata.ParseFromArray(incomingBuffer_.data(), metadataSize)) { |
| LOG_ERROR(cnxString_ << "[consumer id " << incomingCmd_.message().consumer_id() // |
| << ", message ledger id " << incomingCmd_.message().message_id().ledgerid() // |
| << ", entry id " << incomingCmd_.message().message_id().entryid() << "] Error parsing message metadata"); |
| close(); |
| return; |
| } |
| |
| incomingBuffer_.consume(metadataSize); |
| |
| uint32_t payloadSize = frameSize - (cmdSize + 4) - (metadataSize + 4); |
| SharedBuffer payload = SharedBuffer::copy(incomingBuffer_.data(), payloadSize); |
| incomingBuffer_.consume(payloadSize); |
| handleIncomingMessage(incomingCmd_.message(), isChecksumValid, msgMetadata, payload); |
| } else { |
| handleIncomingCommand(); |
| } |
| } |
| |
| if (incomingBuffer_.readableBytes() > 0) { |
| // We still have 1 to 3 bytes from the next frame |
| assert(incomingBuffer_.readableBytes() < sizeof(uint32_t)); |
| |
| // Restart with a new buffer and copy the the few bytes at the beginning |
| incomingBuffer_ = SharedBuffer::copyFrom(incomingBuffer_, DefaultBufferSize); |
| |
| // At least we need to read 4 bytes to have the complete frame size |
| uint32_t minReadSize = sizeof(uint32_t) - incomingBuffer_.readableBytes(); |
| |
| asyncReceive( |
| incomingBuffer_.asio_buffer(), |
| customAllocReadHandler( |
| boost::bind(&ClientConnection::handleRead, shared_from_this(), _1, _2, |
| minReadSize))); |
| return; |
| } |
| |
| // We have read everything we had in the buffer |
| // Rollback the indexes to reuse the same buffer |
| incomingBuffer_.reset(); |
| |
| readNextCommand(); |
| } |
| |
| bool ClientConnection::verifyChecksum(SharedBuffer& incomingBuffer_, proto::BaseCommand& incomingCmd_) { |
| int readerIndex = incomingBuffer_.readerIndex(); |
| bool isChecksumValid = true; |
| if (incomingBuffer_.readUnsignedShort() == Commands::magicCrc32c) { |
| uint32_t storedChecksum = incomingBuffer_.readUnsignedInt(); |
| // compute metadata-payload checksum |
| int metadataPayloadSize = incomingBuffer_.readableBytes(); |
| uint32_t computedChecksum = computeChecksum(0, incomingBuffer_.data(), metadataPayloadSize); |
| // verify checksum |
| isChecksumValid = (storedChecksum == computedChecksum); |
| |
| if (!isChecksumValid) { |
| LOG_ERROR("[consumer id " << incomingCmd_.message().consumer_id() // |
| << ", message ledger id " << incomingCmd_.message().message_id().ledgerid()// |
| << ", entry id " << incomingCmd_.message().message_id().entryid()// |
| << "stored-checksum" << storedChecksum << "computedChecksum" << computedChecksum// |
| << "] Checksum verification failed"); |
| } |
| } else { |
| incomingBuffer_.setReaderIndex(readerIndex); |
| } |
| return isChecksumValid; |
| } |
| |
| void ClientConnection::handleIncomingMessage(const proto::CommandMessage& msg, bool isChecksumValid, |
| proto::MessageMetadata& msgMetadata, |
| SharedBuffer& payload) { |
| LOG_DEBUG(cnxString_ << "Received a message from the server for consumer: " << msg.consumer_id()); |
| |
| Lock lock(mutex_); |
| ConsumersMap::iterator it = consumers_.find(msg.consumer_id()); |
| if (it != consumers_.end()) { |
| ConsumerImplPtr consumer = it->second.lock(); |
| |
| if (consumer) { |
| // Unlock the mutex before notifying the consumer of the |
| // new received message |
| lock.unlock(); |
| consumer->messageReceived(shared_from_this(), msg, isChecksumValid, msgMetadata, payload); |
| } else { |
| consumers_.erase(msg.consumer_id()); |
| LOG_DEBUG( |
| cnxString_ << "Ignoring incoming message for already destroyed consumer " << msg.consumer_id()); |
| } |
| } else { |
| LOG_DEBUG(cnxString_ << "Got invalid consumer Id in " // |
| << msg.consumer_id() << " -- msg: "<< msgMetadata.sequence_id()); |
| } |
| } |
| |
| void ClientConnection::handleIncomingCommand() { |
| LOG_DEBUG( |
| cnxString_ << "Handling incoming command: " << Commands::messageType(incomingCmd_.type())); |
| |
| switch (state_) { |
| case Pending: { |
| LOG_ERROR(cnxString_ << "Connection is not ready yet"); |
| break; |
| } |
| |
| case TcpConnected: { |
| // Handle Pulsar Connected |
| if (incomingCmd_.type() != BaseCommand::CONNECTED) { |
| // Wrong cmd |
| close(); |
| } else { |
| handlePulsarConnected(incomingCmd_.connected()); |
| } |
| break; |
| } |
| |
| case Disconnected: { |
| LOG_ERROR(cnxString_ << "Connection already disconnected"); |
| break; |
| } |
| |
| case Ready: { |
| // Handle normal commands |
| switch (incomingCmd_.type()) { |
| case BaseCommand::SEND_RECEIPT: { |
| const CommandSendReceipt& sendReceipt = incomingCmd_.send_receipt(); |
| int producerId = sendReceipt.producer_id(); |
| uint64_t sequenceId = sendReceipt.sequence_id(); |
| |
| LOG_DEBUG( |
| cnxString_ << "Got receipt for producer: " << producerId << " -- msg: "<< sequenceId); |
| |
| Lock lock(mutex_); |
| ProducersMap::iterator it = producers_.find(producerId); |
| if (it != producers_.end()) { |
| ProducerImplPtr producer = it->second.lock(); |
| lock.unlock(); |
| |
| if (producer) { |
| if (!producer->ackReceived(sequenceId)) { |
| // If the producer fails to process the ack, we need to close the connection to give it a chance to recover from there |
| close(); |
| } |
| } |
| } else { |
| LOG_ERROR(cnxString_ << "Got invalid producer Id in SendReceipt: " // |
| << producerId << " -- msg: "<< sequenceId); |
| } |
| |
| break; |
| } |
| |
| case BaseCommand::SEND_ERROR: { |
| const CommandSendError& error = incomingCmd_.send_error(); |
| LOG_WARN(cnxString_ << "Received send error from server: " << error.message()); |
| if (ChecksumError == error.error()) { |
| long producerId = error.producer_id(); |
| long sequenceId = error.sequence_id(); |
| Lock lock(mutex_); |
| ProducersMap::iterator it = producers_.find(producerId); |
| if (it != producers_.end()) { |
| ProducerImplPtr producer = it->second.lock(); |
| lock.unlock(); |
| |
| if (producer) { |
| if (!producer->removeCorruptMessage(sequenceId)) { |
| // If the producer fails to remove corrupt msg, we need to close the connection to give it a chance to recover from there |
| close(); |
| } |
| } |
| } |
| } else { |
| close(); |
| } |
| break; |
| } |
| |
| case BaseCommand::SUCCESS: { |
| const CommandSuccess& success = incomingCmd_.success(); |
| LOG_DEBUG( |
| cnxString_ << "Received success response from server. req_id: " << success.request_id()); |
| |
| Lock lock(mutex_); |
| PendingRequestsMap::iterator it = pendingRequests_.find(success.request_id()); |
| if (it != pendingRequests_.end()) { |
| PendingRequestData requestData = it->second; |
| pendingRequests_.erase(it); |
| lock.unlock(); |
| |
| requestData.promise.setValue(""); |
| requestData.timer->cancel(); |
| } |
| break; |
| } |
| |
| case BaseCommand::PARTITIONED_METADATA_RESPONSE: { |
| const CommandPartitionedTopicMetadataResponse& partitionMetadataResponse = |
| incomingCmd_.partitionmetadataresponse(); |
| LOG_DEBUG( |
| cnxString_ << "Received partition-metadata response from server. req_id: " << partitionMetadataResponse.request_id()); |
| |
| Lock lock(mutex_); |
| PendingLookupRequestsMap::iterator it = pendingLookupRequests_.find( |
| partitionMetadataResponse.request_id()); |
| if (it != pendingLookupRequests_.end()) { |
| LookupDataResultPromisePtr lookupDataPromise = it->second; |
| pendingLookupRequests_.erase(it); |
| numOfPendingLookupRequest_--; |
| lock.unlock(); |
| |
| if (!partitionMetadataResponse.has_response() |
| || (partitionMetadataResponse.response() |
| == CommandPartitionedTopicMetadataResponse::Failed)) { |
| if (partitionMetadataResponse.has_error()) { |
| LOG_ERROR( |
| cnxString_ << "Failed partition-metadata lookup req_id: " << partitionMetadataResponse.request_id() << " error: " << partitionMetadataResponse.error()); |
| } else { |
| LOG_ERROR( |
| cnxString_ << "Failed partition-metadata lookup req_id: " << partitionMetadataResponse.request_id() << " with empty response: "); |
| } |
| lookupDataPromise->setFailed(ResultConnectError); |
| } else { |
| LookupDataResultPtr lookupResultPtr = boost::make_shared< |
| LookupDataResult>(); |
| lookupResultPtr->setPartitions(partitionMetadataResponse.partitions()); |
| lookupDataPromise->setValue(lookupResultPtr); |
| } |
| |
| } else { |
| LOG_WARN( |
| "Received unknown request id from server: " << partitionMetadataResponse.request_id()); |
| } |
| break; |
| } |
| |
| case BaseCommand::CONSUMER_STATS_RESPONSE: { |
| const CommandConsumerStatsResponse& consumerStatsResponse = incomingCmd_.consumerstatsresponse(); |
| LOG_DEBUG( |
| cnxString_ |
| << "ConsumerStatsResponse command - Received consumer stats response from server. req_id: " |
| << consumerStatsResponse.request_id()); |
| Lock lock(mutex_); |
| PendingConsumerStatsMap::iterator it = pendingConsumerStatsMap_.find( |
| consumerStatsResponse.request_id()); |
| if (it != pendingConsumerStatsMap_.end()) { |
| Promise<Result, BrokerConsumerStats> consumerStatsPromise = it->second; |
| pendingConsumerStatsMap_.erase(it); |
| lock.unlock(); |
| |
| if (consumerStatsResponse.has_error_code()) { |
| if (consumerStatsResponse.has_error_message()) { |
| LOG_ERROR(cnxString_ << " Failed to get consumer stats - " |
| << consumerStatsResponse.error_message()); |
| } |
| consumerStatsPromise.setFailed(getResult(consumerStatsResponse.error_code())); |
| } else { |
| LOG_DEBUG( |
| cnxString_ |
| << "ConsumerStatsResponse command - Received consumer stats response from server. req_id: " |
| << consumerStatsResponse.request_id() << " Stats: "); |
| boost::posix_time::ptime validTill = now() + milliseconds(consumerStatsTTLMs_); |
| BrokerConsumerStats brokerStats = |
| BrokerConsumerStats(validTill, |
| consumerStatsResponse.msgrateout(), |
| consumerStatsResponse.msgthroughputout(), |
| consumerStatsResponse.msgrateredeliver(), |
| consumerStatsResponse.consumername(), |
| consumerStatsResponse.availablepermits(), |
| consumerStatsResponse.unackedmessages(), |
| consumerStatsResponse.blockedconsumeronunackedmsgs(), |
| consumerStatsResponse.address(), |
| consumerStatsResponse.connectedsince(), |
| consumerStatsResponse.type(), |
| consumerStatsResponse.msgrateexpired(), |
| consumerStatsResponse.msgbacklog()); |
| consumerStatsPromise.setValue(brokerStats); |
| } |
| } else { |
| LOG_WARN( |
| "ConsumerStatsResponse command - Received unknown request id from server: " |
| << consumerStatsResponse.request_id()); |
| } |
| break; |
| |
| } |
| |
| case BaseCommand::LOOKUP_RESPONSE: { |
| const CommandLookupTopicResponse& lookupTopicResponse = incomingCmd_.lookuptopicresponse(); |
| LOG_DEBUG( |
| cnxString_ << "Received lookup response from server. req_id: " << lookupTopicResponse.request_id()); |
| |
| Lock lock(mutex_); |
| PendingLookupRequestsMap::iterator it = pendingLookupRequests_.find( |
| lookupTopicResponse.request_id()); |
| if (it != pendingLookupRequests_.end()) { |
| LookupDataResultPromisePtr lookupDataPromise = it->second; |
| pendingLookupRequests_.erase(it); |
| numOfPendingLookupRequest_--; |
| lock.unlock(); |
| |
| if (!lookupTopicResponse.has_response() |
| || (lookupTopicResponse.response() |
| == CommandLookupTopicResponse::Failed)) { |
| if (lookupTopicResponse.has_error()) { |
| LOG_ERROR( |
| cnxString_ << "Failed lookup req_id: " << lookupTopicResponse.request_id() << " error: " << lookupTopicResponse.error()); |
| } else { |
| LOG_ERROR( |
| cnxString_ << "Failed lookup req_id: " << lookupTopicResponse.request_id() << " with empty response: "); |
| } |
| lookupDataPromise -> setFailed(ResultConnectError); |
| } else { |
| LOG_DEBUG( |
| cnxString_ << "Received lookup response from server. req_id: " << lookupTopicResponse.request_id() // |
| << " -- broker-url: " << lookupTopicResponse.brokerserviceurl() << " -- broker-tls-url: "// |
| << lookupTopicResponse.brokerserviceurltls() << " authoritative: " << lookupTopicResponse.authoritative()// |
| << " redirect: " << lookupTopicResponse.response()); |
| LookupDataResultPtr lookupResultPtr = |
| boost::make_shared<LookupDataResult>(); |
| |
| if (tlsSocket_) { |
| lookupResultPtr->setBrokerUrl(lookupTopicResponse.brokerserviceurltls()); |
| } else { |
| lookupResultPtr->setBrokerUrl(lookupTopicResponse.brokerserviceurl()); |
| } |
| |
| lookupResultPtr->setBrokerUrlSsl( |
| lookupTopicResponse.brokerserviceurltls()); |
| lookupResultPtr->setAuthoritative(lookupTopicResponse.authoritative()); |
| lookupResultPtr->setRedirect( |
| lookupTopicResponse.response() |
| == CommandLookupTopicResponse::Redirect); |
| lookupDataPromise->setValue(lookupResultPtr); |
| } |
| |
| } else { |
| LOG_WARN( |
| "Received unknown request id from server: " << lookupTopicResponse.request_id()); |
| } |
| break; |
| } |
| |
| |
| case BaseCommand::PRODUCER_SUCCESS: { |
| const CommandProducerSuccess& producerSuccess = incomingCmd_.producer_success(); |
| LOG_DEBUG( |
| cnxString_ << "Received success producer response from server. req_id: " << producerSuccess.request_id() // |
| << " -- producer name: " << producerSuccess.producer_name()); |
| |
| Lock lock(mutex_); |
| PendingRequestsMap::iterator it = pendingRequests_.find( |
| producerSuccess.request_id()); |
| if (it != pendingRequests_.end()) { |
| PendingRequestData requestData = it->second; |
| pendingRequests_.erase(it); |
| lock.unlock(); |
| |
| requestData.promise.setValue(producerSuccess.producer_name()); |
| requestData.timer->cancel(); |
| } |
| break; |
| } |
| |
| case BaseCommand::ERROR: { |
| const CommandError& error = incomingCmd_.error(); |
| Result result = getResult(error.error()); |
| LOG_WARN( |
| cnxString_ << "Received error response from server: " << result << " -- req_id: "<< error.request_id()); |
| |
| Lock lock(mutex_); |
| PendingRequestsMap::iterator it = pendingRequests_.find(error.request_id()); |
| if (it != pendingRequests_.end()) { |
| PendingRequestData requestData = it->second; |
| pendingRequests_.erase(it); |
| lock.unlock(); |
| |
| requestData.promise.setFailed(getResult(error.error())); |
| requestData.timer->cancel(); |
| } else { |
| lock.unlock(); |
| } |
| break; |
| } |
| |
| case BaseCommand::CLOSE_PRODUCER: { |
| const CommandCloseProducer& closeProducer = incomingCmd_.close_producer(); |
| int producerId = closeProducer.producer_id(); |
| |
| LOG_DEBUG("Broker notification of Closed producer: " << producerId); |
| |
| Lock lock(mutex_); |
| ProducersMap::iterator it = producers_.find(producerId); |
| if (it != producers_.end()) { |
| ProducerImplPtr producer = it->second.lock(); |
| lock.unlock(); |
| |
| if (producer) { |
| producer->disconnectProducer(); |
| } |
| } else { |
| LOG_ERROR( |
| cnxString_ << "Got invalid producer Id in closeProducer command: "<< producerId); |
| } |
| |
| break; |
| } |
| |
| case BaseCommand::CLOSE_CONSUMER: { |
| const CommandCloseConsumer& closeconsumer = incomingCmd_.close_consumer(); |
| int consumerId = closeconsumer.consumer_id(); |
| |
| LOG_DEBUG("Broker notification of Closed consumer: " << consumerId); |
| |
| Lock lock(mutex_); |
| ConsumersMap::iterator it = consumers_.find(consumerId); |
| if (it != consumers_.end()) { |
| ConsumerImplPtr consumer = it->second.lock(); |
| lock.unlock(); |
| |
| if (consumer) { |
| consumer->disconnectConsumer(); |
| } |
| } else { |
| LOG_ERROR( |
| cnxString_ << "Got invalid consumer Id in closeConsumer command: "<< consumerId); |
| } |
| |
| break; |
| } |
| |
| case BaseCommand::PING: { |
| // Respond to ping request |
| LOG_DEBUG(cnxString_ << "Replying to ping command"); |
| sendCommand(Commands::newPong()); |
| break; |
| } |
| |
| case BaseCommand::PONG: { |
| LOG_DEBUG(cnxString_ << "Received response to ping message"); |
| havePendingPingRequest_ = false; |
| break; |
| } |
| |
| default: { |
| LOG_WARN(cnxString_ << "Received invalid message from server"); |
| close(); |
| break; |
| } |
| } |
| } |
| } |
| } |
| |
| Future<Result, BrokerConsumerStats> |
| ClientConnection::newConsumerStats(const std::string topicName, const std::string subscriptionName, |
| uint64_t consumerId, uint64_t requestId) { |
| Lock lock(mutex_); |
| Promise<Result, BrokerConsumerStats> promise; |
| if (isClosed()) { |
| lock.unlock(); |
| LOG_ERROR(cnxString_ << " Client is not connected to the broker"); |
| promise.setFailed(ResultNotConnected); |
| } |
| pendingConsumerStatsMap_.insert(std::make_pair(requestId, promise)); |
| lock.unlock(); |
| sendCommand(Commands::newConsumerStats(outgoingCmd_, topicName, subscriptionName, consumerId, requestId)); |
| return promise.getFuture(); |
| } |
| |
| void ClientConnection::newTopicLookup(const std::string& destinationName, bool authoritative, |
| const uint64_t requestId, |
| LookupDataResultPromisePtr promise) { |
| newLookup(Commands::newLookup(outgoingCmd_, destinationName, authoritative, requestId), |
| requestId, promise); |
| } |
| |
| void ClientConnection::newPartitionedMetadataLookup(const std::string& destinationName, |
| const uint64_t requestId, |
| LookupDataResultPromisePtr promise) { |
| newLookup(Commands::newPartitionMetadataRequest(outgoingCmd_, destinationName, requestId), requestId, |
| promise); |
| } |
| |
| void ClientConnection::newLookup(const SharedBuffer& cmd, const uint64_t requestId, |
| LookupDataResultPromisePtr promise) { |
| Lock lock(mutex_); |
| boost::shared_ptr<LookupDataResultPtr> lookupDataResult; |
| lookupDataResult = boost::make_shared<LookupDataResultPtr>(); |
| if (isClosed()) { |
| lock.unlock(); |
| promise->setFailed(ResultNotConnected); |
| return; |
| } else if (numOfPendingLookupRequest_ >= maxPendingLookupRequest_) { |
| lock.unlock(); |
| promise->setFailed(ResultTooManyLookupRequestException); |
| return; |
| } |
| pendingLookupRequests_.insert(std::make_pair(requestId, promise)); |
| numOfPendingLookupRequest_++; |
| lock.unlock(); |
| sendCommand(cmd); |
| } |
| |
| void ClientConnection::sendCommand(const SharedBuffer& cmd) { |
| Lock lock(mutex_); |
| |
| if (pendingWriteOperations_++ == 0) { |
| |
| // Write immediately to socket |
| asyncWrite( |
| cmd.const_asio_buffer(), |
| customAllocWriteHandler( |
| boost::bind(&ClientConnection::handleSend, shared_from_this(), _1, cmd))); |
| } else { |
| // Queue to send later |
| pendingWriteBuffers_.push_back(cmd); |
| } |
| } |
| |
| void ClientConnection::sendMessage(const OpSendMsg& opSend) { |
| Lock lock(mutex_); |
| |
| if (pendingWriteOperations_++ == 0) { |
| |
| PairSharedBuffer buffer = Commands::newSend(outgoingBuffer_, outgoingCmd_, |
| opSend.producerId_, opSend.sequenceId_, |
| getChecksumType(), opSend.msg_); |
| |
| // Write immediately to socket |
| asyncWrite( |
| buffer, |
| customAllocWriteHandler( |
| boost::bind(&ClientConnection::handleSendPair, shared_from_this(), _1))); |
| } else { |
| // Queue to send later |
| pendingWriteBuffers_.push_back(opSend); |
| } |
| } |
| |
| void ClientConnection::handleSend(const boost::system::error_code& err, const SharedBuffer&) { |
| if (err) { |
| LOG_WARN(cnxString_ << "Could not send message on connection: " << err << " " << err.message()); |
| close(); |
| } else { |
| sendPendingCommands(); |
| } |
| } |
| |
| void ClientConnection::handleSendPair(const boost::system::error_code& err) { |
| if (err) { |
| LOG_WARN(cnxString_ << "Could not send pair message on connection: " << err << " " << err.message()); |
| close(); |
| } else { |
| sendPendingCommands(); |
| } |
| } |
| |
| void ClientConnection::sendPendingCommands() { |
| Lock lock(mutex_); |
| |
| if (--pendingWriteOperations_ > 0) { |
| assert(!pendingWriteBuffers_.empty()); |
| boost::any any = pendingWriteBuffers_.front(); |
| pendingWriteBuffers_.pop_front(); |
| |
| if (any.type() == typeid(SharedBuffer)) { |
| SharedBuffer buffer = boost::any_cast<SharedBuffer>(any); |
| asyncWrite( |
| buffer.const_asio_buffer(), |
| customAllocWriteHandler( |
| boost::bind(&ClientConnection::handleSend, shared_from_this(), _1, |
| buffer))); |
| } else { |
| assert(any.type() == typeid(OpSendMsg)); |
| |
| const OpSendMsg& op = boost::any_cast<const OpSendMsg&>(any); |
| PairSharedBuffer buffer = Commands::newSend(outgoingBuffer_, outgoingCmd_, |
| op.producerId_, op.sequenceId_, |
| getChecksumType(), op.msg_); |
| |
| asyncWrite( |
| buffer, |
| customAllocWriteHandler( |
| boost::bind(&ClientConnection::handleSendPair, shared_from_this(), |
| _1))); |
| } |
| } else { |
| // No more pending writes |
| outgoingBuffer_.reset(); |
| } |
| } |
| |
| Future<Result, std::string> ClientConnection::sendRequestWithId(SharedBuffer cmd, int requestId) { |
| Lock lock(mutex_); |
| |
| if (isClosed()) { |
| lock.unlock(); |
| Promise<Result, std::string> promise; |
| promise.setFailed(ResultNotConnected); |
| return promise.getFuture(); |
| } |
| |
| PendingRequestData requestData; |
| requestData.timer = executor_->createDeadlineTimer(); |
| requestData.timer->expires_from_now(operationsTimeout_); |
| requestData.timer->async_wait( |
| boost::bind(&ClientConnection::handleRequestTimeout, shared_from_this(), _1, |
| requestData)); |
| |
| pendingRequests_.insert(std::make_pair(requestId, requestData)); |
| lock.unlock(); |
| |
| sendCommand(cmd); |
| return requestData.promise.getFuture(); |
| } |
| |
| void ClientConnection::handleRequestTimeout(const boost::system::error_code& ec, |
| PendingRequestData pendingRequestData) { |
| if (!ec) { |
| pendingRequestData.promise.setFailed(ResultTimeout); |
| } |
| } |
| |
| void ClientConnection::handleKeepAliveTimeout() { |
| if (isClosed()) { |
| return; |
| } |
| |
| if (havePendingPingRequest_) { |
| LOG_WARN(cnxString_ << "Forcing connection to close after keep-alive timeout"); |
| close(); |
| } else { |
| // Send keep alive probe to peer |
| LOG_DEBUG(cnxString_ << "Sending ping message"); |
| havePendingPingRequest_ = true; |
| sendCommand(Commands::newPing()); |
| |
| keepAliveTimer_->expires_from_now(boost::posix_time::seconds(KeepAliveIntervalInSeconds)); |
| keepAliveTimer_->async_wait( |
| boost::bind(&ClientConnection::handleKeepAliveTimeout, shared_from_this())); |
| } |
| } |
| |
| void ClientConnection::handleConsumerStatsTimeout(const boost::system::error_code& ec, |
| std::vector<uint64_t> consumerStatsRequests) { |
| if (ec) { |
| LOG_DEBUG(cnxString_ << " Ignoring timer cancelled event, code[" << ec << "]"); |
| return; |
| } |
| startConsumerStatsTimer(consumerStatsRequests); |
| } |
| |
| void ClientConnection::close() { |
| Lock lock(mutex_); |
| state_ = Disconnected; |
| boost::system::error_code err; |
| socket_->close(err); |
| lock.unlock(); |
| |
| LOG_INFO(cnxString_ << "Connection closed"); |
| |
| if (keepAliveTimer_) { |
| keepAliveTimer_->cancel(); |
| } |
| |
| if (consumerStatsRequestTimer_) { |
| consumerStatsRequestTimer_->cancel(); |
| } |
| for (ProducersMap::iterator it = producers_.begin(); it != producers_.end(); ++it) { |
| HandlerBase::handleDisconnection(ResultConnectError, shared_from_this(), it->second); |
| } |
| |
| for (ConsumersMap::iterator it = consumers_.begin(); it != consumers_.end(); ++it) { |
| HandlerBase::handleDisconnection(ResultConnectError, shared_from_this(), it->second); |
| } |
| |
| connectPromise_.setFailed(ResultConnectError); |
| |
| // Fail all pending operations on the connection |
| for (PendingRequestsMap::iterator it = pendingRequests_.begin(); it != pendingRequests_.end(); ++it ) { |
| it->second.promise.setFailed(ResultConnectError); |
| } |
| |
| // Fail all pending lookup-requests on the connection |
| lock.lock(); |
| PendingLookupRequestsMap pendingLookupRequests; |
| pendingLookupRequests_.swap(pendingLookupRequests); |
| numOfPendingLookupRequest_ -= pendingLookupRequests.size(); |
| |
| PendingConsumerStatsMap pendingConsumerStatsMap; |
| pendingConsumerStatsMap_.swap(pendingConsumerStatsMap); |
| lock.unlock(); |
| |
| for (PendingLookupRequestsMap::iterator it = pendingLookupRequests.begin(); it != pendingLookupRequests.end(); ++it) { |
| it->second->setFailed(ResultConnectError); |
| } |
| |
| |
| for (PendingConsumerStatsMap::iterator it = pendingConsumerStatsMap.begin(); |
| it != pendingConsumerStatsMap.end(); ++it) { |
| LOG_ERROR(cnxString_ << " Closing Client Connection, please try again later"); |
| it->second.setFailed(ResultConnectError); |
| } |
| |
| if (tlsSocket_) { |
| tlsSocket_->lowest_layer().close(); |
| } |
| } |
| |
| bool ClientConnection::isClosed() const { |
| return state_ == Disconnected; |
| } |
| |
| Future<Result, ClientConnectionWeakPtr> ClientConnection::getConnectFuture() { |
| return connectPromise_.getFuture(); |
| } |
| |
| void ClientConnection::registerProducer(int producerId, ProducerImplPtr producer) { |
| Lock lock(mutex_); |
| producers_.insert(std::make_pair(producerId, producer)); |
| } |
| |
| void ClientConnection::registerConsumer(int consumerId, ConsumerImplPtr consumer) { |
| Lock lock(mutex_); |
| consumers_.insert(std::make_pair(consumerId, consumer)); |
| } |
| |
| void ClientConnection::removeProducer(int producerId) { |
| Lock lock(mutex_); |
| producers_.erase(producerId); |
| } |
| |
| void ClientConnection::removeConsumer(int consumerId) { |
| Lock lock(mutex_); |
| consumers_.erase(consumerId); |
| } |
| |
| const std::string& ClientConnection::brokerAddress() const { |
| return address_; |
| } |
| |
| const std::string& ClientConnection::cnxString() const { |
| return cnxString_; |
| } |
| |
| int ClientConnection::getServerProtocolVersion() const { |
| return serverProtocolVersion_; |
| } |
| |
| Commands::ChecksumType ClientConnection::getChecksumType() const { |
| return getServerProtocolVersion() >= proto::v6 ? |
| Commands::Crc32c : Commands::None; |
| } |
| } |
