[master] Merge branch 'trac3221'

Integrates use of NameChangeUDPSender in D2ClientMgr.

src/lib/dhcp_ddns/Makefile.am

@@ -34,6 +34,7 @@ libb10_dhcp_ddns_la_SOURCES += dhcp_ddns_log.cc dhcp_ddns_log.h
 libb10_dhcp_ddns_la_SOURCES += ncr_io.cc ncr_io.h
 libb10_dhcp_ddns_la_SOURCES += ncr_msg.cc ncr_msg.h
 libb10_dhcp_ddns_la_SOURCES += ncr_udp.cc ncr_udp.h
+libb10_dhcp_ddns_la_SOURCES += watch_socket.cc watch_socket.h
 
 nodist_libb10_dhcp_ddns_la_SOURCES = dhcp_ddns_messages.cc dhcp_ddns_messages.h
 

src/lib/dhcp_ddns/dhcp_ddns_messages.mes

@@ -1,4 +1,4 @@
-# Copyright (C) 2013  Internet Systems Consortium, Inc. ("ISC")
+# Copyright (C) 2013-2014  Internet Systems Consortium, Inc. ("ISC")
 #
 # Permission to use, copy, modify, and/or distribute this software for any
 # purpose with or without fee is hereby granted, provided that the above
@@ -32,19 +32,25 @@ start another read after receiving a request.  While possible, this is highly
 unlikely and is probably a programmatic error.  The application should recover
 on its own.
 
-% DHCP_DDNS_NCR_SEND_CLOSE_ERROR DHCP-DDNS client encountered an error while closing the sender connection used to send NameChangeRequests : %1
+% DHCP_DDNS_NCR_SEND_CLOSE_ERROR DHCP-DDNS client encountered an error while closing the sender connection used to send NameChangeRequests: %1
 This is an error message that indicates the DHCP-DDNS client was unable to
 close the connection used to send NameChangeRequests.  Closure may occur during
 the course of error recovery or during normal shutdown procedure.  In either
 case the error is unlikely to impair the client's ability to send requests but
 it should be reported for analysis.
 
-% DHCP_DDNS_NCR_SEND_NEXT_ERROR DHCP-DDNS client could not initiate the next request send following send completion.
+% DHCP_DDNS_NCR_SEND_NEXT_ERROR DHCP-DDNS client could not initiate the next request send following send completion: %1
 This is a error message indicating that NameChangeRequest sender could not
 start another send after completing the send of the previous request.  While
 possible, this is highly unlikely and is probably a programmatic error.  The
 application should recover on its own.
 
+% DHCP_DDNS_NCR_UDP_CLEAR_READY_ERROR NCR UDP watch socket failed to clear: %1
+This is an error message that indicates the application was unable to reset the
+UDP NCR sender ready status after completing a send.  This is programmatic error
+that should be reported.  The application may or may not continue to operate
+correctly.
+
 % DHCP_DDNS_NCR_UDP_RECV_CANCELED UDP socket receive was canceled while listening for DNS Update requests: %1
 This is an informational  message indicating that the listening over a UDP socket for DNS update requests has been canceled.  This is a normal part of suspending listening operations.
 
@@ -74,3 +80,15 @@ This is an error message that indicates that an exception was thrown but not
 caught in the application's send completion handler.  This is a programmatic
 error that needs to be reported.  Dependent upon the nature of the error the
 client may or may not continue operating normally.
+
+% DHCP_DDNS_WATCH_SINK_CLOSE_ERROR Sink-side watch socket failed to close: %1
+This is an error message that indicates the application was unable to close
+the inbound side of a NCR sender's watch socket.  While technically possible
+this error is highly unlikely to occur and should not impair the application's
+ability to process requests.
+
+% DHCP_DDNS_WATCH_SOURCE_CLOSE_ERROR Source-side watch socket failed to close: %1
+This is an error message that indicates the application was unable to close
+the outbound side of a NCR sender's watch socket.  While technically possible
+this error is highly unlikely to occur and should not impair the application's
+ability to process requests.

src/lib/dhcp_ddns/ncr_io.cc

@@ -1,4 +1,4 @@
-// Copyright (C) 2013 Internet Systems Consortium, Inc. ("ISC")
+// Copyright (C) 2013-2014 Internet Systems Consortium, Inc. ("ISC")
 //
 // Permission to use, copy, modify, and/or distribute this software for any
 // purpose with or without fee is hereby granted, provided that the above
@@ -23,7 +23,7 @@ namespace dhcp_ddns {
 NameChangeProtocol stringToNcrProtocol(const std::string& protocol_str) {
     if (boost::iequals(protocol_str, "UDP")) {
         return (NCR_UDP);
-    } 
+    }
 
     if (boost::iequals(protocol_str, "TCP")) {
         return (NCR_TCP);
@@ -162,10 +162,7 @@ NameChangeSender::NameChangeSender(RequestSendHandler& send_handler,
       send_queue_max_(send_queue_max) {
 
     // Queue size must be big enough to hold at least 1 entry.
-    if (send_queue_max == 0) {
-        isc_throw(NcrSenderError, "NameChangeSender constructor"
-                  " queue size must be greater than zero");
-    }
+    setQueueMaxSize(send_queue_max);
 }
 
 void
@@ -318,5 +315,57 @@ NameChangeSender::clearSendQueue() {
     send_queue_.clear();
 }
 
+void
+NameChangeSender::setQueueMaxSize(const size_t new_max) {
+    if (new_max == 0) {
+        isc_throw(NcrSenderError, "NameChangeSender:"
+                  " queue size must be greater than zero");
+    }
+
+    send_queue_max_ = new_max;
+
+}
+const NameChangeRequestPtr&
+NameChangeSender::peekAt(const size_t index) const {
+    if (index >= getQueueSize()) {
+        isc_throw(NcrSenderError,
+                  "NameChangeSender::peekAt peek beyond end of queue attempted"
+                  << " index: " << index << " queue size: " << getQueueSize());
+    }
+
+    return (send_queue_.at(index));
+}
+
+
+void
+NameChangeSender::assumeQueue(NameChangeSender& source_sender) {
+    if (source_sender.amSending()) {
+        isc_throw(NcrSenderError, "Cannot assume queue:"
+                  " source sender is actively sending");
+    }
+
+    if (amSending()) {
+        isc_throw(NcrSenderError, "Cannot assume queue:"
+                  " target sender is actively sending");
+    }
+
+    if (getQueueMaxSize() < source_sender.getQueueSize()) {
+        isc_throw(NcrSenderError, "Cannot assume queue:"
+                  " source queue count exceeds target queue max");
+    }
+
+    if (!send_queue_.empty()) {
+        isc_throw(NcrSenderError, "Cannot assume queue:"
+                  " target queue is not empty");
+    }
+
+    send_queue_.swap(source_sender.getSendQueue());
+}
+
+int
+NameChangeSender::getSelectFd() {
+    isc_throw(NotImplemented, "NameChangeSender::getSelectFd is not supported");
+}
+
 } // namespace isc::dhcp_ddns
 } // namespace isc

src/lib/dhcp_ddns/ncr_io.h

@@ -1,4 +1,4 @@
-// Copyright (C) 2013 Internet Systems Consortium, Inc. ("ISC")
+// Copyright (C) 2013-2014 Internet Systems Consortium, Inc. ("ISC")
 //
 // Permission to use, copy, modify, and/or distribute this software for any
 // purpose with or without fee is hereby granted, provided that the above
@@ -547,6 +547,35 @@ public:
     /// capacity.
     void sendRequest(NameChangeRequestPtr& ncr);
 
+    /// @brief Move all queued requests from a given sender into the send queue
+    ///
+    /// Moves all of the entries in the given sender's queue and places them
+    /// into send queue.  This provides a mechanism of reassigning queued
+    /// messages from one sender to another. This is useful for dealing with
+    /// dynamic configuration changes.
+    ///
+    /// @param source_sender from whom the queued messages will be taken
+    ///
+    /// @throw NcrSenderError if either sender is in send mode, if the number of
+    /// messages in the source sender's queue is larger than this sender's
+    /// maxium queue size, or if this sender's queue is not empty.
+    void assumeQueue(NameChangeSender& source_sender);
+
+    /// @brief Returns a file descriptor suitable for use with select
+    ///
+    /// The value returned is an open file descriptor which can be used with
+    /// select() system call to monitor the sender for IO events.  This allows
+    /// NameChangeSenders to be used in applications which use select, rather
+    /// than IOService to wait for IO events to occur.
+    ///
+    /// @warning Attempting other use of this value may lead to unpredictable
+    /// behavior in the sender.
+    ///
+    /// @return Returns an "open" file descriptor
+    ///
+    /// @throw NcrSenderError if the sender is not in send mode,
+    virtual int getSelectFd() = 0;
+
 protected:
     /// @brief Dequeues and sends the next request on the send queue.
     ///
@@ -659,11 +688,39 @@ public:
         return (send_queue_max_);
     }
 
+    /// @brief Sets the maxium queue size to the given value.
+    ///
+    /// Sets the maximum number of entries allowed in the queue to the
+    /// the given value.
+    ///
+    /// @param new_max the new value to use as the maximum
+    ///
+    /// @throw NcrSenderError if the value is less than one.
+    void setQueueMaxSize(const size_t new_max);
+
     /// @brief Returns the number of entries currently in the send queue.
     size_t getQueueSize() const {
         return (send_queue_.size());
     }
 
+    /// @brief Returns the entry at a given position in the queue.
+    ///
+    /// Note that the entry is not removed from the queue.
+    /// @param index the index of the entry in the queue to fetch.
+    /// Valid values are 0 (front of the queue) to (queue size - 1).
+    ///
+    /// @return Pointer reference to the queue entry.
+    ///
+    /// @throw NcrSenderError if the given index is beyond the
+    /// end of the queue.
+    const NameChangeRequestPtr& peekAt(const size_t index) const;
+
+protected:
+    /// @brief Returns a reference to the send queue.
+    SendQueue& getSendQueue() {
+        return (send_queue_);
+    }
+
 private:
     /// @brief Sets the sending indicator to the given value.
     ///

src/lib/dhcp_ddns/ncr_msg.h

@@ -72,7 +72,7 @@ enum NameChangeFormat {
 
 /// @brief Function which converts labels to  NameChangeFormat enum values.
 ///
-/// @param format_str text to convert to an enum.
+/// @param fmt_str text to convert to an enum.
 /// Valid string values: "JSON"
 ///
 /// @return NameChangeFormat value which maps to the given string.

src/lib/dhcp_ddns/ncr_udp.cc

@@ -1,4 +1,4 @@
-// Copyright (C) 2013 Internet Systems Consortium, Inc. ("ISC")
+// Copyright (C) 2013-2014 Internet Systems Consortium, Inc. ("ISC")
 //
 // Permission to use, copy, modify, and/or distribute this software for any
 // purpose with or without fee is hereby granted, provided that the above
@@ -255,6 +255,8 @@ NameChangeUDPSender::open(isc::asiolink::IOService& io_service) {
                            UDPEndpoint(server_address_, server_port_));
 
     send_callback_->setDataSource(server_endpoint_);
+
+    watch_socket_.reset(new WatchSocket());
 }
 
 void
