#include <dhcp_ddns/dhcp_ddns_log.h>
#include <dhcp_ddns/ncr_io.h>
+#include <asio.hpp>
#include <boost/algorithm/string/predicate.hpp>
namespace isc {
NameChangeSender::NameChangeSender(RequestSendHandler& send_handler,
size_t send_queue_max)
: sending_(false), send_handler_(send_handler),
- send_queue_max_(send_queue_max) {
+ send_queue_max_(send_queue_max), io_service_(NULL) {
// Queue size must be big enough to hold at least 1 entry.
setQueueMaxSize(send_queue_max);
// Call implementation dependent open.
try {
+ // Remember io service we're given.
+ io_service_ = &io_service;
open(io_service);
} catch (const isc::Exception& ex) {
stopSending();
void
NameChangeSender::stopSending() {
+ // Set it send indicator to false, no matter what. This allows us to at
+ // least try to re-open via startSending(). Also, setting it false now,
+ // allows us to break sendNext() chain in invokeSendHandler.
+ setSending(false);
+
+ // If there is an outstanding IO to complete, attempt to process it.
+ if (ioReady() && io_service_ != NULL) {
+ try {
+ runReadyIO();
+ } catch (const std::exception& ex) {
+ // Swallow exceptions. If we have some sort of error we'll log
+ // it but we won't propagate the throw.
+ LOG_ERROR(dhcp_ddns_logger,
+ DHCP_DDNS_NCR_FLUSH_IO_ERROR).arg(ex.what());
+ }
+ }
+
try {
// Call implementation dependent close.
close();
DHCP_DDNS_NCR_SEND_CLOSE_ERROR).arg(ex.what());
}
- // Set it false, no matter what. This allows us to at least try to
- // re-open via startSending().
- setSending(false);
+ io_service_ = NULL;
}
void
// Set up the next send
try {
- sendNext();
+ if (amSending()) {
+ sendNext();
+ }
} catch (const isc::Exception& ex) {
// It is possible though unlikely, for sendNext to fail without
// scheduling the send. While, unlikely, it does mean the callback
isc_throw(NotImplemented, "NameChangeSender::getSelectFd is not supported");
}
+void
+NameChangeSender::runReadyIO() {
+ if (!io_service_) {
+ isc_throw(NcrSenderError, "NameChangeSender::runReadyIO"
+ " sender io service is null");
+ }
+
+ // We shouldn't be here if IO isn't ready to execute.
+ // By running poll we're gauranteed not to hang.
+ /// @todo Trac# 3325 requests that asiolink::IOService provide a
+ /// wrapper for poll().
+ io_service_->get_io_service().poll_one();
+}
+
+
} // namespace isc::dhcp_ddns
} // namespace isc
/// @throw NcrSenderError if the sender is not in send mode,
virtual int getSelectFd() = 0;
+ virtual bool ioReady() = 0;
+
protected:
/// @brief Dequeues and sends the next request on the send queue.
///
/// end of the queue.
const NameChangeRequestPtr& peekAt(const size_t index) const;
+ /// @brief Processes sender IO events
+ ///
+ /// Executes at most one ready handler on the sender's IO service. If
+ /// no handlers are ready it returns immediately.
+ /// @warning - Running all ready handlers, in theory, could process all
+ /// messages currently queued.
+ virtual void runReadyIO();
+
protected:
/// @brief Returns a reference to the send queue.
SendQueue& getSendQueue() {
/// @brief Pointer to the request which is in the process of being sent.
NameChangeRequestPtr ncr_to_send_;
+
+ /// @brief Pointer to the IOService currently being used by the sender.
+ /// @note We need to remember the io_service but we receive it by
+ /// reference. Use a raw pointer to store it. This value should never be
+ /// exposed and is only valid while in send mode.
+ asiolink::IOService* io_service_;
};
/// @brief Defines a smart pointer to an instance of a sender.
// Verify select_fd is valid and currently shows no ready to read.
ASSERT_NE(dhcp_ddns::WatchSocket::INVALID_SOCKET, select_fd);
+
+ // Make sure select_fd does evaluates to not ready via select and
+ // that ioReady() method agrees.
ASSERT_EQ(0, selectCheck(select_fd));
+ ASSERT_FALSE(sender.ioReady());
// Iterate over a series of messages, sending each one. Since we
// do not invoke IOService::run, then the messages should accumulate
// IOService::run_one. This should complete the send of exactly one
// message and the queue count should decrement accordingly.
for (int i = num_msgs; i > 0; i--) {
- // Verify that sender shows IO ready.
+ // Make sure select_fd does evaluates to ready via select and
+ // that ioReady() method agrees.
ASSERT_TRUE(selectCheck(select_fd) > 0);
+ ASSERT_TRUE(sender.ioReady());
// Execute at one ready handler.
- io_service.run_one();
+ ASSERT_NO_THROW(sender.runReadyIO());
// Verify that the queue count decrements in step with each run.
EXPECT_EQ(i-1, sender.getQueueSize());
}
- // Verify that sender shows no IO ready.
- EXPECT_EQ(0, selectCheck(select_fd));
+ // Make sure select_fd does evaluates to not ready via select and
+ // that ioReady() method agrees.
+ ASSERT_EQ(0, selectCheck(select_fd));
+ ASSERT_FALSE(sender.ioReady());
// Verify that the queue is empty.
EXPECT_EQ(0, sender.getQueueSize());
// Verify that flushing the queue is not allowed in sending state.
EXPECT_THROW(sender.clearSendQueue(), NcrSenderError);
- // Put a message on the queue.
- EXPECT_NO_THROW(sender.sendRequest(ncr));
- EXPECT_EQ(1, sender.getQueueSize());
+ // Put num_msgs messages on the queue.
+ for (int i = 0; i < num_msgs; i++) {
+ ASSERT_NO_THROW(ncr = NameChangeRequest::fromJSON(valid_msgs[i]));
+ EXPECT_NO_THROW(sender.sendRequest(ncr));
+ }
+
+ // Make sure we have number of messages expected.
+ EXPECT_EQ(num_msgs, sender.getQueueSize());
// Verify that we can gracefully stop sending.
EXPECT_NO_THROW(sender.stopSending());
EXPECT_FALSE(sender.amSending());
// Verify that the queue is preserved after leaving sending state.
- EXPECT_EQ(1, sender.getQueueSize());
+ EXPECT_EQ(num_msgs - 1, sender.getQueueSize());
// Verify that flushing the queue works when not sending.
EXPECT_NO_THROW(sender.clearSendQueue());
ASSERT_NO_THROW(sender.startSending(io_service));
EXPECT_TRUE(sender.amSending());
- // Fetch the sender's select-fd.
- int select_fd = sender.getSelectFd();
-
// Create and queue up a message.
NameChangeRequestPtr ncr;
ASSERT_NO_THROW(ncr = NameChangeRequest::fromJSON(valid_msgs[0]));
EXPECT_NO_THROW(sender.sendRequest(ncr));
EXPECT_EQ(1, sender.getQueueSize());
- // message and the queue count should decrement accordingly.
- // Execute at one ready handler.
- ASSERT_TRUE(selectCheck(select_fd) > 0);
- ASSERT_NO_THROW(io_service.run_one());
+ // Verify we have a ready IO, then execute at one ready handler.
+ ASSERT_TRUE(sender.ioReady());
+ ASSERT_NO_THROW(sender.runReadyIO());
// Verify that sender shows no IO ready.
// and that the queue is empty.
- EXPECT_EQ(0, selectCheck(select_fd));
+ ASSERT_FALSE(sender.ioReady());
EXPECT_EQ(0, sender.getQueueSize());
}
// Take sender1 out of send mode.
ASSERT_NO_THROW(sender1.stopSending());
ASSERT_FALSE(sender1.amSending());
+ // Stopping should have completed the first message.
+ --num_msgs;
+ EXPECT_EQ(num_msgs, sender1.getQueueSize());
// Transfer should succeed. Verify sender1 has none,
// and sender2 has num_msgs queued.
// Run one handler. This should execute the send completion handler
// after sending the first message. Duing completion handling, we will
// attempt to queue the second message which should fail.
- ASSERT_NO_THROW(io_service.run_one());
+ ASSERT_NO_THROW(sender.runReadyIO());
// Verify handler got called twice. First request should have be sent
// without error, second call should have failed to send due to watch
// after sending the message. Duing completion handling clearing the
// watch socket should fail, which will close the socket, but not
// result in a throw.
- ASSERT_NO_THROW(io_service.run_one());
+ ASSERT_NO_THROW(sender.runReadyIO());
// Verify handler got called twice. First request should have be sent
// without error, second call should have failed to send due to watch
projects / thirdparty / kea.git / commitdiff