@@ -281,6 +283,8 @@ NameChangeUDPSender::close() {
     }
 
     socket_.reset();
+
+    watch_socket_.reset();
 }
 
 void
@@ -297,11 +301,32 @@ NameChangeUDPSender::doSend(NameChangeRequestPtr& ncr) {
     // Call the socket's asychronous send, passing our callback
     socket_->asyncSend(send_callback_->getData(), send_callback_->getPutLen(),
                        send_callback_->getDataSource().get(), *send_callback_);
+
+    // Set IO ready marker so sender activity is visible to select() or poll().
+    // Note, if this call throws it will manifest itself as a throw from
+    // from sendRequest() which the application calls directly and is documented
+    // as throwing exceptions; or caught inside invokeSendHandler() which
+    // will invoke the application's send_handler with an error status.
+    watch_socket_->markReady();
 }
 
 void
 NameChangeUDPSender::sendCompletionHandler(const bool successful,
                                            const UDPCallback *send_callback) {
+    // Clear the IO ready marker.
+    try {
+        watch_socket_->clearReady();
+    } catch (const std::exception& ex) {
+        // This can only happen if the WatchSocket's select_fd has been
+        // compromised which is a programmatic error. We'll log the error
+        // here, then continue on and process the IO result we were given.
+        // WatchSocket issue will resurface on the next send as a closed
+        // fd in markReady().  This allows application's handler to deal
+        // with watch errors more uniformly.
+        LOG_ERROR(dhcp_ddns_logger, DHCP_DDNS_NCR_UDP_CLEAR_READY_ERROR)
+                 .arg(ex.what());
+    }
+
     Result result;
     if (successful) {
         result = SUCCESS;
@@ -323,5 +348,17 @@ NameChangeUDPSender::sendCompletionHandler(const bool successful,
     // Call the application's registered request send handler.
     invokeSendHandler(result);
 }
+
+int
+NameChangeUDPSender::getSelectFd() {
+    if (!amSending()) {
+        isc_throw(NotImplemented, "NameChangeUDPSender::getSelectFd"
+                                  " not in send mode");
+    }
+
+    return(watch_socket_->getSelectFd());
+}
+
+
 }; // end of isc::dhcp_ddns namespace
 }; // end of isc namespace

src/lib/dhcp_ddns/ncr_udp.h

@@ -1,4 +1,4 @@
-// Copyright (C) 2013 Internet Systems Consortium, Inc. ("ISC")
+// Copyright (C) 2013-2014 Internet Systems Consortium, Inc. ("ISC")
 //
 // Permission to use, copy, modify, and/or distribute this software for any
 // purpose with or without fee is hereby granted, provided that the above
@@ -112,10 +112,12 @@
 #include <asiolink/udp_endpoint.h>
 #include <asiolink/udp_socket.h>
 #include <dhcp_ddns/ncr_io.h>
+#include <dhcp_ddns/watch_socket.h>
 #include <util/buffer.h>
 
 #include <boost/shared_array.hpp>
 
+
 /// responsibility of the completion handler to perform the steps necessary
 /// to interpret the raw data provided by the service outcome.   The
 /// UDPCallback operator implementation is mostly a pass through.
@@ -502,6 +504,7 @@ public:
     /// asyncSend() method is called, passing in send_callback_ member's
     /// transfer buffer as the send buffer and the send_callback_ itself
     /// as the callback object.
+    /// @param ncr NameChangeRequest to send.
     virtual void doSend(NameChangeRequestPtr& ncr);
 
     /// @brief Implements the NameChangeRequest level send completion handler.
@@ -524,6 +527,21 @@ public:
     void sendCompletionHandler(const bool successful,
                                const UDPCallback* send_callback);
 
+    /// @brief Returns a file descriptor suitable for use with select
+    ///
+    /// The value returned is an open file descriptor which can be used with
+    /// select() system call to monitor the sender for IO events.  This allows
+    /// NameChangeUDPSenders to be used in applications which use select,
+    /// rather than IOService to wait for IO events to occur.
+    ///
+    /// @warning Attempting other use of this value may lead to unpredictable
+    /// behavior in the sender.
+    ///
+    /// @return Returns an "open" file descriptor
+    ///
+    /// @throw NcrSenderError if the sender is not in send mode,
+    virtual int getSelectFd();
+
 private:
     /// @brief IP address from which to send.
     isc::asiolink::IOAddress ip_address_;
@@ -554,6 +572,9 @@ private:
 
     /// @brief Flag which enables the reuse address socket option if true.
     bool reuse_address_;
+
+    /// @brief Pointer to WatchSocket instance supplying the "select-fd".
+    WatchSocketPtr watch_socket_;
 };
 
 } // namespace isc::dhcp_ddns

src/lib/dhcp_ddns/tests/Makefile.am

@@ -29,6 +29,8 @@ TESTS += libdhcp_ddns_unittests
 libdhcp_ddns_unittests_SOURCES  = run_unittests.cc
 libdhcp_ddns_unittests_SOURCES += ncr_unittests.cc
 libdhcp_ddns_unittests_SOURCES += ncr_udp_unittests.cc
+libdhcp_ddns_unittests_SOURCES += test_utils.cc test_utils.h
+libdhcp_ddns_unittests_SOURCES += watch_socket_unittests.cc
 
 libdhcp_ddns_unittests_CPPFLAGS = $(AM_CPPFLAGS) $(GTEST_INCLUDES) $(LOG4CPLUS_INCLUDES)
 

src/lib/dhcp_ddns/tests/ncr_udp_unittests.cc

@@ -1,4 +1,4 @@
-// Copyright (C) 2013  Internet Systems Consortium, Inc. ("ISC")
+// Copyright (C) 2013-2014  Internet Systems Consortium, Inc. ("ISC")
 //
 // Permission to use, copy, modify, and/or distribute this software for any
 // purpose with or without fee is hereby granted, provided that the above
@@ -16,6 +16,7 @@
 #include <dhcp_ddns/ncr_io.h>
 #include <dhcp_ddns/ncr_udp.h>
 #include <util/time_utilities.h>
+#include <test_utils.h>
 
 #include <asio/ip/udp.hpp>
 #include <boost/function.hpp>
@@ -23,6 +24,8 @@
 #include <gtest/gtest.h>
 #include <algorithm>
 
+#include <sys/select.h>
+
 using namespace std;
 using namespace isc;
 using namespace isc::dhcp_ddns;
@@ -113,6 +116,7 @@ TEST(NameChangeUDPListenerBasicTest, basicListenTests) {
 
     // Verify that we can start listening.
     EXPECT_NO_THROW(listener->startListening(io_service));
+
     // Verify that we are in listening mode.
     EXPECT_TRUE(listener->amListening());
     // Verify that a read is in progress.
@@ -268,9 +272,20 @@ TEST_F(NameChangeUDPListenerTest, basicReceivetest) {
 /// @brief A NOP derivation for constructor test purposes.
 class SimpleSendHandler : public NameChangeSender::RequestSendHandler {
 public:
-    virtual void operator ()(const NameChangeSender::Result,
+    SimpleSendHandler() : pass_count_(0), error_count_(0) {
+    }
+
+    virtual void operator ()(const NameChangeSender::Result result,
                              NameChangeRequestPtr&) {
+        if (result == NameChangeSender::SUCCESS) {
+            ++pass_count_;
+        } else {
+            ++error_count_;
+        }
     }
+
+    int pass_count_;
+    int error_count_;
 };
 
 /// @brief Tests the NameChangeUDPSender constructors.
@@ -311,7 +326,6 @@ TEST(NameChangeUDPSenderBasicTest, constructionTests) {
 /// This test verifies that:
 TEST(NameChangeUDPSenderBasicTest, basicSendTests) {
     isc::asiolink::IOAddress ip_address(TEST_ADDRESS);
-    uint32_t port = SENDER_PORT;
     isc::asiolink::IOService io_service;
     SimpleSendHandler ncr_handler;
 
@@ -320,8 +334,9 @@ TEST(NameChangeUDPSenderBasicTest, basicSendTests) {
 
     // Create the sender, setting the queue max equal to the number of
     // messages we will have in the list.
-    NameChangeUDPSender sender(ip_address, port, ip_address, port,
-                               FMT_JSON, ncr_handler, num_msgs);
+    NameChangeUDPSender sender(ip_address, SENDER_PORT, ip_address,
+                               LISTENER_PORT, FMT_JSON, ncr_handler,
+                               num_msgs, true);
 
     // Verify that we can start sending.
     EXPECT_NO_THROW(sender.startSending(io_service));
@@ -341,30 +356,55 @@ TEST(NameChangeUDPSenderBasicTest, basicSendTests) {
     EXPECT_NO_THROW(sender.startSending(io_service));
     EXPECT_TRUE(sender.amSending());
 
+    // Fetch the sender's select-fd.
+    int select_fd = sender.getSelectFd();
+
+    // Verify select_fd is valid and currently shows no ready to read.
+    ASSERT_NE(dhcp_ddns::WatchSocket::INVALID_SOCKET, select_fd);
+    ASSERT_EQ(0, selectCheck(select_fd));
+
     // Iterate over a series of messages, sending each one. Since we
     // do not invoke IOService::run, then the messages should accumulate
     // in the queue.
     NameChangeRequestPtr ncr;
+    NameChangeRequestPtr ncr2;
     for (int i = 0; i < num_msgs; i++) {
         ASSERT_NO_THROW(ncr = NameChangeRequest::fromJSON(valid_msgs[i]));
         EXPECT_NO_THROW(sender.sendRequest(ncr));
         // Verify that the queue count increments in step with each send.
         EXPECT_EQ(i+1, sender.getQueueSize());
+
+        // Verify that peekAt(i) returns the NCR we just added.
+        ASSERT_NO_THROW(ncr2 = sender.peekAt(i));
+        ASSERT_TRUE(ncr2);
+        EXPECT_TRUE(*ncr == *ncr2);
     }
 
+    // Verify that attempting to peek beyond the end of the queue, throws.
+    ASSERT_THROW(sender.peekAt(sender.getQueueSize()+1), NcrSenderError);
+
     // Verify that attempting to send an additional message results in a
     // queue full exception.
     EXPECT_THROW(sender.sendRequest(ncr), NcrSenderQueueFull);
 
-    // Loop for the number of valid messages and invoke IOService::run_one.
-    // This should send exactly one message and the queue count should
-    // decrement accordingly.
+    // Loop for the number of valid messages. So long as there is at least
+    // on NCR in the queue, select-fd indicate ready to read. Invoke
+    // 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.
+        ASSERT_TRUE(selectCheck(select_fd) > 0);
+
+        // Execute at one ready handler.
         io_service.run_one();
+
         // 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));
+
     // Verify that the queue is empty.
     EXPECT_EQ(0, sender.getQueueSize());
 
@@ -395,6 +435,111 @@ TEST(NameChangeUDPSenderBasicTest, basicSendTests) {
     EXPECT_EQ(0, sender.getQueueSize());
 }
 
+/// @brief Tests NameChangeUDPSender basic send  with INADDR_ANY and port 0.
+TEST(NameChangeUDPSenderBasicTest, anyAddressSend) {
+    isc::asiolink::IOAddress ip_address(TEST_ADDRESS);
+    isc::asiolink::IOAddress any_address("0.0.0.0");
+    isc::asiolink::IOService io_service;
+    SimpleSendHandler ncr_handler;
+
+    // Tests are based on a list of messages, get the count now.
+    int num_msgs = sizeof(valid_msgs)/sizeof(char*);
+
+    // Create the sender, setting the queue max equal to the number of
+    // messages we will have in the list.
+    NameChangeUDPSender sender(any_address, 0, ip_address, LISTENER_PORT,
+                               FMT_JSON, ncr_handler, num_msgs);
+
+    // Enter send mode.
+    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 that sender shows no IO ready.
+    // and that the queue is empty.
+    EXPECT_EQ(0, selectCheck(select_fd));
+    EXPECT_EQ(0, sender.getQueueSize());
+}
+
+/// @brief Test the NameChangeSender::assumeQueue method.
+TEST(NameChangeSender, assumeQueue) {
+    isc::asiolink::IOAddress ip_address(TEST_ADDRESS);
+    uint32_t port = SENDER_PORT;
+    isc::asiolink::IOService io_service;
+    SimpleSendHandler ncr_handler;
+    NameChangeRequestPtr ncr;
+
+    // Tests are based on a list of messages, get the count now.
+    int num_msgs = sizeof(valid_msgs)/sizeof(char*);
+
+    // Create two senders with queue max equal to the number of
+    // messages we will have in the list.
+    NameChangeUDPSender sender1(ip_address, port, ip_address, port,
+                               FMT_JSON, ncr_handler, num_msgs);
+
+    NameChangeUDPSender sender2(ip_address, port+1, ip_address, port,
+                                FMT_JSON, ncr_handler, num_msgs);
+
+    // Place sender1 into send mode and queue up messages.
+    ASSERT_NO_THROW(sender1.startSending(io_service));
+    for (int i = 0; i < num_msgs; i++) {
+        ASSERT_NO_THROW(ncr = NameChangeRequest::fromJSON(valid_msgs[i]));
+        ASSERT_NO_THROW(sender1.sendRequest(ncr));
+    }
+
+    // Make sure sender1's queue count is as expected.
+    ASSERT_EQ(num_msgs, sender1.getQueueSize());
+
+    // Verify sender1 is sending, sender2 is not.
+    ASSERT_TRUE(sender1.amSending());
+    ASSERT_FALSE(sender2.amSending());
+
+    // Transfer from sender1 to sender2 should fail because
+    // sender1 is in send mode.
+    ASSERT_THROW(sender2.assumeQueue(sender1), NcrSenderError);
+
+    // Take sender1 out of send mode.
+    ASSERT_NO_THROW(sender1.stopSending());
+    ASSERT_FALSE(sender1.amSending());
+
+    // Transfer should succeed. Verify sender1 has none,
+    // and sender2 has num_msgs queued.
+    EXPECT_NO_THROW(sender2.assumeQueue(sender1));
+    EXPECT_EQ(0, sender1.getQueueSize());
+    EXPECT_EQ(num_msgs, sender2.getQueueSize());
+
+    // Reduce sender1's max queue size.
+    ASSERT_NO_THROW(sender1.setQueueMaxSize(num_msgs - 1));
+
+    // Transfer should fail as sender1's queue is not large enough.
+    ASSERT_THROW(sender1.assumeQueue(sender2), NcrSenderError);
+
+    // Place sender1 into send mode and queue up a message.
+    ASSERT_NO_THROW(sender1.startSending(io_service));
+    ASSERT_NO_THROW(ncr = NameChangeRequest::fromJSON(valid_msgs[0]));
+    ASSERT_NO_THROW(sender1.sendRequest(ncr));
+
+    // Take sender1 out of send mode.
+    ASSERT_NO_THROW(sender1.stopSending());
+
+    // Try to transfer from sender1 to sender2. This should fail
+    // as sender2's queue is not empty.
+    ASSERT_THROW(sender2.assumeQueue(sender1), NcrSenderError);
+}
+
 /// @brief Text fixture that allows testing a listener and sender together
 /// It derives from both the receive and send handler classes and contains
 /// and instance of UDP listener and UDP sender.
@@ -422,9 +567,9 @@ public:
                                       *this, true));
 
         // Create our sender instance. Note that reuse_address is true.
-        sender_.reset(
-            new NameChangeUDPSender(addr, SENDER_PORT, addr, LISTENER_PORT,
-                                    FMT_JSON, *this, 100, true));
+         sender_.reset(
+             new NameChangeUDPSender(addr, SENDER_PORT, addr, LISTENER_PORT,
+                                     FMT_JSON, *this, 100, true));
 
         // Set the test timeout to break any running tasks if they hang.
         test_timer_.setup(boost::bind(&NameChangeUDPTest::testTimeoutHandler,
@@ -515,4 +660,123 @@ TEST_F (NameChangeUDPTest, roundTripTest) {
     EXPECT_FALSE(sender_->amSending());
 }
 
+// Tests error handling of a failure to mark the watch socket ready, when
+// sendRequestt() is called.
+TEST(NameChangeUDPSenderBasicTest, watchClosedBeforeSendRequest) {
+    isc::asiolink::IOAddress ip_address(TEST_ADDRESS);
+    isc::asiolink::IOService io_service;
+    SimpleSendHandler ncr_handler;
+
+    // Create the sender and put into send mode.
+    NameChangeUDPSender sender(ip_address, 0, ip_address, LISTENER_PORT,
+                               FMT_JSON, ncr_handler, 100, true);
+    ASSERT_NO_THROW(sender.startSending(io_service));
+    ASSERT_TRUE(sender.amSending());
+
+    // Create an NCR.
+    NameChangeRequestPtr ncr;
+    ASSERT_NO_THROW(ncr = NameChangeRequest::fromJSON(valid_msgs[0]));
+
+    // Tamper with the watch socket by closing the select-fd.
+    close(sender.getSelectFd());
+
+    // Send should fail as we interferred by closing the select-fd.
+    ASSERT_THROW(sender.sendRequest(ncr), WatchSocketError);
+
+    // Verify we didn't invoke the handler.
+    EXPECT_EQ(0, ncr_handler.pass_count_);
+    EXPECT_EQ(0, ncr_handler.error_count_);
+
+    // Request remains in the queue. Technically it was sent but its
+    // completion handler won't get called.
+    EXPECT_EQ(1, sender.getQueueSize());
+}
+
+// Tests error handling of a failure to mark the watch socket ready, when
+// sendNext() is called during completion handling.
+TEST(NameChangeUDPSenderBasicTest, watchClosedAfterSendRequest) {
+    isc::asiolink::IOAddress ip_address(TEST_ADDRESS);
+    isc::asiolink::IOService io_service;
+    SimpleSendHandler ncr_handler;
+
+    // Create the sender and put into send mode.
+    NameChangeUDPSender sender(ip_address, 0, ip_address, LISTENER_PORT,
+                               FMT_JSON, ncr_handler, 100, true);
+    ASSERT_NO_THROW(sender.startSending(io_service));
+    ASSERT_TRUE(sender.amSending());
+
+    // Build and queue up 2 messages.  No handlers will get called yet.
+    for (int i = 0; i < 2; i++) {
+        NameChangeRequestPtr ncr;
+        ASSERT_NO_THROW(ncr = NameChangeRequest::fromJSON(valid_msgs[i]));
+        sender.sendRequest(ncr);
+        EXPECT_EQ(i+1, sender.getQueueSize());
+    }
+
+    // Tamper with the watch socket by closing the select-fd.
+    close (sender.getSelectFd());
+
+    // 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());
+
+    // Verify handler got called twice. First request should have be sent
+    // without error, second call should have failed to send due to watch
+    // socket markReady failure.
+    EXPECT_EQ(1, ncr_handler.pass_count_);
+    EXPECT_EQ(1, ncr_handler.error_count_);
+
+    // The second request should still be in the queue.
+    EXPECT_EQ(1, sender.getQueueSize());
+}
+
+// Tests error handling of a failure to clear the watch socket during
+// completion handling.
+TEST(NameChangeUDPSenderBasicTest, watchSocketBadRead) {
+    isc::asiolink::IOAddress ip_address(TEST_ADDRESS);
+    isc::asiolink::IOService io_service;
+    SimpleSendHandler ncr_handler;
+
+    // Create the sender and put into send mode.
+    NameChangeUDPSender sender(ip_address, 0, ip_address, LISTENER_PORT,
+                               FMT_JSON, ncr_handler, 100, true);
+    ASSERT_NO_THROW(sender.startSending(io_service));
+    ASSERT_TRUE(sender.amSending());
+
+    // Build and queue up 2 messages.  No handlers will get called yet.
+    for (int i = 0; i < 2; i++) {
+        NameChangeRequestPtr ncr;
+        ASSERT_NO_THROW(ncr = NameChangeRequest::fromJSON(valid_msgs[i]));
+        sender.sendRequest(ncr);
+        EXPECT_EQ(i+1, sender.getQueueSize());
+    }
+
+    // Fetch the sender's select-fd.
+    int select_fd = sender.getSelectFd();
+
+    // Verify that select_fd appears ready.
+    ASSERT_TRUE(selectCheck(select_fd) > 0);
+
+    // Interfere by reading part of the marker from the select-fd.
+    uint32_t buf = 0;
+    ASSERT_EQ((read (select_fd, &buf, 1)), 1);
+    ASSERT_NE(WatchSocket::MARKER, buf);
+
+    // Run one handler. This should execute the send completion handler
+    // 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());
+
+    // Verify handler got called twice. First request should have be sent
+    // without error, second call should have failed to send due to watch
+    // socket markReady failure.
+    EXPECT_EQ(1, ncr_handler.pass_count_);
+    EXPECT_EQ(1, ncr_handler.error_count_);
+
+    // The second request should still be in the queue.
+    EXPECT_EQ(1, sender.getQueueSize());
+}
+
 } // end of anonymous namespace

src/lib/dhcp_ddns/tests/test_utils.cc

@@ -0,0 +1,43 @@
+// Copyright (C) 2014  Internet Systems Consortium, Inc. ("ISC")
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
+// OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+#include <gtest/gtest.h>
+
+#include <sys/select.h>
+#include <sys/ioctl.h>
+
+using namespace std;
+
+namespace isc {
+namespace dhcp_ddns {
+
+int selectCheck(int fd_to_check) {
+    fd_set read_fds;
+    int maxfd = 0;
+
+    FD_ZERO(&read_fds);
+
+    // Add this socket to listening set
+    FD_SET(fd_to_check,  &read_fds);
+    maxfd = fd_to_check;
+
+    struct timeval select_timeout;
+    select_timeout.tv_sec = 0;
+    select_timeout.tv_usec = 0;
+
+    return (select(maxfd + 1, &read_fds, NULL, NULL, &select_timeout));
+}
+
+}; // namespace isc::d2
+}; // namespace isc

src/lib/dhcp_ddns/tests/test_utils.h

@@ -0,0 +1,37 @@
+// Copyright (C) 2014  Internet Systems Consortium, Inc. ("ISC")
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
+// OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+#ifndef TEST_UTILS_H
+#define TEST_UTILS_H
+
+/// @file test_utils.h Common dhcp_ddns testing elements
+
+#include <gtest/gtest.h>
+
+
+namespace isc {
+namespace dhcp_ddns {
+
+/// @brief Returns the result of select() given an fd to check for read status.
+///
+/// @param fd_to_check The file descriptor to test
+///
+/// @return Returns less than one on an error, 0 if the fd is not ready to
+/// read, > 0 if it is ready to read. 
+int selectCheck(int fd_to_check);
+
+}; // namespace isc::dhcp_ddns;
+}; // namespace isc;
+
+#endif 

src/lib/dhcp_ddns/tests/watch_socket_unittests.cc

@@ -0,0 +1,207 @@
+// Copyright (C) 2014  Internet Systems Consortium, Inc. ("ISC")
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
+// OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+#include <dhcp_ddns/watch_socket.h>
+#include <test_utils.h>
+
+#include <gtest/gtest.h>
+
+#include <sys/select.h>
+#include <sys/ioctl.h>
+
+using namespace std;
+using namespace isc;
+using namespace isc::dhcp_ddns;
+
+namespace {
+
+/// @brief Tests the basic functionality of WatchSocket.
+TEST(WatchSocketTest, basics) {
+    WatchSocketPtr watch;
+
+    /// Verify that we can construct a WatchSocket.
+    ASSERT_NO_THROW(watch.reset(new WatchSocket()));
+    ASSERT_TRUE(watch);
+
+    /// Verify that post-construction the state the select-fd is valid.
+    int select_fd = watch->getSelectFd();
+    EXPECT_NE(select_fd, WatchSocket::INVALID_SOCKET);
+
+    /// Verify that isReady() is false and that a call to select agrees.
+    EXPECT_FALSE(watch->isReady());
+    EXPECT_EQ(0, selectCheck(select_fd));
+
+    /// Verify that the socket can be marked ready.
+    ASSERT_NO_THROW(watch->markReady());
+
+    /// Verify that we have exactly one marker waiting to be read.
+    int count = 0;
+    EXPECT_FALSE(ioctl(select_fd, FIONREAD, &count));
+    EXPECT_EQ(sizeof(WatchSocket::MARKER), count);
+
+    /// Verify that we can call markReady again without error.
+    ASSERT_NO_THROW(watch->markReady());
+
+    /// Verify that we STILL have exactly one marker waiting to be read.
+    EXPECT_FALSE(ioctl(select_fd, FIONREAD, &count));
+    EXPECT_EQ(sizeof(WatchSocket::MARKER), count);
+
+    /// Verify that isReady() is true and that a call to select agrees.
+    EXPECT_TRUE(watch->isReady());
+    EXPECT_EQ(1, selectCheck(select_fd));
+
+    /// Verify that the socket can be cleared.
+    ASSERT_NO_THROW(watch->clearReady());
+
+    /// Verify that isReady() is false and that a call to select agrees.
+    EXPECT_FALSE(watch->isReady());
+    EXPECT_EQ(0, selectCheck(select_fd));
+}
+
+/// @brief Checks behavior when select_fd is closed externally while in the
+/// "cleared" state.
+TEST(WatchSocketTest, closedWhileClear) {
+    WatchSocketPtr watch;
+
+    /// Verify that we can construct a WatchSocket.
+    ASSERT_NO_THROW(watch.reset(new WatchSocket()));
+    ASSERT_TRUE(watch);
+
+    /// Verify that post-construction the state the select-fd is valid.
+    int select_fd = watch->getSelectFd();
+    ASSERT_NE(select_fd, WatchSocket::INVALID_SOCKET);
+
+    // Verify that socket does not appear ready.
+    ASSERT_EQ(0, watch->isReady());
+
+    // Interfere by closing the fd.
+    ASSERT_EQ(0, close(select_fd));
+
+    // Verify that socket does not appear ready.
+    ASSERT_EQ(0, watch->isReady());
+
+    // Verify that clear does NOT throw.
+    ASSERT_NO_THROW(watch->clearReady());
+
+    // Verify that trying to mark it fails.
+    ASSERT_THROW(watch->markReady(), WatchSocketError);
+
+    // Verify that clear does NOT throw.
+    ASSERT_NO_THROW(watch->clearReady());
+
+    // Verify that getSelectFd() returns invalid socket.
+    ASSERT_EQ(WatchSocket::INVALID_SOCKET, watch->getSelectFd());
+}
+
+/// @brief Checks behavior when select_fd has closed while in the "ready"
+/// state.
+TEST(WatchSocketTest, closedWhileReady) {
+    WatchSocketPtr watch;
+
+    /// Verify that we can construct a WatchSocket.
+    ASSERT_NO_THROW(watch.reset(new WatchSocket()));
+    ASSERT_TRUE(watch);
+
+    /// Verify that post-construction the state the select-fd is valid.
+    int select_fd = watch->getSelectFd();
+    ASSERT_NE(select_fd, WatchSocket::INVALID_SOCKET);
+
+    /// Verify that the socket can be marked ready.
+    ASSERT_NO_THROW(watch->markReady());
+    EXPECT_EQ(1, selectCheck(select_fd));
+
+    // Interfere by closing the fd.
+    ASSERT_EQ(0, close(select_fd));
+
+    // Verify that trying to clear it does not throw.
+    ASSERT_NO_THROW(watch->clearReady());
+
+    // Verify the select_fd fails as socket is invalid/closed.
+    EXPECT_EQ(-1, selectCheck(select_fd));
+
+    // Verify that subsequent attempts to mark it will fail.
+    ASSERT_THROW(watch->markReady(), WatchSocketError);
+}
+
+/// @brief Checks behavior when select_fd has been marked ready but then
+/// emptied by an external read.
+TEST(WatchSocketTest, emptyReadySelectFd) {
+    WatchSocketPtr watch;
+
+    /// Verify that we can construct a WatchSocket.
+    ASSERT_NO_THROW(watch.reset(new WatchSocket()));
+    ASSERT_TRUE(watch);
+
+    /// Verify that post-construction the state the select-fd is valid.
+    int select_fd = watch->getSelectFd();
+    ASSERT_NE(select_fd, WatchSocket::INVALID_SOCKET);
+
+    /// Verify that the socket can be marked ready.
+    ASSERT_NO_THROW(watch->markReady());
+    EXPECT_EQ(1, selectCheck(select_fd));
+
+    // Interfere by reading the fd. This should empty the read pipe.
+    uint32_t buf = 0;
+    ASSERT_EQ((read (select_fd, &buf, sizeof(buf))), sizeof(buf));
+    ASSERT_EQ(WatchSocket::MARKER, buf);
+
+    // Really nothing that can be done to protect against this, but let's
+    // make sure we aren't in a weird state.
+    ASSERT_NO_THROW(watch->clearReady());
+
+    // Verify the select_fd fails as socket is invalid/closed.
+    EXPECT_EQ(0, selectCheck(select_fd));
+
+    // Verify that getSelectFd() returns is still good.
+    ASSERT_EQ(select_fd, watch->getSelectFd());
+}
+
+/// @brief Checks behavior when select_fd has been marked ready but then
+/// contents have been "corrupted" by a partial read.
+TEST(WatchSocketTest, badReadOnClear) {
+    WatchSocketPtr watch;
+
+    /// Verify that we can construct a WatchSocket.
+    ASSERT_NO_THROW(watch.reset(new WatchSocket()));
+    ASSERT_TRUE(watch);
+
+    /// Verify that post-construction the state the select-fd is valid.
+    int select_fd = watch->getSelectFd();
+    ASSERT_NE(select_fd, WatchSocket::INVALID_SOCKET);
+
+    /// Verify that the socket can be marked ready.
+    ASSERT_NO_THROW(watch->markReady());
+    EXPECT_EQ(1, selectCheck(select_fd));
+
+    // Interfere by reading the fd. This should empty the read pipe.
+    uint32_t buf = 0;
+    ASSERT_EQ((read (select_fd, &buf, 1)), 1);
+    ASSERT_NE(WatchSocket::MARKER, buf);
+
+    // Really nothing that can be done to protect against this, but let's
+    // make sure we aren't in a weird state.
+    /// @todo maybe clear should never throw, log only
+    ASSERT_THROW(watch->clearReady(), WatchSocketError);
+
+    // Verify the select_fd fails as socket is invalid/closed.
+    EXPECT_EQ(-1, selectCheck(select_fd));
+
+    // Verify that getSelectFd() returns INVALID.
+    ASSERT_EQ(WatchSocket::INVALID_SOCKET, watch->getSelectFd());
+
+    // Verify that subsequent attempt to mark it fails.
+    ASSERT_THROW(watch->markReady(), WatchSocketError);
+}
+
+} // end of anonymous namespace

src/lib/dhcp_ddns/watch_socket.cc

@@ -0,0 +1,152 @@
+// Copyright (C) 2014 Internet Systems Consortium, Inc. ("ISC")
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
+// OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+/// @file watch_socket.cc
+
+#include <dhcp_ddns/dhcp_ddns_log.h>
+#include <dhcp_ddns/watch_socket.h>
+
+#include <fcntl.h>
+#include <errno.h>
+#include <sys/select.h>
+
+namespace isc {
+namespace dhcp_ddns {
+
+
+const int WatchSocket::INVALID_SOCKET;
+const uint32_t WatchSocket::MARKER;
+
+WatchSocket::WatchSocket()
+    : source_(INVALID_SOCKET), sink_(INVALID_SOCKET) {
+    // Open the pipe.
+    int fds[2];
+    if (pipe(fds)) {
+        const char* errstr = strerror(errno);
+        isc_throw(WatchSocketError, "Cannot construct pipe: " << errstr);
+    }
+
+    source_ = fds[1];
+    sink_ = fds[0];
+
+    if (fcntl(sink_, F_SETFL, O_NONBLOCK)) {
+        const char* errstr = strerror(errno);
+        isc_throw(WatchSocketError, "Cannot set sink to non-blocking: "
+                                     << errstr);
+    }
+}
+
+WatchSocket::~WatchSocket() {
+    closeSocket();
+}
+
+void
+WatchSocket::markReady() {
+    // Make sure it hasn't been orphaned!  Otherwise we may get SIGPIPE.  We
+    // use fcntl to check as select() on some systems may show it as ready to
+    // read.
+    if (fcntl(sink_, F_GETFL) < 0) {
+        closeSocket();
+        isc_throw(WatchSocketError, "WatchSocket markReady failed:"
+                  " select_fd was closed!");
+    }
+
+    if (!isReady()) {
+        int nbytes = write (source_, &MARKER, sizeof(MARKER));
+        if (nbytes != sizeof(MARKER)) {
+            // If there's an error get the error message than close
+            // the pipe.  This should ensure any further use of the socket
+            // or testing the fd with select_fd will fail.
+            const char* errstr = strerror(errno);
+            closeSocket();
+            isc_throw(WatchSocketError, "WatchSocket markReady failed:"
+                      << " bytes written: " << nbytes << " : " << errstr);
+        }
+    }
+}
+
+bool
+WatchSocket::isReady() {
+    // Report it as not ready rather than error here.
+    if (sink_ == INVALID_SOCKET) {
+        return (false);
+    }
+
+    fd_set read_fds;
+    FD_ZERO(&read_fds);
+
+    // Add select_fd socket to listening set
+    FD_SET(sink_,  &read_fds);
+
+    // Set zero timeout (non-blocking).
+    struct timeval select_timeout;
+    select_timeout.tv_sec = 0;
+    select_timeout.tv_usec = 0;
+
+    // Return true only if read ready, treat error same as not ready.
+    return (select(sink_ + 1, &read_fds, NULL, NULL, &select_timeout) > 0);
+}
+
+void
+WatchSocket::clearReady() {
+    if (isReady()) {
+        uint32_t buf = 0;
+        int nbytes = read (sink_, &buf, sizeof(buf));
+        if ((nbytes != sizeof(MARKER) || (buf != MARKER))) {
+            // If there's an error get the error message than close
+            // the pipe.  This should ensure any further use of the socket
+            // or testing the fd with select_fd will fail.
+            const char* errstr = strerror(errno);
+            closeSocket();
+            isc_throw(WatchSocketError, "WatchSocket clearReady failed:"
+                      << " bytes read: " << nbytes << " : "
+                      << " value read: " << buf << " error :" <<errstr);
+        }
+    }
+}
+
+void
+WatchSocket::closeSocket() {
+    // Close the pipe fds.  Technically a close can fail (hugely unlikely)
+    // but there's no recovery for it either.  If one does fail we log it
+    // and go on. Plus this is called by the destructor and no one likes
+    // destructors that throw.
+    if (source_ != INVALID_SOCKET) {
+        if (close(source_)) {
+            const char* errstr = strerror(errno);
+            LOG_ERROR(dhcp_ddns_logger, DHCP_DDNS_WATCH_SOURCE_CLOSE_ERROR)
+                      .arg(errstr);
+        }
+
+        source_ = INVALID_SOCKET;
+    }
+
+    if (sink_ != INVALID_SOCKET) {
+        if (close(sink_)) {
+            const char* errstr = strerror(errno);
+            LOG_ERROR(dhcp_ddns_logger, DHCP_DDNS_WATCH_SINK_CLOSE_ERROR)
+                      .arg(errstr);
+        }
+
+        sink_ = INVALID_SOCKET;
+    }
+}
+
+int
+WatchSocket::getSelectFd() {
+    return (sink_);
+}
+
+} // namespace isc::dhcp_ddns
+} // namespace isc

src/lib/dhcp_ddns/watch_socket.h

@@ -0,0 +1,138 @@
+// Copyright (C) 2014 Internet Systems Consortium, Inc. ("ISC")
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
+// OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+#ifndef WATCH_SOCKET_H
+#define WATCH_SOCKET_H
+
+/// @file watch_socket.h Defines the class, WatchSocket.
+
+#include <exceptions/exceptions.h>
+
+#include <boost/shared_ptr.hpp>
+
+#include <stdint.h>
+
+namespace isc {
+namespace dhcp_ddns {
+
+/// @brief Exception thrown if an error occurs during IO source open.
+class WatchSocketError : public isc::Exception {
+public:
+    WatchSocketError(const char* file, size_t line, const char* what) :
+        isc::Exception(file, line, what) { };
+};
+
+/// @brief Provides an IO "ready" semaphore for use with select() or poll()
+/// WatchSocket exposes a single open file descriptor, the "select-fd" which
+/// can be marked as being ready to read (i.e. !EWOULDBLOCK) and cleared
+/// (i.e. EWOULDBLOCK).  The select-fd can be used with select(), poll(), or
+/// their variants alongside other file descriptors.
+///
+/// Internally, WatchSocket uses a pipe.  The select-fd is the "read" end of
+/// pipe.  To mark the socket as ready to read, an integer marker is written
+/// to the pipe.  To clear the socket, the marker is read from the pipe.  Note
+/// that WatchSocket will only write the marker if it is not already marked.
+/// This prevents the socket's pipe from filling endlessly.
+///
+/// @warning Because the read or "sink" side of the pipe is used as the select_fd,
+/// it is possible for that fd to be interfered with, albeit only from within the
+/// process space which owns it.  Performing operations that may alter the fd's state
+/// such as close, read, or altering behavior flags with fcntl or ioctl can have
+/// unpredictable results.  It is intended strictly use with functions such as select()
+/// poll() or their variants.
+class WatchSocket {
+public:
+    /// @brief Value used to signify an invalid descriptor.
+    static const int INVALID_SOCKET = -1;
+    /// @brief Value written to the source when marking the socket as ready.
+    /// The value itself is arbitrarily chosen as one that is unlikely to occur
+    /// otherwise and easy to debug.
+    static const uint32_t MARKER = 0xDEADBEEF;
+
+    /// @brief Constructor
+    ///
+    /// Constructs an instance of the WatchSocket in the cleared (EWOULDBLOCK)
+    /// state.
+    WatchSocket();
+
+    /// @brief Destructor
+    ///
+    /// Closes all internal resources, including the select-fd.
+    virtual ~WatchSocket();
+
+    /// @brief Marks the select-fd as ready to read.
+    ///
+    /// Marks the socket as ready to read, if is not already so marked.
+    /// If an error occurs, closeSocket is called. This will force any further
+    /// use of the select_fd to fail rather than show the fd as READY.  Such
+    /// an error is almost surely a programmatic error which has corrupted the
+    /// select_fd.
+    ///
+    /// @throw WatchSocketError if an error occurs marking the socket.
+    void markReady();
+
+    /// @brief Returns true the if socket is marked as ready.
+    ///
+    /// This method uses a non-blocking call to  select() to test read state of the
+    /// select_fd.  Rather than track what the status "should be" it tests the status.
+    /// This should eliminate conditions where the select-fd appear to be perpetually
+    /// ready.
+    /// @return  Returns true if select_fd is not INVALID_SOCKET and select() reports it
+    /// as !EWOULDBLOCK, otherwise it returns false.
+    /// This method is guaranteed NOT to throw.
+    bool isReady();
+
+    /// @brief Clears the socket's ready to read marker.
+    ///
+    /// Clears the socket if it is currently marked as ready to read.
+    /// If an error occurs, closeSocket is called. This will force any further
+    /// use of the select_fd to fail rather than show the fd as READY.  Such
+    /// an error is almost surely a programmatic error which has corrupted the
+    /// select_fd.
+    ///
+    /// @throw WatchSocketError if an error occurs clearing the socket
+    /// marker.
+    void clearReady();
+
+    /// @brief Returns the file descriptor to use to monitor the socket.
+    ///
+    /// @note Using this file descriptor as anything other than an argument
+    /// to select() or similar methods can have unpredictable results.
+    ///
+    /// @return The file descriptor associated with read end of the socket's
+    /// pipe.
+    int getSelectFd();
+
+private:
+    /// @brief Closes the descriptors associated with the socket.
+    ///
+    /// Used internally in the destructor and if an error occurs marking or
+    /// clearing the socket.
+    void closeSocket();
+
+    /// @brief The end of the pipe to which the marker is written
+    int source_;
+
+    /// @brief The end of the pipe from which the marker is read.
+    /// This is the value returned as the select-fd.
+    int sink_;
+};
+
+/// @brief Defines a smart pointer to an instance of a WatchSocket.
+typedef boost::shared_ptr<WatchSocket> WatchSocketPtr;
+
+} // namespace isc::dhcp_ddns
+} // namespace isc
+
+#endif

src/lib/dhcpsrv/d2_client.cc

@@ -12,6 +12,7 @@
 // OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
 // PERFORMANCE OF THIS SOFTWARE.
 
+#include <dhcp_ddns/ncr_udp.h>
 #include <dhcpsrv/d2_client.h>
 #include <dhcpsrv/dhcpsrv_log.h>
 
@@ -22,6 +23,8 @@ using namespace std;
 namespace isc {
 namespace dhcp {
 
+//***************************** D2ClientConfig ********************************
+
 D2ClientConfig::D2ClientConfig(const  bool enable_updates,
                                const isc::asiolink::IOAddress& server_ip,
                                const size_t server_port,
@@ -135,7 +138,12 @@ operator<<(std::ostream& os, const D2ClientConfig& config) {
     return (os);
 }
 
-D2ClientMgr::D2ClientMgr() : d2_client_config_(new D2ClientConfig()) {
+
+//******************************** D2ClientMgr ********************************
+
+
+D2ClientMgr::D2ClientMgr() : d2_client_config_(new D2ClientConfig()),
+    name_change_sender_(), private_io_service_(), sender_io_service_(NULL) {
     // Default constructor initializes with a disabled configuration.
 }
 
@@ -149,16 +157,57 @@ D2ClientMgr::setD2ClientConfig(D2ClientConfigPtr& new_config) {
                   "D2ClientMgr cannot set DHCP-DDNS configuration to NULL.");
     }
 
-    // @todo When NameChangeSender is integrated, we will need to handle these
-    // scenarios:
-    // 1. D2 was enabled but now it is disabled
-    //     - destroy the sender, flush any queued
-    // 2. D2 is still enabled but server parameters have changed
-    //     - preserve any queued,  reconnect based on sender parameters
-    // 3. D2 was was disabled now it is enabled.
-    //     - create sender
-    //
-    // For now we just update the configuration.
+    // Don't do anything unless configuration values are actually different.
+    if (*d2_client_config_ != *new_config) {
+        if (!new_config->getEnableUpdates()) {
+            // Updating has been turned off, destroy current sender.
+            // Any queued requests are tossed.
+            name_change_sender_.reset();
+        } else {
+            dhcp_ddns::NameChangeSenderPtr new_sender;
+            switch (new_config->getNcrProtocol()) {
+            case dhcp_ddns::NCR_UDP: {
+                /// @todo Should we be able to configure a sender's client
+                /// side ip and port?  We should certainly be able to
+                /// configure a maximum queue size.  These were overlooked
+                /// but are covered in Trac# 3328.
+                isc::asiolink::IOAddress any_addr("0.0.0.0");
+                uint32_t any_port = 0;
+                uint32_t queue_max = 1024;
+
+                // Instantiate a new sender.
+                new_sender.reset(new dhcp_ddns::NameChangeUDPSender(
+                                                any_addr, any_port,
+                                                new_config->getServerIp(),
+                                                new_config->getServerPort(),
+                                                new_config->getNcrFormat(),
+                                                *this, queue_max));
+                break;
+                }
+            default:
+                // In theory you can't get here.
+                isc_throw(D2ClientError, "Invalid sender Protocol: "
+                          << new_config->getNcrProtocol());
+                break;
+            }
+
+            // Transfer queued requests from previous sender to the new one.
+            /// @todo - Should we consider anything queued to be wrong?
+            /// If only server values changed content might still be right but
+            /// if content values changed (e.g. suffix or an override flag)
+            /// then the queued contents might now be invalid.  There is
+            /// no way to regenerate them if they are wrong.
+            if (name_change_sender_) {
+                name_change_sender_->stopSending();
+                new_sender->assumeQueue(*name_change_sender_);
+            }
+
+            // Replace the old sender with the new one.
+            name_change_sender_ = new_sender;
+        }
+    }
+
+    // Update the configuration.
     d2_client_config_ = new_config;
     LOG_DEBUG(dhcpsrv_logger, DHCPSRV_DBG_TRACE, DHCPSRV_CFGMGR_CFG_DHCP_DDNS)
               .arg(!ddnsEnabled() ? "DHCP-DDNS updates disabled" :
@@ -257,7 +306,141 @@ D2ClientMgr::qualifyName(const std::string& partial_name) const {
     return (gen_name.str());
 }
 
+void
+D2ClientMgr::startSender(D2ClientErrorHandler error_handler) {
+    // Create a our own service instance when we are not being multiplexed
+    // into an external service..
+    private_io_service_.reset(new asiolink::IOService());
+    startSender(error_handler, *private_io_service_);
+}
+
+void
+D2ClientMgr::startSender(D2ClientErrorHandler error_handler,
+                         isc::asiolink::IOService& io_service) {
+    if (!name_change_sender_)  {
+        isc_throw(D2ClientError, "D2ClientMgr::startSender sender is null");
+    }
+
+    if (!error_handler) {
+        isc_throw(D2ClientError, "D2ClientMgr::startSender handler is null");
+    }
+
+    // Set the error handler.
+    client_error_handler_ = error_handler;
+
+    // Remember the io service being used.
+    sender_io_service_ = &io_service;
+
+    // Start the sender on the given service.
+    name_change_sender_->startSending(*sender_io_service_);
+
+    /// @todo need to register sender's select-fd with IfaceMgr once 3315 is
+    /// done.
+}
+
+bool
+D2ClientMgr::amSending() const {
+    return (name_change_sender_ && name_change_sender_->amSending());
+}
 
+void
+D2ClientMgr::stopSender() {
+    if (!name_change_sender_)  {
+        isc_throw(D2ClientError, "D2ClientMgr::stopSender sender is null");
+    }
+
+    /// @todo need to unregister sender's select-fd with IfaceMgr once 3315 is
+    /// done.
+
+    name_change_sender_->stopSending();
+}
+
+void
+D2ClientMgr::sendRequest(dhcp_ddns::NameChangeRequestPtr& ncr) {
+    if (!name_change_sender_) {
+        isc_throw(D2ClientError, "D2ClientMgr::sendRequest sender is null");
+    }
+
+    name_change_sender_->sendRequest(ncr);
+}
+
+size_t
+D2ClientMgr::getQueueSize() const {
+    if (!name_change_sender_) {
+        isc_throw(D2ClientError, "D2ClientMgr::getQueueSize sender is null");
+    }
+
+    return(name_change_sender_->getQueueSize());
+}
+
+
+const dhcp_ddns::NameChangeRequestPtr&
+D2ClientMgr::peekAt(const size_t index) const {
+    if (!name_change_sender_) {
+        isc_throw(D2ClientError, "D2ClientMgr::peekAt sender is null");
+    }
+
+    return (name_change_sender_->peekAt(index));
+}
+
+void
+D2ClientMgr::clearQueue() {
+    if (!name_change_sender_) {
+        isc_throw(D2ClientError, "D2ClientMgr::clearQueue sender is null");
+    }
+
+    name_change_sender_->clearSendQueue();
+}
+
+void
+D2ClientMgr::operator()(const dhcp_ddns::NameChangeSender::Result result,
+                        dhcp_ddns::NameChangeRequestPtr& ncr) {
+    if (result == dhcp_ddns::NameChangeSender::SUCCESS) {
+        LOG_DEBUG(dhcpsrv_logger, DHCPSRV_DBG_TRACE_DETAIL,
+                  DHCPSRV_DHCP_DDNS_NCR_SENT).arg(ncr->toText());
+    } else {
+        // Handler is mandatory but test it just to be safe.
+        /// @todo Until we have a better feel for how errors need to be
+        /// handled we farm it out to the application layer.
+        if (client_error_handler_) {
+            // Handler is not supposed to throw, but catch just in case.
+            try {
+                (client_error_handler_)(result, ncr);
+            } catch (const std::exception& ex) {
+                LOG_ERROR(dhcpsrv_logger, DHCPSRV_DHCP_DDNS_ERROR_EXCEPTION)
+                          .arg(ex.what());
+            }
+        } else {
+            LOG_ERROR(dhcpsrv_logger, DHCPSRV_DHCP_DDNS_HANDLER_NULL);
+        }
+   }
+}
+
+int
+D2ClientMgr::getSelectFd() {
+    if (!amSending()) {
+        isc_throw (D2ClientError, "D2ClientMgr::getSelectFd "
+                   " not in send mode");
+    }
+
+    return (name_change_sender_->getSelectFd());
+}
+
+void
+D2ClientMgr::runReadyIO() {
+    if (!sender_io_service_) {
+        // This should never happen.
+        isc_throw(D2ClientError, "D2ClientMgr::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().
+    sender_io_service_->get_io_service().poll();
+}
 
 };  // namespace dhcp
+
 };  // namespace isc

src/lib/dhcpsrv/d2_client.h

@@ -211,14 +211,54 @@ operator<<(std::ostream& os, const D2ClientConfig& config);
 /// @brief Defines a pointer for D2ClientConfig instances.
 typedef boost::shared_ptr<D2ClientConfig> D2ClientConfigPtr;
 
+/// @brief Defines the type for D2 IO error handler.
+/// This callback is invoked when a send to b10-dhcp-ddns completes with a
+/// failed status.  This provides the application layer (Kea) with a means to
+/// handle the error appropriately.
+///
+/// @param result Result code of the send operation.
+/// @param ncr NameChangeRequest which failed to send.
+///
+/// @note Handlers are expected not to throw. In the event a hanlder does
+/// throw invoking code logs the exception and then swallows it.
+typedef
+boost::function<void(const dhcp_ddns::NameChangeSender::Result result,
+                     dhcp_ddns::NameChangeRequestPtr& ncr)> D2ClientErrorHandler;
+
 /// @brief D2ClientMgr isolates Kea from the details of being a D2 client.
 ///
-/// Provides services for managing the current D2ClientConfig and managing
-/// communications with D2. (@todo The latter will be added once communication
-/// with D2 is implemented through the integration of
-/// dhcp_ddns::NameChangeSender interface(s)).
+/// Provides services for managing the current dhcp-ddns configuration and
+/// as well as communications with b10-dhcp-ddns.  Regarding configuration it
+/// provides services to store, update, and access the current dhcp-ddns
+/// configuration.  As for b10-dhcp-ddns communications, D2ClientMgr creates
+/// maintains a NameChangeSender appropriate to the current configuration and
+/// provides services to start, stop, and post NCRs to the sender.  Additionally
+/// there are methods to examine the queue of requests currently waiting for
+/// transmission.
+///
+/// The manager also provides the mechanics to integrate the ASIO-based IO
+/// used by the NCR IPC with the select-driven IO used by Kea.  Senders expose
+/// a file descriptor, the "select-fd" that can monitored for read-readiness
+/// with the select() function (or variants).  D2ClientMgr provides a method,
+/// runReadyIO(), that will process all ready events on a sender's
+/// IOservice.  Track# 3315 is extending Kea's IfaceMgr to support the
+/// registration of multiple external sockets with callbacks that are then
+/// monitored with IO readiness via select().
+/// @todo D2ClientMgr will be modified to register the sender's select-fd and
+/// runReadyIO() with IfaceMgr when entering the send mode and will
+/// unregister when exiting send mode.
+///
+/// To place the manager in send mode, the calling layer must supply an error
+/// handler and optionally an IOService instance.  The error handler is invoked
+/// if a send completes with a failed status. This provides the calling layer
+/// an opportunity act upon the error.
+///
+/// If the caller supplies an IOService, that service will be used to process
+/// the sender's IO.  If not supplied, D2ClientMgr pass a private IOService
+/// into the sender.  Using a private service isolates the sender's IO from
+/// any other services.
 ///
-class D2ClientMgr {
+class D2ClientMgr : public dhcp_ddns::NameChangeSender::RequestSendHandler {
 public:
     /// @brief Constructor
     ///
@@ -331,9 +371,141 @@ public:
     template <class T>
     void adjustDomainName(const T& fqdn, T& fqdn_resp);
 
+    /// @brief Enables sending NameChangeRequests to b10-dhcp-ddns
+    ///
+    /// Places the NameChangeSender into send mode. This instructs the
+    /// sender to begin dequeuing and transmitting requests and to accept
+    /// additional requests via the sendRequest() method.
+    ///
+    /// @param error_handler application level error handler to cope with
+    /// sends that complete with a failed status.  A valid function must be
+    /// supplied as the manager cannot know how an application should deal
+    /// with send failures.
+    /// @param io_service IOService to be used for sender IO event processing
+    ///
+    /// @throw D2ClientError if sender instance is null. Underlying layer
+    /// may throw NCRSenderExceptions exceptions.
+    void startSender(D2ClientErrorHandler error_handler,
+                     isc::asiolink::IOService& io_service);
+
+    /// @brief Enables sending NameChangeRequests to b10-dhcp-ddns
+    ///
+    /// Places the NameChangeSender into send mode. This instructs the
+    /// sender to begin dequeuing and transmitting requests and to accept
+    /// additional requests via the sendRequest() method.  The manager
+    /// will create a new, private instance of an IOService for the sender
+    /// to use for IO event processing.
+    ///
+    /// @param error_handler application level error handler to cope with
+    /// sends that complete with a failed status.  A valid function must be
+    /// supplied as the manager cannot know how an application should deal
+    /// with send failures.
+    ///
+    /// @throw D2ClientError if sender instance is null. Underlying layer
+    /// may throw NCRSenderExceptions exceptions.
+    void startSender(D2ClientErrorHandler error_handler);
+
+    /// @brief Returns true if the sender is in send mode, false otherwise.
+    ///
+    /// A true value indicates that the sender is present and in accepting
+    /// messages for transmission, false otherwise.
+    bool amSending() const;
+
+    /// @brief Disables sending NameChangeRequests to b10-dhcp-ddns
+    ///
+    /// Takes the NameChangeSender out of send mode.  The sender will stop
+    /// transmitting requests, though any queued requests remain queued.
+    /// Attempts to queue additional requests via sendRequest will fail.
+    ///
+    /// @throw D2ClientError if sender instance is null. Underlying layer
+    /// may throw NCRSenderExceptions exceptions.
+    void stopSender();
+
+    /// @brief Send the given NameChangeRequests to b10-dhcp-ddns
+    ///
+    /// Passes NameChangeRequests to the NCR sender for transmission to
+    /// b10-dhcp-ddns.
+    ///
+    /// @param ncr NameChangeRequest to send
+    ///
+    /// @throw D2ClientError if sender instance is null. Underlying layer
+    /// may throw NCRSenderExceptions exceptions.
+    void sendRequest(dhcp_ddns::NameChangeRequestPtr& ncr);
+
+    /// @brief Returns the number of NCRs queued for transmission.
+    size_t getQueueSize() const;
+
+    /// @brief Returns the nth NCR queued for transmission.
+    ///
+    /// Note that the entry is not removed from the queue.
+    /// @param index the index of the entry in the queue to fetch.
+    /// Valid values are 0 (front of the queue) to (queue size - 1).
+    /// @note This method is for test purposes only.
+    ///
+    /// @return Pointer reference to the queue entry.
+    ///
+    /// @throw D2ClientError if sender instance is null. Underlying layer
+    /// may throw NCRSenderExceptions exceptions.
+    const dhcp_ddns::NameChangeRequestPtr& peekAt(const size_t index) const;
+
+    /// @brief Removes all NCRs queued for transmission.
+    ///
+    /// @throw D2ClientError if sender instance is null. Underlying layer
+    /// may throw NCRSenderExceptions exceptions.
+    void clearQueue();
+
+    /// @brief Processes sender IO events
+    ///
+    /// Runs all handlers ready for execution on the sender's IO service.
+    void runReadyIO();
+
+protected:
+    /// @brief Function operator implementing the NCR sender callback.
+    ///
+    /// This method is invoked each time the NameChangeSender completes
+    /// an asychronous send.
+    ///
+    /// @param result contains that send outcome status.
+    /// @param ncr is a pointer to the NameChangeRequest that was
+    /// delivered (or attempted).
+    ///
+    /// @throw This method MUST NOT throw.
+    virtual void operator ()(const dhcp_ddns::NameChangeSender::Result result,
+                             dhcp_ddns::NameChangeRequestPtr& ncr);
+
+    /// @brief Fetches the sender's select-fd.
+    ///
+    /// The select-fd may be used with select() or poll().  If the sender has
+    /// IO waiting to process, the fd will evaluate as !EWOULDBLOCK.
+    /// @note This is only exposed for testing purposes.
+    ///
+    /// @return The sender's select-fd
+    ///
+    /// @throw D2ClientError if the sender does not exist or is not in send
+    /// mode.
+    int getSelectFd();
+
 private:
     /// @brief Container class for DHCP-DDNS configuration parameters.
     D2ClientConfigPtr d2_client_config_;
+
+    /// @brief Pointer to the current interface to DHCP-DDNS.
+    dhcp_ddns::NameChangeSenderPtr name_change_sender_;
+
+    /// @brief Private IOService to use if calling layer doesn't wish to
+    /// supply one.
+    boost::shared_ptr<asiolink::IOService> private_io_service_;
+
+    /// @brief Application supplied error handler invoked when a send
+    /// completes with a failed status.
+    D2ClientErrorHandler client_error_handler_;
+
+    /// @brief Pointer to the IOService currently being used by the sender.
+    /// @note We need to remember the io_service given to the sender however
+    /// we may have received only a referenece to it from the calling layer.
+    /// Use a raw pointer to store it.  This value should never be exposed
+    /// and is only valid while in send mode.
+    asiolink::IOService* sender_io_service_;
 };
 
 template <class T>

src/lib/dhcpsrv/dhcpsrv_messages.mes

@@ -1,4 +1,4 @@
-# Copyright (C) 2012-2013  Internet Systems Consortium, Inc. ("ISC")
+# Copyright (C) 2012-2014  Internet Systems Consortium, Inc. ("ISC")
 #
 # Permission to use, copy, modify, and/or distribute this software for any
 # purpose with or without fee is hereby granted, provided that the above
@@ -344,3 +344,18 @@ indicate an error in the source code, please submit a bug report.
 % DHCPSRV_UNKNOWN_DB unknown database type: %1
 The database access string specified a database type (given in the
 message) that is unknown to the software.  This is a configuration error.
+
+% DHCPSRV_DHCP_DDNS_HANDLER_NULL error handler for DHCP_DDNS IO is not set.
+This is an error message that occurs when an attempt to send a request to
+b10-dhcp-ddns fails and there is no registered error handler.  This is a
+programmatic error which should never occur and should be reported.
+
+% DHCPSRV_DHCP_DDNS_ERROR_EXCEPTION error handler for DHCP_DDNS IO generated an expected exception: %1
+This is an error message that occurs when an attempt to send a request to
+b10-dhcp-ddns fails there registered error handler threw an uncaught exception.
+This is a programmatic error which should not occur. By convention, the error
+handler should not propagate exceptions. Please report this error.
+
+% DHCPSRV_DHCP_DDNS_NCR_SENT NameChangeRequest sent to b10-dhcp-ddns: %1
+A debug message issued when a NameChangeRequest has been successfully sent to
+b10-dhcp-ddns.

src/lib/dhcpsrv/tests/Makefile.am

@@ -54,6 +54,7 @@ libdhcpsrv_unittests_SOURCES += alloc_engine_unittest.cc
 libdhcpsrv_unittests_SOURCES += callout_handle_store_unittest.cc
 libdhcpsrv_unittests_SOURCES += cfgmgr_unittest.cc
 libdhcpsrv_unittests_SOURCES += d2_client_unittest.cc
+libdhcpsrv_unittests_SOURCES += d2_udp_unittest.cc
 libdhcpsrv_unittests_SOURCES += dbaccess_parser_unittest.cc
 libdhcpsrv_unittests_SOURCES += lease_unittest.cc
 libdhcpsrv_unittests_SOURCES += lease_mgr_factory_unittest.cc

src/lib/dhcpsrv/tests/d2_udp_unittest.cc

@@ -0,0 +1,379 @@
+// Copyright (C) 2014 Internet Systems Consortium, Inc. ("ISC")
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
+// OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+/// @file d2_upd_unittest.cc Unit tests for D2ClientMgr UDP communications.
+/// Note these tests are not intended to verify the actual send and receive
+/// across UDP sockets.  This level of testing is done in libdhcp-ddns.
+
+#include <asio.hpp>
+#include <asiolink/io_service.h>
+#include <config.h>
+#include <dhcpsrv/d2_client.h>
+#include <exceptions/exceptions.h>
+
+#include <boost/function.hpp>
+#include <boost/bind.hpp>
+
+#include <gtest/gtest.h>
+
+#include <sys/select.h>
+
+using namespace std;
+using namespace isc::dhcp;
+using namespace isc;
+
+namespace {
+
+/// @brief Test fixture for excerising D2ClientMgr send management
+/// services.  It inherents from D2ClientMgr to allow overriding various
+/// methods and accessing otherwise restricted member.  In particular it
+/// overrides the NameChangeSender completion completion callback, allowing
+/// the injection of send errors.
+class D2ClientMgrTest : public D2ClientMgr, public ::testing::Test {
+public:
+    /// @brief If true simulates a send which completed with a failed status.
+    bool simulate_send_failure_;
+    /// @brief If true causes an exception throw in the client error handler.
+    bool error_handler_throw_;
+    /// @brief Tracks the number times the completion handler is called.
+    int callback_count_;
+    /// @brief Tracks the number of times the client error handler was called.
+    int error_handler_count_;
+
+    /// @brief Constructor
+    D2ClientMgrTest() : simulate_send_failure_(false),
+                       error_handler_throw_(false),
+                       callback_count_(0), error_handler_count_(0) {
+    }
+
+    /// @brief virtual Destructor
+    virtual ~D2ClientMgrTest(){
+    }
+
+    /// @brief Updates the D2ClientMgr's configuration to DDNS disabled.
+    void disableDdns() {
+        D2ClientConfigPtr new_cfg;
+        ASSERT_NO_THROW(new_cfg.reset(new D2ClientConfig()));
+        ASSERT_NO_THROW(setD2ClientConfig(new_cfg));
+        ASSERT_FALSE(ddnsEnabled());
+    }
+
+    /// @brief Updates the D2ClientMgr's configuration to DDNS enabled.
+    ///
+    /// @param server_address IP address of b10-dhcp-ddns.
+    /// @param server_port IP port number of b10-dhcp-ddns.
+    /// @param protocol NCR protocol to use. (Currently only UDP is
+    /// supported).
+    void enableDdns(const std::string& server_address,
+                    const size_t server_port,
+                    const dhcp_ddns::NameChangeProtocol protocol) {
+        // Update the configuration with one that is enabled.
+        D2ClientConfigPtr new_cfg;
+        ASSERT_NO_THROW(new_cfg.reset(new D2ClientConfig(true,
+                                  isc::asiolink::IOAddress(server_address),
+                                  server_port,
+                                  protocol, dhcp_ddns::FMT_JSON,
+                                  true, true, true, true,
+                                  "myhost", ".example.com.")));
+        ASSERT_NO_THROW(setD2ClientConfig(new_cfg));
+        ASSERT_TRUE(ddnsEnabled());
+    }
+
+    /// @brief Checks sender's select-fd against an expected state of readiness.
+    ///
+    /// Uses select() to determine if the sender's select_fd is marked as
+    /// ready to read, and compares this against the expected state.  The
+    /// select function is called with a timeout of 0.0 (non blocking).
+    ///
+    /// @param expect_ready Expected state of readiness (True if expecting
+    /// a ready to ready result,  false if expecting otherwise).
+    void selectCheck(bool expect_ready) {
+        fd_set read_fds;
+        int maxfd = 0;
+
+        FD_ZERO(&read_fds);
+
+        int select_fd = -1;
+        ASSERT_NO_THROW(select_fd = getSelectFd());
+
+        FD_SET(select_fd,  &read_fds);
+        maxfd = select_fd;
+
+        struct timeval select_timeout;
+        select_timeout.tv_sec = 0;
+        select_timeout.tv_usec = 0;
+
+        int result = (select(maxfd + 1, &read_fds, NULL, NULL,
+                      &select_timeout));
+
+        if (result < 0) {
+            const char *errstr = strerror(errno);
+            FAIL() << "select failed :" << errstr;
+        }
+
+        if (expect_ready) {
+            ASSERT_TRUE(result > 0);
+        } else {
+            ASSERT_TRUE(result == 0);
+        }
+    }
+
+    /// @brief Overrides base class completion callback.
+    ///
+    /// This method will be invoked each time a send completes. It allows
+    /// intervention prior to calling the production implemenation in the
+    /// base.  If simulate_send_failure_ is true, the base call impl will
+    /// be called with an error status, otherwise it will be called with
+    /// the result paramater given.
+    ///
+    /// @param result Result code of the send operation.
+    /// @param ncr NameChangeRequest which failed to send.
+    virtual void operator()(const dhcp_ddns::NameChangeSender::Result result,
+                            dhcp_ddns::NameChangeRequestPtr& ncr) {
+        ++callback_count_;
+        if (simulate_send_failure_) {
+            simulate_send_failure_ = false;
+            D2ClientMgr::operator()(dhcp_ddns::NameChangeSender::ERROR, ncr);
+        } else {
+            D2ClientMgr::operator()(result, ncr);
+        }
+    }
+
+    /// @brief Serves as the "application level" client error handler.
+    ///
+    /// This method is passed into calls to startSender as the client error
+    /// handler.  It should be invoked whenever the completion callback is
+    /// passed a result other than SUCCESS.  If error_handler_throw_
+    /// is true it will throw an exception.
+    ///
+    /// @param result unused - Result code of the send operation.
+    /// @param ncr unused -NameChangeRequest which failed to send.
+    void error_handler(const dhcp_ddns::NameChangeSender::Result /*result*/,
+                       dhcp_ddns::NameChangeRequestPtr& /*ncr*/) {
+        if (error_handler_throw_) {
+            error_handler_throw_ = false;
+            isc_throw(isc::InvalidOperation, "Simulated client handler throw");
+        }
+
+        ++error_handler_count_;
+    }
+
+    /// @brief Returns D2ClientErroHandler bound to this::error_handler_.
+    D2ClientErrorHandler getErrorHandler() {
+        return (boost::bind(&D2ClientMgrTest::error_handler, this, _1, _2));
+    }
+
+    /// @brief Contructs a NameChangeRequest message from a fixed JSON string.
+    dhcp_ddns::NameChangeRequestPtr buildTestNcr() {
+        // Build an NCR from json string.
+        const char* ncr_str =
+            "{"
+            " \"change_type\" : 0 , "
+            " \"forward_change\" : true , "
+            " \"reverse_change\" : false , "
+            " \"fqdn\" : \"myhost.example.com.\" , "
+            " \"ip_address\" : \"192.168.2.1\" , "
+            " \"dhcid\" : \"010203040A7F8E3D\" , "
+            " \"lease_expires_on\" : \"20140121132405\" , "
+            " \"lease_length\" : 1300 "
+            "}";
+
+        return (dhcp_ddns::NameChangeRequest::fromJSON(ncr_str));
+    }
+
+    /// Expose restricted members.
+    using D2ClientMgr::getSelectFd;
+};
+
+
+/// @brief Checks that D2ClientMgr disable and enable a UDP sender.
+TEST_F(D2ClientMgrTest, udpSenderEnableDisable) {
+    // Verify DDNS is disabled by default.
+    ASSERT_FALSE(ddnsEnabled());
+
+    // Verify we are not in send mode.
+    ASSERT_FALSE(amSending());
+
+    // Enable DDNS with server at 127.0.0.1/prot 53001 via UDP.
+    enableDdns("127.0.0.1", 530001, dhcp_ddns::NCR_UDP);
+    ASSERT_FALSE(amSending());
+
+    ASSERT_NO_THROW(startSender(getErrorHandler()));
+    ASSERT_TRUE(amSending());
+
+    // Verify that we take sender out of send mode.
+    ASSERT_NO_THROW(stopSender());
+    ASSERT_FALSE(amSending());
+}
+
+/// @brief Checks D2ClientMgr queuing methods with a UDP sender.
+TEST_F(D2ClientMgrTest, udpSenderQueing) {
+    // Enable DDNS with server at 127.0.0.1/prot 53001 via UDP.
+    enableDdns("127.0.0.1", 530001, dhcp_ddns::NCR_UDP);
+    ASSERT_FALSE(amSending());
+
+    // Queue should be empty.
+    EXPECT_EQ(0, getQueueSize());
+
+    // Trying to peek past the end of the queue should throw.
+    EXPECT_THROW(peekAt(1), dhcp_ddns::NcrSenderError);
+
+    // Trying to send a NCR when not in send mode should fail.
+    dhcp_ddns::NameChangeRequestPtr ncr = buildTestNcr();
+    EXPECT_THROW(sendRequest(ncr), dhcp_ddns::NcrSenderError);
+
+    // Place sender in send mode.
+    ASSERT_NO_THROW(startSender(getErrorHandler()));
+    ASSERT_TRUE(amSending());
+
+    // Send should succeed now.
+    ASSERT_NO_THROW(sendRequest(ncr));
+
+    // Queue should have 1 entry.
+    EXPECT_EQ(1, getQueueSize());
+
+    // Attempt to fetch the entry we just queued.
+    dhcp_ddns::NameChangeRequestPtr ncr2;
+    ASSERT_NO_THROW(ncr2 = peekAt(0));
+
+    // Verify what we queued matches what we fetched.
+    EXPECT_TRUE(*ncr == *ncr2);
+
+    // Clearing the queue while in send mode should fail.
+    ASSERT_THROW(clearQueue(), dhcp_ddns::NcrSenderError);
+
+    // We should still have 1 in the queue.
+    EXPECT_EQ(1, getQueueSize());
+
+    // Get out of send mode.
+    ASSERT_NO_THROW(stopSender());
+    ASSERT_FALSE(amSending());
+
+    // Clear queue should succeed now.
+    ASSERT_NO_THROW(clearQueue());
+    EXPECT_EQ(0, getQueueSize());
+}
+
+/// @brief Checks that D2ClientMgr can send with a UDP sender and
+/// a private IOService.
+TEST_F(D2ClientMgrTest, udpSend) {
+    // Enable DDNS with server at 127.0.0.1/prot 53001 via UDP.
+    enableDdns("127.0.0.1", 530001, dhcp_ddns::NCR_UDP);
+
+    // Trying to fetch the select-fd when not sending should fail.
+    ASSERT_THROW(getSelectFd(), D2ClientError);
+
+    // Place sender in send mode.
+    ASSERT_NO_THROW(startSender(getErrorHandler()));
+
+    // select_fd should evaluate to NOT ready to read.
+    selectCheck(false);
+
+    // Build a test request and send it.
+    dhcp_ddns::NameChangeRequestPtr ncr = buildTestNcr();
+    ASSERT_NO_THROW(sendRequest(ncr));
+
+    // select_fd should evaluate to ready to read.
+    selectCheck(true);
+
+    // Call service handler.
+    runReadyIO();
+
+    // select_fd should evaluate to not ready to read.
+    selectCheck(false);
+}
+
+/// @brief Checks that D2ClientMgr can send with a UDP sender and
+/// an external IOService.
+TEST_F(D2ClientMgrTest, udpSendExternalIOService) {
+    // Enable DDNS with server at 127.0.0.1/prot 53001 via UDP.
+    enableDdns("127.0.0.1", 530001, dhcp_ddns::NCR_UDP);
+
+    // Place sender in send mode using an external IO service.
+    asiolink::IOService io_service;
+    ASSERT_NO_THROW(startSender(getErrorHandler(), io_service));
+
+    // select_fd should evaluate to NOT ready to read.
+    selectCheck(false);
+
+    // Build a test request and send it.
+    dhcp_ddns::NameChangeRequestPtr ncr = buildTestNcr();
+    ASSERT_NO_THROW(sendRequest(ncr));
+
+    // select_fd should evaluate to ready to read.
+    selectCheck(true);
+
+    // Call service handler.
+    runReadyIO();
+
+    // select_fd should evaluate to not ready to read.
+    selectCheck(false);
+}
+
+/// @brief Checks that D2ClientMgr invokes the client error handler
+/// when send errors occur.
+TEST_F(D2ClientMgrTest, udpSendErrorHandler) {
+    // Enable DDNS with server at 127.0.0.1/prot 53001 via UDP.
+    enableDdns("127.0.0.1", 530001, dhcp_ddns::NCR_UDP);
+
+    // Trying to fetch the select-fd when not sending should fail.
+    ASSERT_THROW(getSelectFd(), D2ClientError);
+
+    // Place sender in send mode.
+    ASSERT_NO_THROW(startSender(getErrorHandler()));
+
+    // select_fd should evaluate to NOT ready to read.
+    selectCheck(false);
+
+    // Simulate a failed response in the send call back. This should
+    // cause the error handler to get invoked.
+    simulate_send_failure_ = true;
+
+    ASSERT_EQ(0, error_handler_count_);
+
+    // Send a test request.
+    dhcp_ddns::NameChangeRequestPtr ncr = buildTestNcr();
+    ASSERT_NO_THROW(sendRequest(ncr));
+
+    // select_fd should evaluate to ready to read.
+    selectCheck(true);
+
+    // Call service handler.
+    runReadyIO();
+
+    // select_fd should evaluate to not ready to read.
+    selectCheck(false);
+
+    ASSERT_EQ(1, error_handler_count_);
+
+    // Simulate a failed response in the send call back. This should
+    // cause the error handler to get invoked.
+    simulate_send_failure_ = true;
+    error_handler_throw_ = true;
+
+    // Send a test request.
+    ncr = buildTestNcr();
+    ASSERT_NO_THROW(sendRequest(ncr));
+
+    // Call the io service handler.
+    runReadyIO();
+
+    // Simulation flag should be false.
+    ASSERT_FALSE(error_handler_throw_);
+
+    // Count should still be 1.
+    ASSERT_EQ(1, error_handler_count_);
+}
+
+} // end of anonymous namespace