merged branches/trac221b for trac #221.

git-svn-id: svn://bind10.isc.org/svn/bind10/trunk@2565 e5f2f494-b856-4b98-b285-d166d9295462

doc/Doxyfile

@@ -568,7 +568,7 @@ WARN_LOGFILE           =
 # directories like "/usr/src/myproject". Separate the files or directories
 # with spaces.
 
-INPUT                  = ../src/lib/cc ../src/lib/config ../src/lib/dns ../src/lib/exceptions ../src/lib/datasrc
+INPUT                  = ../src/lib/cc ../src/lib/config ../src/lib/dns ../src/lib/exceptions ../src/lib/datasrc ../src/bin/auth
 
 # This tag can be used to specify the character encoding of the source files
 # that doxygen parses. Internally doxygen uses the UTF-8 encoding, which is

src/bin/auth/Makefile.am

@@ -1,6 +1,7 @@
 SUBDIRS = . tests
 
 AM_CPPFLAGS = -I$(top_srcdir)/src/lib -I$(top_builddir)/src/lib
+AM_CPPFLAGS += -I$(top_srcdir)/src/bin -I$(top_builddir)/src/bin
 AM_CPPFLAGS += -I$(top_srcdir)/src/lib/dns -I$(top_builddir)/src/lib/dns
 AM_CPPFLAGS += -I$(top_builddir)/src/lib/cc
 AM_CPPFLAGS += -I$(top_srcdir)/src/bin -I$(top_builddir)/src/bin
@@ -54,8 +55,8 @@ b10_auth_LDADD += $(top_builddir)/src/lib/config/.libs/libcfgclient.a
 b10_auth_LDADD += $(top_builddir)/src/lib/cc/.libs/libcc.a
 b10_auth_LDADD += $(top_builddir)/src/lib/exceptions/.libs/libexceptions.a
 b10_auth_LDADD += $(top_builddir)/src/bin/auth/libasio_link.a
-b10_auth_LDADD += $(SQLITE_LIBS)
 b10_auth_LDADD += $(top_builddir)/src/lib/xfr/.libs/libxfr.a
+b10_auth_LDADD += $(SQLITE_LIBS)
 
 # TODO: config.h.in is wrong because doesn't honor pkgdatadir
 # and can't use @datadir@ because doesn't expand default ${prefix}

src/bin/auth/asio_link.cc

@@ -17,83 +17,167 @@
 #include <config.h>
 
 #include <unistd.h>             // for some IPC/network system calls
+#include <sys/socket.h>
+#include <netinet/in.h>
+
 #include <asio.hpp>
 #include <boost/lexical_cast.hpp>
 #include <boost/bind.hpp>
 
+#include <boost/shared_ptr.hpp>
+
 #include <dns/buffer.h>
 #include <dns/message.h>
 #include <dns/messagerenderer.h>
 
-#include <xfr/xfrout_client.h>
-
 #include <asio_link.h>
 
-#include <auth/spec_config.h>   // for XFROUT.  should not be here.
 #include <auth/auth_srv.h>
 #include <auth/common.h>
 
 using namespace asio;
-using ip::udp;
-using ip::tcp;
+using asio::ip::udp;
+using asio::ip::tcp;
 
 using namespace std;
 using namespace isc::dns;
-using namespace isc::xfr;
-
-namespace {
-// As a short term workaround, we have XFROUT specific code.  We should soon
-// refactor the code with some abstraction so that we can separate this level
-// details from the (AS)IO module.
-
-// This was contained in an ifdef USE_XFROUT, but we should really check
-// live if we do xfrout
-//TODO. The sample way for checking axfr query, the code should be merged to auth server class
-bool
-check_axfr_query(char* const msg_data, const uint16_t msg_len) {
-    if (msg_len < 15) {
-        return false;
-    }
 
-    const uint16_t query_type = *(uint16_t *)(msg_data + (msg_len - 4));
-    if ( query_type == 0xFC00) {
-        return true;
+namespace asio_link {
+IOAddress::IOAddress(const string& address_str)
+    // XXX: we cannot simply construct the address in the initialization list
+    // because we'd like to throw our own exception on failure.
+{
+    error_code err;
+    asio_address_ = ip::address::from_string(address_str, err);
+    if (err) {
+        isc_throw(IOError, "Failed to convert string to address '"
+                  << address_str << "': " << err.message());
     }
-    
-    return false;
 }
 
-//TODO. Send the xfr query to xfrout module, the code should be merged to auth server class
-//BIGGERTODO: stop using hardcoded install-path locations! 
-void
-dispatch_axfr_query(const int tcp_sock, char const axfr_query[],
-                    const uint16_t query_len)
-{
-    string path;
-    if (getenv("B10_FROM_BUILD")) {
-        path = string(getenv("B10_FROM_BUILD")) + "/auth_xfrout_conn";
-    } else {
-        path = UNIX_SOCKET_FILE;
-    }
-    
-    if (getenv("B10_FROM_BUILD")) {
-        path = string(getenv("B10_FROM_BUILD")) + "/auth_xfrout_conn";
+IOAddress::IOAddress(const ip::address& asio_address) :
+    asio_address_(asio_address)
+{}
+
+string
+IOAddress::toText() const {
+    return (asio_address_.to_string());
+}
+
+// Note: this implementation is optimized for the case where this object
+// is created from an ASIO endpoint object in a receiving code path
+// by avoiding to make a copy of the base endpoint.  For TCP it may not be
+// a bug deal, but when we receive UDP packets at a high rate, the copy
+// overhead might be significant.
+class TCPEndpoint : public IOEndpoint {
+public:
+    TCPEndpoint(const IOAddress& address, const unsigned short port) :
+        asio_endpoint_placeholder_(
+            new tcp::endpoint(ip::address::from_string(address.toText()),
+                              port)),
+        asio_endpoint_(*asio_endpoint_placeholder_)
+    {}
+    TCPEndpoint(const tcp::endpoint& asio_endpoint) :
+        asio_endpoint_placeholder_(NULL), asio_endpoint_(asio_endpoint)
+    {}
+        
+    ~TCPEndpoint() { delete asio_endpoint_placeholder_; }
+    virtual IOAddress getAddress() const {
+        return (asio_endpoint_.address());
     }
-    XfroutClient xfr_client(path);
-    try {
-        xfr_client.connect();
-        xfr_client.sendXfroutRequestInfo(tcp_sock, (uint8_t *)axfr_query,
-                                         query_len);
-        xfr_client.disconnect();
+private:
+    const tcp::endpoint* asio_endpoint_placeholder_;
+    const tcp::endpoint& asio_endpoint_;
+};
+
+class UDPEndpoint : public IOEndpoint {
+public:
+    UDPEndpoint(const IOAddress& address, const unsigned short port) :
+        asio_endpoint_placeholder_(
+            new udp::endpoint(ip::address::from_string(address.toText()),
+                              port)),
+        asio_endpoint_(*asio_endpoint_placeholder_)
+    {}
+    UDPEndpoint(const udp::endpoint& asio_endpoint) :
+        asio_endpoint_placeholder_(NULL), asio_endpoint_(asio_endpoint)
+    {}
+    ~UDPEndpoint() { delete asio_endpoint_placeholder_; }
+    virtual IOAddress getAddress() const {
+        return (asio_endpoint_.address());
     }
-    catch (const exception & err) {
-        //if (verbose_mode)
-        cerr << "error handle xfr query " << UNIX_SOCKET_FILE << ":" << err.what() << endl;
+private:
+    const udp::endpoint* asio_endpoint_placeholder_;
+    const udp::endpoint& asio_endpoint_;
+};
+
+const IOEndpoint*
+IOEndpoint::create(const int protocol, const IOAddress& address,
+                   const unsigned short port)
+{
+    if (protocol == IPPROTO_UDP) {
+        return (new UDPEndpoint(address, port));
+    } else if (protocol == IPPROTO_TCP) {
+        return (new TCPEndpoint(address, port));
     }
+    isc_throw(IOError,
+              "IOEndpoint creation attempt for unsupported protocol: " <<
+              protocol);
 }
+
+class TCPSocket : public IOSocket {
+private:
+    TCPSocket(const TCPSocket& source);
+    TCPSocket& operator=(const TCPSocket& source);
+public:
+    TCPSocket(tcp::socket& socket) : socket_(socket) {}
+    virtual int getNative() const { return (socket_.native()); }
+    virtual int getProtocol() const { return (IPPROTO_TCP); }
+private:
+    tcp::socket& socket_;
+};
+
+class UDPSocket : public IOSocket {
+private:
+    UDPSocket(const UDPSocket& source);
+    UDPSocket& operator=(const UDPSocket& source);
+public:
+    UDPSocket(udp::socket& socket) : socket_(socket) {}
+    virtual int getNative() const { return (socket_.native()); }
+    virtual int getProtocol() const { return (IPPROTO_UDP); }
+private:
+    udp::socket& socket_;
+};
+
+class DummySocket : public IOSocket {
+private:
+    DummySocket(const DummySocket& source);
+    DummySocket& operator=(const DummySocket& source);
+public:
+    DummySocket(const int protocol) : protocol_(protocol) {}
+    virtual int getNative() const { return (-1); }
+    virtual int getProtocol() const { return (protocol_); }
+private:
+    const int protocol_;
+};
+
+IOSocket&
+IOSocket::getDummyUDPSocket() {
+    static DummySocket socket(IPPROTO_UDP);
+    return (socket);
 }
 
-namespace asio_link {
+IOSocket&
+IOSocket::getDummyTCPSocket() {
+    static DummySocket socket(IPPROTO_TCP);
+    return (socket);
+}
+
+IOMessage::IOMessage(const void* data, const size_t data_size,
+                     IOSocket& io_socket, const IOEndpoint& remote_endpoint) :
+    data_(data), data_size_(data_size), io_socket_(io_socket),
+    remote_endpoint_(remote_endpoint)
+{}
+
 //
 // Helper classes for asynchronous I/O using asio
 //
@@ -102,15 +186,18 @@ public:
     TCPClient(AuthSrv* auth_server, io_service& io_service) :
         auth_server_(auth_server),
         socket_(io_service),
+        io_socket_(socket_),
         response_buffer_(0),
         responselen_buffer_(TCP_MESSAGE_LENGTHSIZE),
         response_renderer_(response_buffer_),
-        dns_message_(Message::PARSE)
+        dns_message_(Message::PARSE),
+        custom_callback_(NULL)
     {}
 
     void start() {
         // Check for queued configuration commands
-        if (auth_server_->configSession()->hasQueuedMsgs()) {
+        if (auth_server_ != NULL &&
+            auth_server_->configSession()->hasQueuedMsgs()) {
             auth_server_->configSession()->checkCommand();
         }
         async_read(socket_, asio::buffer(data_, TCP_MESSAGE_LENGTHSIZE),
@@ -129,7 +216,6 @@ public:
 
             uint16_t msglen = dnsbuffer.readUint16();
             async_read(socket_, asio::buffer(data_, msglen),
-
                        boost::bind(&TCPClient::requestRead, this,
                                    placeholders::error,
                                    placeholders::bytes_transferred));
@@ -142,25 +228,28 @@ public:
                      size_t bytes_transferred)
     {
         if (!error) {
-            InputBuffer dnsbuffer(data_, bytes_transferred);
-            if (check_axfr_query(data_, bytes_transferred)) {
-                dispatch_axfr_query(socket_.native(), data_, bytes_transferred); 
-                // start to get new query ?
+            const TCPEndpoint remote_endpoint(socket_.remote_endpoint());
+            const IOMessage io_message(data_, bytes_transferred, io_socket_,
+                                       remote_endpoint);
+            // currently, for testing purpose only
+            if (custom_callback_ != NULL) {
+                (*custom_callback_)(io_message);
                 start();
+                return;
+            }
+
+            if (auth_server_->processMessage(io_message, dns_message_,
+                                             response_renderer_)) {
+                responselen_buffer_.writeUint16(
+                    response_buffer_.getLength());
+                async_write(socket_,
+                            asio::buffer(
+                                responselen_buffer_.getData(),
+                                responselen_buffer_.getLength()),
+                            boost::bind(&TCPClient::responseWrite, this,
+                                        placeholders::error));
             } else {
-                if (auth_server_->processMessage(dnsbuffer, dns_message_,
-                                                response_renderer_, false)) {
-                    responselen_buffer_.writeUint16(
-                        response_buffer_.getLength());
-                    async_write(socket_,
-                                asio::buffer(
-                                    responselen_buffer_.getData(),
-                                    responselen_buffer_.getLength()),
-                                boost::bind(&TCPClient::responseWrite, this,
-                                            placeholders::error));
-                } else {
-                    delete this;
-                }
+                delete this;
             }
         } else {
             delete this;
@@ -171,9 +260,9 @@ public:
         if (!error) {
                 async_write(socket_,
                             asio::buffer(response_buffer_.getData(),
-                                                response_buffer_.getLength()),
-                        boost::bind(&TCPClient::handleWrite, this,
-                                    placeholders::error));
+                                         response_buffer_.getLength()),
+                            boost::bind(&TCPClient::handleWrite, this,
+                                        placeholders::error));
         } else {
             delete this;
         }
@@ -187,9 +276,15 @@ public:
       }
     }
 
+    // Currently this is for tests only
+    void setCallBack(const IOService::IOCallBack* callback) {
+        custom_callback_ = callback;
+    }
+
 private:
     AuthSrv* auth_server_;
     tcp::socket socket_;
+    TCPSocket io_socket_;
     OutputBuffer response_buffer_;
     OutputBuffer responselen_buffer_;
     MessageRenderer response_renderer_;
@@ -197,21 +292,25 @@ private:
     enum { MAX_LENGTH = 65535 };
     static const size_t TCP_MESSAGE_LENGTHSIZE = 2;
     char data_[MAX_LENGTH];
+
+    // currently, for testing purpose only.
+    const IOService::IOCallBack* custom_callback_;
 };
 
 class TCPServer {
 public:
     TCPServer(AuthSrv* auth_server, io_service& io_service,
-              int af, uint16_t port) :
+              const ip::address& addr, const uint16_t port) :
         auth_server_(auth_server), io_service_(io_service),
         acceptor_(io_service_), listening_(new TCPClient(auth_server_,
-                                                         io_service_))
+                                                         io_service_)),
+        custom_callback_(NULL)
     {
-        tcp::endpoint endpoint(af == AF_INET6 ? tcp::v6() : tcp::v4(), port);
+        tcp::endpoint endpoint(addr, port);
         acceptor_.open(endpoint.protocol());
         // Set v6-only (we use a different instantiation for v4,
         // otherwise asio will bind to both v4 and v6
-        if (af == AF_INET6) {
+        if (addr.is_v6()) {
             acceptor_.set_option(ip::v6_only(true));
         }
         acceptor_.set_option(tcp::acceptor::reuse_address(true));
@@ -222,23 +321,6 @@ public:
                                            listening_, placeholders::error));
     }
 
-    TCPServer(AuthSrv* auth_server, io_service& io_service,
-              asio::ip::address addr, uint16_t port) :
-        auth_server_(auth_server),
-        io_service_(io_service), acceptor_(io_service_),
-        listening_(new TCPClient(auth_server, io_service_))
-    {
-        tcp::endpoint endpoint(addr, port);
-        acceptor_.open(endpoint.protocol());
-
-        acceptor_.set_option(tcp::acceptor::reuse_address(true));
-        acceptor_.bind(endpoint);
-        acceptor_.listen();
-        acceptor_.async_accept(listening_->getSocket(),
-                               boost::bind(&TCPServer::handleAccept, this,
-                                           listening_, placeholders::error));
-    }
-
     ~TCPServer() { delete listening_; }
 
     void handleAccept(TCPClient* new_client,
@@ -246,6 +328,7 @@ public:
     {
         if (!error) {
             assert(new_client == listening_);
+            new_client->setCallBack(custom_callback_);
             new_client->start();
             listening_ = new TCPClient(auth_server_, io_service_);
             acceptor_.async_accept(listening_->getSocket(),
@@ -257,61 +340,68 @@ public:
         }
     }
 
+    // Currently this is for tests only
+    void setCallBack(const IOService::IOCallBack* callback) {
+        custom_callback_ = callback;
+    }
+
 private:
     AuthSrv* auth_server_;
     io_service& io_service_;
     tcp::acceptor acceptor_;
     TCPClient* listening_;
+
+    // currently, for testing purpose only.
+    const IOService::IOCallBack* custom_callback_;
 };
 
 class UDPServer {
 public:
     UDPServer(AuthSrv* auth_server, io_service& io_service,
-              int af, uint16_t port) :
+              const ip::address& addr, const uint16_t port) :
         auth_server_(auth_server),
         io_service_(io_service),
-        socket_(io_service, af == AF_INET6 ? udp::v6() : udp::v4()),
+        socket_(io_service, addr.is_v6() ? udp::v6() : udp::v4()),
+        io_socket_(socket_),
         response_buffer_(0),
         response_renderer_(response_buffer_),
-        dns_message_(Message::PARSE)
+        dns_message_(Message::PARSE),
+        custom_callback_(NULL)
     {
         // Set v6-only (we use a different instantiation for v4,
         // otherwise asio will bind to both v4 and v6
-        if (af == AF_INET6) {
+        if (addr.is_v6()) {
             socket_.set_option(asio::ip::v6_only(true));
-            socket_.bind(udp::endpoint(udp::v6(), port));
+            socket_.bind(udp::endpoint(addr, port));
         } else {
-            socket_.bind(udp::endpoint(udp::v4(), port));
+            socket_.bind(udp::endpoint(addr, port));
         }
         startReceive();
     }
 
-    UDPServer(AuthSrv* auth_server, io_service& io_service,
-              asio::ip::address addr, uint16_t port) :
-        auth_server_(auth_server), io_service_(io_service),
-        socket_(io_service, addr.is_v6() ? udp::v6() : udp::v4()),
-        response_buffer_(0),
-        response_renderer_(response_buffer_),
-        dns_message_(Message::PARSE)
-    {
-        socket_.bind(udp::endpoint(addr, port));
-        startReceive();
-    }
-
     void handleRequest(const asio::error_code& error,
                        size_t bytes_recvd)
     {
         // Check for queued configuration commands
-        if (auth_server_->configSession()->hasQueuedMsgs()) {
+        if (auth_server_ != NULL &&
+            auth_server_->configSession()->hasQueuedMsgs()) {
             auth_server_->configSession()->checkCommand();
         }
         if (!error && bytes_recvd > 0) {
-            InputBuffer request_buffer(data_, bytes_recvd);
+            const UDPEndpoint remote_endpoint(sender_endpoint_);
+            const IOMessage io_message(data_, bytes_recvd, io_socket_,
+                                       remote_endpoint);
+            // currently, for testing purpose only
+            if (custom_callback_ != NULL) {
+                (*custom_callback_)(io_message);
+                startReceive();
+                return;
+            }
 
             dns_message_.clear(Message::PARSE);
             response_renderer_.clear();
-            if (auth_server_->processMessage(request_buffer, dns_message_,
-                                            response_renderer_, true)) {
+            if (auth_server_->processMessage(io_message, dns_message_,
+                                             response_renderer_)) {
                 socket_.async_send_to(
                     asio::buffer(response_buffer_.getData(),
                                         response_buffer_.getLength()),
@@ -335,6 +425,11 @@ public:
         // the next request.
         startReceive();
     }
+
+    // Currently this is for tests only
+    void setCallBack(const IOService::IOCallBack* callback) {
+        custom_callback_ = callback;
+    }
 private:
     void startReceive() {
         socket_.async_receive_from(
@@ -348,121 +443,99 @@ private:
     AuthSrv* auth_server_;
     io_service& io_service_;
     udp::socket socket_;
+    UDPSocket io_socket_;
     OutputBuffer response_buffer_;
     MessageRenderer response_renderer_;
     Message dns_message_;
     udp::endpoint sender_endpoint_;
     enum { MAX_LENGTH = 4096 };
     char data_[MAX_LENGTH];
-};
 
-// This is a helper structure just to make the construction of IOServiceImpl
-// exception safe.  If the constructor of {UDP/TCP}Server throws an exception,
-// the destructor of this class will automatically perform the necessary
-// cleanup.
-struct ServerSet {
-    ServerSet() : udp4_server(NULL), udp6_server(NULL),
-                  tcp4_server(NULL), tcp6_server(NULL)
-    {}
-    ~ServerSet() {
-        delete udp4_server;
-        delete udp6_server;
-        delete tcp4_server;
-        delete tcp6_server;
-    }
-    UDPServer* udp4_server;
-    UDPServer* udp6_server;
-    TCPServer* tcp4_server;
-    TCPServer* tcp6_server;
+    // currently, for testing purpose only.
+    const IOService::IOCallBack* custom_callback_;
 };
 
 class IOServiceImpl {
 public:
-    IOServiceImpl(AuthSrv* auth_server, const char* address, const char* port,
-                  const bool use_ipv4, const bool use_ipv6);
-    ~IOServiceImpl();
+    IOServiceImpl(AuthSrv* auth_server, const char& port,
+                  const ip::address* v4addr, const ip::address* v6addr);
     asio::io_service io_service_;
     AuthSrv* auth_server_;
-    UDPServer* udp4_server_;
-    UDPServer* udp6_server_;
-    TCPServer* tcp4_server_;
-    TCPServer* tcp6_server_;
+
+    typedef boost::shared_ptr<UDPServer> UDPServerPtr;
+    typedef boost::shared_ptr<TCPServer> TCPServerPtr;
+    UDPServerPtr udp4_server_;
+    UDPServerPtr udp6_server_;
+    TCPServerPtr tcp4_server_;
+    TCPServerPtr tcp6_server_;
+
+    // This member is used only for testing at the moment.
+    IOService::IOCallBack callback_;
 };
 
-IOServiceImpl::IOServiceImpl(AuthSrv* auth_server, const char* const address,
-                             const char* const port, const bool use_ipv4,
-                             const bool use_ipv6) :
-    auth_server_(auth_server), udp4_server_(NULL), udp6_server_(NULL),
-    tcp4_server_(NULL), tcp6_server_(NULL)
+IOServiceImpl::IOServiceImpl(AuthSrv* auth_server, const char& port,
+                             const ip::address* const v4addr,
+                             const ip::address* const v6addr) :
+    auth_server_(auth_server),
+    udp4_server_(UDPServerPtr()), udp6_server_(UDPServerPtr()),
+    tcp4_server_(TCPServerPtr()), tcp6_server_(TCPServerPtr())
 {
-    ServerSet servers;
-    uint16_t portnum = atoi(port);
+    uint16_t portnum;
 
     try {
-        portnum = boost::lexical_cast<uint16_t>(port);
-    } catch (const std::exception& ex) {
-        isc_throw(FatalError, "[b10-auth] Invalid port number '"
-                              << port << "'");
+        portnum = boost::lexical_cast<uint16_t>(&port);
+    } catch (const boost::bad_lexical_cast& ex) {
+        isc_throw(IOError, "Invalid port number '" << &port << "': " <<
+                  ex.what());
     }
 
-    if (address != NULL) {
-        asio::ip::address addr = asio::ip::address::from_string(address);
-
-        if ((addr.is_v6() && !use_ipv6)) {
-            isc_throw(FatalError,
-                      "[b10-auth] Error: -4 conflicts with " << addr);
-        }
-
-        if ((addr.is_v4() && !use_ipv4)) {
-            isc_throw(FatalError,
-                      "[b10-auth] Error: -6 conflicts with " << addr);
-        }
-
-        if (addr.is_v4()) {
-            servers.udp4_server = new UDPServer(auth_server, io_service_,
-                                                addr, portnum);
-            servers.tcp4_server = new TCPServer(auth_server, io_service_,
-                                                addr, portnum);
-         } else {
-            servers.udp6_server = new UDPServer(auth_server, io_service_,
-                                                addr, portnum);
-            servers.tcp6_server = new TCPServer(auth_server, io_service_,
-                                                addr, portnum);
-        }
-    } else {
-        if (use_ipv4) {
-            servers.udp4_server = new UDPServer(auth_server, io_service_,
-                                                AF_INET, portnum);
-            servers.tcp4_server = new TCPServer(auth_server, io_service_,
-                                                AF_INET, portnum);
+    try {
+        if (v4addr != NULL) {
+            udp4_server_ = UDPServerPtr(new UDPServer(auth_server, io_service_,
+                                                      *v4addr, portnum));
+            tcp4_server_ = TCPServerPtr(new TCPServer(auth_server, io_service_,
+                                                      *v4addr, portnum));
         }
-        if (use_ipv6) {
-            servers.udp6_server = new UDPServer(auth_server, io_service_,
-                                                AF_INET6, portnum);
-            servers.tcp6_server = new TCPServer(auth_server, io_service_,
-                                                AF_INET6, portnum);
+        if (v6addr != NULL) {
+            udp6_server_ = UDPServerPtr(new UDPServer(auth_server, io_service_,
+                                                      *v6addr, portnum));
+            tcp6_server_ = TCPServerPtr(new TCPServer(auth_server, io_service_,
+                                                      *v6addr, portnum));
         }
+    } catch (const asio::system_error& err) {
+        // We need to catch and convert any ASIO level exceptions.
+        // This can happen for unavailable address, binding a privilege port
+        // without the privilege, etc.
+        isc_throw(IOError, "Failed to initialize network servers: " <<
+                  err.what());
     }
-
-    // Now we don't have to worry about exception, and need to make sure that
-    // the server objects won't be accidentally cleaned up.
-    servers.udp4_server = NULL;
-    servers.udp6_server = NULL;
-    servers.tcp4_server = NULL;
-    servers.tcp6_server = NULL;
 }
 
-IOServiceImpl::~IOServiceImpl() {
-    delete udp4_server_;
-    delete udp6_server_;
-    delete tcp4_server_;
-    delete tcp6_server_;
+IOService::IOService(AuthSrv* auth_server, const char& port,
+                     const char& address) :
+    impl_(NULL)
+{
+    error_code err;
+    const ip::address addr = ip::address::from_string(&address, err);
+    if (err) {
+        isc_throw(IOError, "Invalid IP address '" << &address << "': "
+                  << err.message());
+    }
+
+    impl_ = new IOServiceImpl(auth_server, port,
+                              addr.is_v4() ? &addr : NULL,
+                              addr.is_v6() ? &addr : NULL);
 }
 
-IOService::IOService(AuthSrv* auth_server, const char* const address,
-                     const char* const port, const bool use_ipv4,
-                     const bool use_ipv6) {
-    impl_ = new IOServiceImpl(auth_server, address, port, use_ipv4, use_ipv6);
+IOService::IOService(AuthSrv* auth_server, const char& port,
+                     const bool use_ipv4, const bool use_ipv6) :
+    impl_(NULL)
+{
+    const ip::address v4addr_any = ip::address(ip::address_v4::any());
+    const ip::address* const v4addrp = use_ipv4 ? &v4addr_any : NULL; 
+    const ip::address v6addr_any = ip::address(ip::address_v6::any());
+    const ip::address* const v6addrp = use_ipv6 ? &v6addr_any : NULL;
+    impl_ = new IOServiceImpl(auth_server, port, v4addrp, v6addrp);
 }
 
 IOService::~IOService() {
@@ -483,4 +556,21 @@ asio::io_service&
 IOService::get_io_service() {
     return impl_->io_service_;
 }
+
+void
+IOService::setCallBack(const IOCallBack callback) {
+    impl_->callback_ = callback;
+    if (impl_->udp4_server_ != NULL) {
+        impl_->udp4_server_->setCallBack(&impl_->callback_);
+    }
+    if (impl_->udp6_server_ != NULL) {
+        impl_->udp6_server_->setCallBack(&impl_->callback_);
+    }
+    if (impl_->tcp4_server_ != NULL) {
+        impl_->tcp4_server_->setCallBack(&impl_->callback_);
+    }
+    if (impl_->tcp6_server_ != NULL) {
+        impl_->tcp6_server_->setCallBack(&impl_->callback_);
+    }
+}
 }

src/bin/auth/asio_link.h

@@ -17,20 +17,429 @@
 #ifndef __ASIO_LINK_H
 #define __ASIO_LINK_H 1
 
+// IMPORTANT NOTE: only very few ASIO headers files can be included in
+// this file.  In particular, asio.hpp should never be included here.
+// See the description of the namespace below.
+#include <asio/ip/address.hpp>
+
+#include <functional>
+#include <string>
+
+#include <boost/function.hpp>
+
+#include <exceptions/exceptions.h>
+
+namespace asio {
+// forward declaration for IOService::get_io_service() below
+class io_service;
+}
+
 class AuthSrv;
 
+/// \namespace asio_link
+/// \brief A wrapper interface for the ASIO library.
+///
+/// The \c asio_link namespace is used to define a set of wrapper interfaces
+/// for the ASIO library.
+///
+/// BIND 10 uses the non-Boost version of ASIO because it's header-only,
+/// i.e., does not require a separate library object to be linked, and thus
+/// lowers the bar for introduction.
+///
+/// But the advantage comes with its own costs: since the header-only version
+/// includes more definitions in public header files, it tends to trigger
+/// more compiler warnings for our own sources, and, depending on the
+/// compiler options, may make the build fail.
+///
+/// We also found it may be tricky to use ASIO and standard C++ libraries
+/// in a single translation unit, i.e., a .cc file: depending on the order
+/// of including header files, ASIO may or may not work on some platforms.
+///
+/// This wrapper interface is intended to centralize these
+/// problematic issues in a single sub module.  Other BIND 10 modules should
+/// simply include \c asio_link.h and use the wrapper API instead of
+/// including ASIO header files and using ASIO-specific classes directly.
+///
+/// This wrapper may be used for other IO libraries if and when we want to
+/// switch, but generality for that purpose is not the primary goal of
+/// this module.  The resulting interfaces are thus straightforward mapping
+/// to the ASIO counterparts.
+///
+/// Notes to developers:
+/// Currently the wrapper interface is specific to the authoritative
+/// server implementation.  But the plan is to generalize it and have
+/// other modules use it.
+///
+/// One obvious drawback of this approach is performance overhead
+/// due to the additional layer.  We should eventually evaluate the cost
+/// of the wrapper abstraction in benchmark tests. Another drawback is
+/// that the wrapper interfaces don't provide all features of ASIO
+/// (at least for the moment).  We should also re-evaluate the
+/// maintenance overhead of providing necessary wrappers as we develop
+/// more.
+///
+/// On the other hand, we may be able to exploit the wrapper approach to
+/// simplify the interfaces (by limiting the usage) and unify performance
+/// optimization points.
+///
+/// As for optimization, we may want to provide a custom allocator for
+/// the placeholder of callback handlers:
+/// http://think-async.com/Asio/asio-1.3.1/doc/asio/reference/asio_handler_allocate.html
+
 namespace asio_link {
 struct IOServiceImpl;
 
+/// \brief An exception that is thrown if an error occurs within the IO
+/// module.  This is mainly intended to be a wrapper exception class for
+/// ASIO specific exceptions.
+class IOError : public isc::Exception {
+public:
+    IOError(const char* file, size_t line, const char* what) :
+        isc::Exception(file, line, what) {}
+};
+
+/// \brief The \c IOAddress class represents an IP addresses (version
+/// agnostic)
+///
+/// This class is a wrapper for the ASIO \c ip::address class.
+class IOAddress {
+public:
+    ///
+    /// \name Constructors and Destructor
+    ///
+    /// This class is copyable.  We use default versions of copy constructor
+    /// and the assignment operator.
+    /// We use the default destructor.
+    //@{
+    /// \brief Constructor from string.
+    ///
+    /// This constructor converts a textual representation of IPv4 and IPv6
+    /// addresses into an IOAddress object.
+    /// If \c address_str is not a valid representation of any type of
+    /// address, an exception of class \c IOError will be thrown.
+    /// This constructor allocates memory for the object, and if that fails
+    /// a corresponding standard exception will be thrown.
+    ///
+    /// \param address_str Textual representation of address.
+    IOAddress(const std::string& address_str);
+
+    /// \brief Constructor from an ASIO \c ip::address object.
+    ///
+    /// This constructor is intended to be used within the wrapper
+    /// implementation; user applications of the wrapper API won't use it.
+    ///
+    /// This constructor never throws an exception.
+    ///
+    /// \param asio_address The ASIO \c ip::address to be converted.
+    IOAddress(const asio::ip::address& asio_adress);
+    //@}
+
+    /// \brief Convert the address to a string.
+    ///
+    /// This method is basically expected to be exception free, but
+    /// generating the string will involve resource allocation,
+    /// and if it fails the corresponding standard exception will be thrown.
+    ///
+    /// \return A string representation of the address.
+    std::string toText() const;
+private:
+    asio::ip::address asio_address_;
+};
+
+/// \brief The \c IOEndpoint class is an abstract base class to represent
+/// a communication endpoint.
+///
+/// This class is a wrapper for the ASIO endpoint classes such as
+/// \c ip::tcp::endpoint and \c ip::udp::endpoint.
+///
+/// Derived class implementations are completely hidden within the
+/// implementation.  User applications only get access to concrete
+/// \c IOEndpoint objects via the abstract interfaces.
+class IOEndpoint {
+    ///
+    /// \name Constructors and Destructor
+    ///
+    /// Note: The copy constructor and the assignment operator are
+    /// intentionally defined as private, making this class non-copyable.
+    //@{
+private:
+    IOEndpoint(const IOEndpoint& source);
+    IOEndpoint& operator=(const IOEndpoint& source);
+protected:
+    /// \brief The default constructor.
+    ///
+    /// This is intentionally defined as \c protected as this base class
+    /// should never be instantiated (except as part of a derived class).
+    IOEndpoint() {}
+public:
+    /// The destructor.
+    virtual ~IOEndpoint() {}
+    //@}
+
+    /// \brief Returns the address of the endpoint.
+    ///
+    /// This method returns an IOAddress object corresponding to \c this
+    /// endpoint.
+    /// Note that the return value is a real object, not a reference or
+    /// a pointer.
+    /// This is aligned with the interface of the ASIO counterpart:
+    /// the \c address() method of \c ip::xxx::endpoint classes returns
+    /// an \c ip::address object.
+    /// This also means handling the address of an endpoint using this method
+    /// can be expensive.  If the address information is necessary in a
+    /// performance sensitive context and there's a more efficient interface
+    /// for that purpose, it's probably better to avoid using this method.
+    ///
+    /// This method never throws an exception.
+    ///
+    /// \return A copy of \c IOAddress object corresponding to the endpoint.
+    virtual IOAddress getAddress() const = 0;
+
+    /// \brief A polymorphic factory of endpoint from address and port.
+    ///
+    /// This method creates a new instance of (a derived class of)
+    /// \c IOEndpoint object that identifies the pair of given address
+    /// and port.
+    /// The appropriate derived class is chosen based on the specified
+    /// transport protocol.  If the \c protocol doesn't specify a protocol
+    /// supported in this implementation, an exception of class \c IOError
+    /// will be thrown.
+    ///
+    /// Memory for the created object will be dynamically allocated.  It's
+    /// caller's responsibility to \c delete it later.
+    /// If resource allocation for the new object fails, a corresponding
+    /// standard exception will be thrown.
+    ///
+    /// \param protocol The transport protocol used for the endpoint.
+    /// Currently, only \c IPPROTO_UDP and \c IPPROTO_TCP can be specified.
+    /// \param address The (IP) address of the endpoint.
+    /// \param port The transport port number of the endpoint
+    /// \return A pointer to a newly created \c IOEndpoint object.
+    static const IOEndpoint* create(int protocol,
+                                    const IOAddress& address,
+                                    unsigned short port);
+};
+
+/// \brief The \c IOSocket class is an abstract base class to represent
+/// various types of network sockets.
+///
+/// This class is a wrapper for the ASIO socket classes such as
+/// \c ip::tcp::socket and \c ip::udp::socket.
+///
+/// Derived class implementations are completely hidden within the
+/// implementation.  User applications only get access to concrete
+/// \c IOSocket objects via the abstract interfaces.
+/// We may revisit this decision when we generalize the wrapper and more
+/// modules use it.  Also, at that point we may define a separate (visible)
+/// derived class for testing purposes rather than providing factory methods
+/// (i.e., getDummy variants below).
+class IOSocket {
+    ///
+    /// \name Constructors and Destructor
+    ///
+    /// Note: The copy constructor and the assignment operator are
+    /// intentionally defined as private, making this class non-copyable.
+    //@{
+private:
+    IOSocket(const IOSocket& source);
+    IOSocket& operator=(const IOSocket& source);
+protected:
+    /// \brief The default constructor.
+    ///
+    /// This is intentionally defined as \c protected as this base class
+    /// should never be instantiated (except as part of a derived class).
+    IOSocket() {}
+public:
+    /// The destructor.
+    virtual ~IOSocket() {}
+    //@}
+
+    /// \brief Return the "native" representation of the socket.
+    ///
+    /// In practice, this is the file descriptor of the socket for
+    /// UNIX-like systems so the current implementation simply uses
+    /// \c int as the type of the return value.
+    /// We may have to need revisit this decision later.
+    ///
+    /// In general, the application should avoid using this method;
+    /// it essentially discloses an implementation specific "handle" that
+    /// can change the internal state of the socket (consider the
+    /// application closes it, for example).
+    /// But we sometimes need to perform very low-level operations that
+    /// requires the native representation.  Passing the file descriptor
+    /// to a different process is one example.
+    /// This method is provided as a necessary evil for such limited purposes.
+    ///
+    /// This method never throws an exception.
+    ///
+    /// \return The native representation of the socket.  This is the socket
+    /// file descriptor for UNIX-like systems.
+    virtual int getNative() const = 0;
+
+    /// \brief Return the transport protocol of the socket.
+    ///
+    /// Currently, it returns \c IPPROTO_UDP for UDP sockets, and
+    /// \c IPPROTO_TCP for TCP sockets.
+    ///
+    /// This method never throws an exception.
+    ///
+    /// \return IPPROTO_UDP for UDP sockets
+    /// \return IPPROTO_TCP for TCP sockets
+    virtual int getProtocol() const = 0;
+
+    /// \brief Return a non-usable "dummy" UDP socket for testing.
+    ///
+    /// This is a class method that returns a "mock" of UDP socket.
+    /// This is not associated with any actual socket, and its only
+    /// responsibility is to return \c IPPROTO_UDP from \c getProtocol().
+    /// The only feasible usage of this socket is for testing so that
+    /// the test code can prepare some "UDP data" even without opening any
+    /// actual socket.
+    ///
+    /// This method never throws an exception.
+    ///
+    /// \return A reference to an \c IOSocket object whose \c getProtocol()
+    /// returns \c IPPROTO_UDP.
+    static IOSocket& getDummyUDPSocket();
+
+    /// \brief Return a non-usable "dummy" TCP socket for testing.
+    ///
+    /// See \c getDummyUDPSocket().  This method is its TCP version.
+    ///
+    /// \return A reference to an \c IOSocket object whose \c getProtocol()
+    /// returns \c IPPROTO_TCP.
+    static IOSocket& getDummyTCPSocket();
+};
+
+/// \brief The \c IOMessage class encapsulates an incoming message received
+/// on a socket.
+///
+/// An \c IOMessage object represents a tuple of a chunk of data
+/// (a UDP packet or some segment of TCP stream), the socket over which the
+/// data is passed, the information about the other end point of the
+/// communication, and perhaps more.
+///
+/// The current design and interfaces of this class is tentative.
+/// It only provides a minimal level of support that is necessary for
+/// the current implementation of the authoritative server.
+/// A future version of this class will definitely support more.
+class IOMessage {
+    ///
+    /// \name Constructors and Destructor
+    ///
+    /// Note: The copy constructor and the assignment operator are
+    /// intentionally defined as private, making this class non-copyable.
+    //@{
+private:
+    IOMessage(const IOMessage& source);
+    IOMessage& operator=(const IOMessage& source);
+public:
+    /// \brief Constructor from message information.
+    ///
+    /// This constructor needs to handle the ASIO \c ip::address class,
+    /// and is intended to be used within this wrapper implementation.
+    /// Once the \c IOMessage object is created, the application can
+    /// get access to the information via the wrapper interface such as
+    /// \c getRemoteAddress().
+    ///
+    /// This constructor never throws an exception.
+    ///
+    /// \param data A pointer to the message data.
+    /// \param data_size The size of the message data in bytes.
+    /// \param io_socket The socket over which the data is given.
+    /// \param remote_endpoint The other endpoint of the socket, that is,
+    /// the sender of the message.
+    IOMessage(const void* data, size_t data_size, IOSocket& io_socket,
+              const IOEndpoint& remote_endpoint);
+    //@}
+
+    /// \brief Returns a pointer to the received data.
+    const void* getData() const { return (data_); }
+
+    /// \brief Returns the size of the received data in bytes.
+    size_t getDataSize() const { return (data_size_); }
+
+    /// \brief Returns the socket on which the message arrives.
+    const IOSocket& getSocket() const { return (io_socket_); }
+
+    /// \brief Returns the endpoint that sends the message.
+    const IOEndpoint& getRemoteEndpoint() const { return (remote_endpoint_); }
+private:
+    const void* data_;
+    const size_t data_size_;
+    IOSocket& io_socket_;
+    const IOEndpoint& remote_endpoint_;
+};
+
+/// \brief The \c IOService class is a wrapper for the ASIO \c io_service
+/// class.
+///
+/// Currently, the interface of this class is very specific to the
+/// authoritative server implementation as indicated in the signature of
+/// the constructor, but the plan is to generalize it so that other BIND 10
+/// modules can use this interface, too.
 class IOService {
+    ///
+    /// \name Constructors and Destructor
+    ///
+    /// These are currently very specific to the authoritative server
+    /// implementation.
+    ///
+    /// Note: The copy constructor and the assignment operator are
+    /// intentionally defined as private, making this class non-copyable.
+    //@{
+private:
+    IOService(const IOService& source);
+    IOService& operator=(const IOService& source);
 public:
-    IOService(AuthSrv* auth_server,
-              const char* const address, const char* const port,
+    /// \brief The constructor with a specific IP address and port on which
+    /// the services listen on.
+    IOService(AuthSrv* auth_server, const char& port, const char& address);
+    /// \brief The constructor with a specific port on which the services
+    /// listen on.
+    ///
+    /// It effectively listens on "any" IPv4 and/or IPv6 addresses.
+    /// IPv4/IPv6 services will be available if and only if \c use_ipv4
+    /// or \c use_ipv6 is \c true, respectively.
+    IOService(AuthSrv* auth_server, const char& port,
               const bool use_ipv4, const bool use_ipv6);
+    /// \brief The destructor.
     ~IOService();
+    //@}
+
+    /// \brief Start the underlying event loop.
+    ///
+    /// This method does not return control to the caller until
+    /// the \c stop() method is called via some handler.
     void run();
+
+    /// \brief Stop the underlying event loop.
+    ///
+    /// This will return the control to the caller of the \c run() method.
     void stop();
+
+    /// \brief Return the native \c io_service object used in this wrapper.
+    ///
+    /// This is a short term work around to support other BIND 10 modules
+    /// that share the same \c io_service with the authoritative server.
+    /// It will eventually be removed once the wrapper interface is
+    /// generalized.
     asio::io_service& get_io_service();
+
+    /// \brief A functor(-like) class that specifies a custom call back
+    /// invoked from the event loop instead of the embedded authoritative
+    /// server callbacks.
+    ///
+    /// Currently, the callback is intended to be used only for testing
+    /// purposes.  But we'll need a generic callback type like this to
+    /// generalize the wrapper interface.
+    typedef boost::function<void(const IOMessage& io_message)> IOCallBack;
+
+    /// \brief Set the custom call back invoked from the event loop.
+    ///
+    /// Right now this method is only for testing, but will eventually be
+    /// generalized.
+    void setCallBack(IOCallBack callback);
 private:
     IOServiceImpl* impl_;
 };

src/bin/auth/auth_srv.cc

@@ -14,6 +14,10 @@
 
 // $Id$
 
+#include <config.h>             // for UNUSED_PARAM
+
+#include <netinet/in.h>
+
 #include <algorithm>
 #include <cassert>
 #include <iostream>
@@ -40,19 +44,25 @@
 
 #include <cc/data.h>
 
+#include <xfr/xfrout_client.h>
+
 #include <auth/common.h>
 #include <auth/auth_srv.h>
+#include <auth/asio_link.h>
 
 #include <boost/lexical_cast.hpp>
 
 using namespace std;
 
 using namespace isc;
+using namespace isc::cc;
 using namespace isc::datasrc;
 using namespace isc::dns;
 using namespace isc::dns::rdata;
 using namespace isc::data;
 using namespace isc::config;
+using namespace isc::xfr;
+using namespace asio_link;
 
 class AuthSrvImpl {
 private:
@@ -60,12 +70,18 @@ private:
     AuthSrvImpl(const AuthSrvImpl& source);
     AuthSrvImpl& operator=(const AuthSrvImpl& source);
 public:
-    AuthSrvImpl(const bool use_cache);
-
+    AuthSrvImpl(const bool use_cache, AbstractXfroutClient& xfrout_client);
+    ~AuthSrvImpl();
     isc::data::ElementPtr setDbFile(const isc::data::ElementPtr config);
 
+    bool processNormalQuery(const IOMessage& io_message, Message& message,
+                            MessageRenderer& response_renderer);
+    bool processAxfrQuery(const IOMessage& io_message, Message& message,
+                            MessageRenderer& response_renderer);
+    bool processNotify(const IOMessage& io_message, Message& message, 
+                            MessageRenderer& response_renderer);
     std::string db_file_;
-    ModuleCCSession* cs_;
+    ModuleCCSession* config_session_;
     MetaDataSrc data_sources_;
     /// We keep a pointer to the currently running sqlite datasource
     /// so that we can specifically remove that one should the database
@@ -74,6 +90,11 @@ public:
 
     bool verbose_mode_;
 
+    AbstractSession* xfrin_session_;
+
+    bool xfrout_connected_;
+    AbstractXfroutClient& xfrout_client_;
+
     /// Currently non-configurable, but will be.
     static const uint16_t DEFAULT_LOCAL_UDPSIZE = 4096;
 
@@ -81,8 +102,12 @@ public:
     isc::datasrc::HotCache cache_;
 };
 
-AuthSrvImpl::AuthSrvImpl(const bool use_cache) :
-    cs_(NULL), verbose_mode_(false)
+AuthSrvImpl::AuthSrvImpl(const bool use_cache,
+                         AbstractXfroutClient& xfrout_client) :
+    config_session_(NULL), verbose_mode_(false),
+    xfrin_session_(NULL),
+    xfrout_connected_(false),
+    xfrout_client_(xfrout_client)
 {
     // cur_datasrc_ is automatically initialized by the default constructor,
     // effectively being an empty (sqlite) data source.  once ccsession is up
@@ -95,9 +120,17 @@ AuthSrvImpl::AuthSrvImpl(const bool use_cache) :
     cache_.setEnabled(use_cache);
 }
 
-AuthSrv::AuthSrv(const bool use_cache) : impl_(new AuthSrvImpl(use_cache)) {
+AuthSrvImpl::~AuthSrvImpl() {
+    if (xfrout_connected_) {
+        xfrout_client_.disconnect();
+        xfrout_connected_ = false;
+    }
 }
 
+AuthSrv::AuthSrv(const bool use_cache, AbstractXfroutClient& xfrout_client) :
+    impl_(new AuthSrvImpl(use_cache, xfrout_client))
+{}
+
 AuthSrv::~AuthSrv() {
     delete impl_;
 }
@@ -125,8 +158,9 @@ makeErrorMessage(Message& message, MessageRenderer& renderer,
     const Opcode& opcode = message.getOpcode();
     vector<QuestionPtr> questions;
 
-    // If this is an error to a query, we should also copy the question section.
-    if (opcode == Opcode::QUERY()) {
+    // If this is an error to a query or notify, we should also copy the
+    // question section.
+    if (opcode == Opcode::QUERY() || opcode == Opcode::NOTIFY()) {
         questions.assign(message.beginQuestion(), message.endQuestion());
     }
 
@@ -164,20 +198,26 @@ AuthSrv::getVerbose() const {
 }
 
 void
-AuthSrv::setConfigSession(ModuleCCSession* cs) {
-    impl_->cs_ = cs;
+AuthSrv::setXfrinSession(AbstractSession* xfrin_session) {
+    impl_->xfrin_session_ = xfrin_session;
+}
+
+void
+AuthSrv::setConfigSession(ModuleCCSession* config_session) {
+    impl_->config_session_ = config_session;
 }
 
 ModuleCCSession*
 AuthSrv::configSession() const {
-    return (impl_->cs_);
+    return (impl_->config_session_);
 }
 
 bool
-AuthSrv::processMessage(InputBuffer& request_buffer, Message& message,
-                        MessageRenderer& response_renderer,
-                        const bool udp_buffer)
+AuthSrv::processMessage(const IOMessage& io_message, Message& message,
+                        MessageRenderer& response_renderer)
 {
+    InputBuffer request_buffer(io_message.getData(), io_message.getDataSize());
+
     // First, check the header part.  If we fail even for the base header,
     // just drop the message.
     try {
@@ -186,7 +226,8 @@ AuthSrv::processMessage(InputBuffer& request_buffer, Message& message,
         // Ignore all responses.
         if (message.getHeaderFlag(MessageFlag::QR())) {
             if (impl_->verbose_mode_) {
-                cerr << "[b10-auth] received unexpected response, ignoring" << endl;
+                cerr << "[b10-auth] received unexpected response, ignoring"
+                     << endl;
             }
             return (false);
         }
@@ -199,8 +240,8 @@ AuthSrv::processMessage(InputBuffer& request_buffer, Message& message,
         message.fromWire(request_buffer);
     } catch (const DNSProtocolError& error) {
         if (impl_->verbose_mode_) {
-            cerr << "[b10-auth] returning " <<  error.getRcode().toText() << ": "
-                 << error.what() << endl;
+            cerr << "[b10-auth] returning " <<  error.getRcode().toText()
+                 << ": " << error.what() << endl;
         }
         makeErrorMessage(message, response_renderer, error.getRcode(),
                          impl_->verbose_mode_);
@@ -220,8 +261,9 @@ AuthSrv::processMessage(InputBuffer& request_buffer, Message& message,
 
     // Perform further protocol-level validation.
 
-    // In this implementation, we only support normal queries
-    if (message.getOpcode() != Opcode::QUERY()) {
+    if (message.getOpcode() == Opcode::NOTIFY()) {
+        return (impl_->processNotify(io_message, message, response_renderer));
+    } else if (message.getOpcode() != Opcode::QUERY()) {
         if (impl_->verbose_mode_) {
             cerr << "[b10-auth] unsupported opcode" << endl;
         }
@@ -236,6 +278,25 @@ AuthSrv::processMessage(InputBuffer& request_buffer, Message& message,
         return (true);
     }
 
+    ConstQuestionPtr question = *message.beginQuestion();
+    const RRType &qtype = question->getType();
+    if (qtype == RRType::AXFR()) {
+        return (impl_->processAxfrQuery(io_message, message,
+                                        response_renderer));
+    } else if (qtype == RRType::IXFR()) {
+        makeErrorMessage(message, response_renderer, Rcode::NOTIMP(),
+                         impl_->verbose_mode_);
+        return (true);
+    } else {
+        return (impl_->processNormalQuery(io_message, message,
+                                          response_renderer));
+    }
+}
+
+bool
+AuthSrvImpl::processNormalQuery(const IOMessage& io_message, Message& message,
+                                MessageRenderer& response_renderer)
+{
     const bool dnssec_ok = message.isDNSSECSupported();
     const uint16_t remote_bufsize = message.getUDPSize();
 
@@ -246,21 +307,23 @@ AuthSrv::processMessage(InputBuffer& request_buffer, Message& message,
     message.setUDPSize(AuthSrvImpl::DEFAULT_LOCAL_UDPSIZE);
 
     try {
-        Query query(message, impl_->cache_, dnssec_ok);
-        impl_->data_sources_.doQuery(query);
+        Query query(message, cache_, dnssec_ok);
+        data_sources_.doQuery(query);
     } catch (const Exception& ex) {
-        if (impl_->verbose_mode_) {
+        if (verbose_mode_) {
             cerr << "[b10-auth] Internal error, returning SERVFAIL: " <<
                 ex.what() << endl;
         }
         makeErrorMessage(message, response_renderer, Rcode::SERVFAIL(),
-                         impl_->verbose_mode_);
+                         verbose_mode_);
         return (true);
     }
 
+    const bool udp_buffer =
+        (io_message.getSocket().getProtocol() == IPPROTO_UDP);
     response_renderer.setLengthLimit(udp_buffer ? remote_bufsize : 65535);
     message.toWire(response_renderer);
-    if (impl_->verbose_mode_) {
+    if (verbose_mode_) {
         cerr << "[b10-auth] sending a response (" <<
             boost::lexical_cast<string>(response_renderer.getLength())
              << " bytes):\n" << message.toText() << endl;
@@ -269,6 +332,137 @@ AuthSrv::processMessage(InputBuffer& request_buffer, Message& message,
     return (true);
 }
 
+
+bool
+AuthSrvImpl::processAxfrQuery(const IOMessage& io_message, Message& message,
+                            MessageRenderer& response_renderer)
+{
+    if (io_message.getSocket().getProtocol() == IPPROTO_UDP) {
+        if (verbose_mode_) {
+            cerr << "[b10-auth] AXFR query over UDP isn't allowed" << endl;
+        }
+        makeErrorMessage(message, response_renderer, Rcode::FORMERR(),
+                         verbose_mode_);
+        return (true);
+    }
+
+    try {
+        if (!xfrout_connected_) {
+            xfrout_client_.connect();
+            xfrout_connected_ = true;
+        }
+        xfrout_client_.sendXfroutRequestInfo(
+            io_message.getSocket().getNative(),
+            io_message.getData(),
+            io_message.getDataSize());
+    } catch (const XfroutError& err) {
+        if (xfrout_connected_) {
+            // discoonect() may trigger an exception, but since we try it
+            // only if we've successfully opened it, it shouldn't happen in
+            // normal condition.  Should this occur, we'll propagate it to the
+            // upper layer.
+            xfrout_client_.disconnect();
+            xfrout_connected_ = false;
+        }
+        
+        if (verbose_mode_) {
+            cerr << "[b10-auth] Error in handling XFR request: " << err.what()
+                 << endl;
+        }
+        makeErrorMessage(message, response_renderer, Rcode::SERVFAIL(),
+                         verbose_mode_);
+        return (true);
+    }
+    return (false);
+}
+
+bool
+AuthSrvImpl::processNotify(const IOMessage& io_message, Message& message, 
+                           MessageRenderer& response_renderer) 
+{
+    // The incoming notify must contain exactly one question for SOA of the
+    // zone name.
+    if (message.getRRCount(Section::QUESTION()) != 1) {
+        if (verbose_mode_) {
+                cerr << "[b10-auth] invalid number of questions in notify: "
+                     << message.getRRCount(Section::QUESTION()) << endl;
+        }
+        makeErrorMessage(message, response_renderer, Rcode::FORMERR(),
+                         verbose_mode_);
+        return (true);
+    }
+    ConstQuestionPtr question = *message.beginQuestion();
+    if (question->getType() != RRType::SOA()) {
+        if (verbose_mode_) {
+                cerr << "[b10-auth] invalid question RR type in notify: "
+                     << question->getType() << endl;
+        }
+        makeErrorMessage(message, response_renderer, Rcode::FORMERR(),
+                         verbose_mode_);
+        return (true);
+    }
+
+    // According to RFC 1996, rcode should be "no error" and AA bit should be
+    // on, but we don't check these conditions.  This behavior is compatible
+    // with BIND 9.
+
+    // TODO check with the conf-mgr whether current server is the auth of the
+    // zone
+
+    // In the code that follows, we simply ignore the notify if any internal
+    // error happens rather than returning (e.g.) SERVFAIL.  RFC 1996 is
+    // silent about such cases, but there doesn't seem to be anything we can
+    // improve at the primary server side by sending an error anyway.
+    if (xfrin_session_ == NULL) {
+            if (verbose_mode_) {
+                cerr << "[b10-auth] "
+                    "session interface for xfrin is not available" << endl;
+            }
+            return (false);
+    }
+    
+    const string remote_ip_address =
+        io_message.getRemoteEndpoint().getAddress().toText();
+    static const string command_template_start =
+        "{\"command\": [\"notify\", {\"zone_name\" : \"";
+    static const string command_template_master = "\", \"master\" : \"";
+    static const string command_template_rrclass = "\", \"rrclass\" : \"";
+    static const string command_template_end = "\"}]}";
+
+    try {
+        ElementPtr notify_command = Element::fromJSON(
+                command_template_start + question->getName().toText() + 
+                command_template_master + remote_ip_address +
+                command_template_rrclass + question->getClass().toText() +
+                command_template_end);
+        const unsigned int seq =
+            xfrin_session_->group_sendmsg(notify_command, "Xfrin",
+                                          "*", "*");
+        ElementPtr env, answer, parsed_answer;
+        xfrin_session_->group_recvmsg(env, answer, false, seq);
+        int rcode;
+        parsed_answer = parseAnswer(rcode, answer);
+        if (rcode != 0) {
+            if (verbose_mode_) {
+                cerr << "[b10-auth] failed to notify Xfrin: "
+                     << parsed_answer->str() << endl; 
+            }
+            return (false);
+        }
+    } catch (const Exception& ex) {
+        if (verbose_mode_) {
+            cerr << "[b10-auth] failed to notify Xfrin: " << ex.what() << endl;
+        }
+        return (false);
+    }
+
+    message.makeResponse();
+    message.setHeaderFlag(MessageFlag::AA());
+    message.setRcode(Rcode::NOERROR());
+    message.toWire(response_renderer);
+    return (true);
+}
+
 ElementPtr
 AuthSrvImpl::setDbFile(const isc::data::ElementPtr config) {
     ElementPtr answer = isc::config::createAnswer();
@@ -277,10 +471,10 @@ AuthSrvImpl::setDbFile(const isc::data::ElementPtr config) {
     if (config && config->contains("database_file")) {
         db_file_ = config->get("database_file")->stringValue();
         final = config;
-    } else if (cs_ != NULL) {
+    } else if (config_session_ != NULL) {
         bool is_default;
         string item("database_file");
-        ElementPtr value = cs_->getValue(is_default, item);
+        ElementPtr value = config_session_->getValue(is_default, item);
         final = Element::createMap();
 
         // If the value is the default, and we are running from

src/bin/auth/auth_srv.h

@@ -28,6 +28,14 @@ class InputBuffer;
 class Message;
 class MessageRenderer;
 }
+
+namespace xfr {
+class AbstractXfroutClient;
+};
+}
+
+namespace asio_link {
+class IOMessage;
 }
 
 class AuthSrvImpl;
@@ -36,28 +44,49 @@ class AuthSrv {
     ///
     /// \name Constructors, Assignment Operator and Destructor.
     ///
-    /// Note: The copy constructor and the assignment operator are intentionally
-    /// defined as private.
+    /// Note: The copy constructor and the assignment operator are
+    /// intentionally defined as private.
     //@{
 private:
     AuthSrv(const AuthSrv& source);
     AuthSrv& operator=(const AuthSrv& source);
 public:
-    explicit AuthSrv(const bool use_cache);
+    /// The constructor.
+    ///
+    /// \param use_cache Whether to enable hot spot cache for lookup results.
+    /// \param xfrout_client Communication interface with a separate xfrout
+    /// process.  It's normally a reference to an xfr::XfroutClient object,
+    /// but can refer to a local mock object for testing (or other
+    /// experimental) purposes.
+    AuthSrv(const bool use_cache,
+            isc::xfr::AbstractXfroutClient& xfrout_client);
     ~AuthSrv();
     //@}
     /// \return \c true if the \message contains a response to be returned;
     /// otherwise \c false.
-    bool processMessage(isc::dns::InputBuffer& request_buffer,
+    bool processMessage(const asio_link::IOMessage& io_message,
                         isc::dns::Message& message,
-                        isc::dns::MessageRenderer& response_renderer,
-                        bool udp_buffer);
+                        isc::dns::MessageRenderer& response_renderer);
     void setVerbose(bool on);
     bool getVerbose() const;
-    void serve(std::string zone_name);
     isc::data::ElementPtr updateConfig(isc::data::ElementPtr config);
     isc::config::ModuleCCSession* configSession() const;
-    void setConfigSession(isc::config::ModuleCCSession* cs);
+    void setConfigSession(isc::config::ModuleCCSession* config_session);
+
+    ///
+    /// Note: this interface is tentative.  We'll revisit the ASIO and session
+    /// frameworks, at which point the session will probably be passed on
+    /// construction of the server.
+    ///
+    /// \param xfrin_session A Session object over which NOTIFY message
+    /// information is exchanged with a XFRIN handler.
+    /// The session must be established before setting in the server
+    /// object.
+    /// Ownership isn't transferred: the caller is responsible for keeping
+    /// this object to be valid while the server object is working and for
+    /// disconnecting the session and destroying the object when the server
+    ///
+    void setXfrinSession(isc::cc::AbstractSession* xfrin_session);
 private:
     AuthSrvImpl* impl_;
 };

src/bin/auth/main.cc

@@ -39,6 +39,8 @@
 #include <cc/data.h>
 #include <config/ccsession.h>
 
+#include <xfr/xfrout_client.h>
+
 #include <auth/spec_config.h>
 #include <auth/common.h>
 #include <auth/auth_srv.h>
@@ -49,6 +51,7 @@ using namespace isc::data;
 using namespace isc::cc;
 using namespace isc::config;
 using namespace isc::dns;
+using namespace isc::xfr;
 
 namespace {
 
@@ -86,7 +89,7 @@ my_command_handler(const string& command, const ElementPtr args) {
 
 void
 usage() {
-    cerr << "Usage: b10-auth [-p port] [-4|-6] [-nv]" << endl;
+    cerr << "Usage: b10-auth [-a address] [-p port] [-4|-6] [-nv]" << endl;
     exit(1);
 }
 } // end of anonymous namespace
@@ -138,9 +141,19 @@ main(int argc, char* argv[]) {
         usage();
     }
 
+    if ((!use_ipv4 || !use_ipv6) && address != NULL) {
+        cerr << "[b10-auth] Error: -4|-6 and -a can't coexist" << endl;
+        usage();
+    }
+
     int ret = 0;
+
+    // XXX: we should eventually pass io_service here.
     Session* cc_session = NULL;
-    ModuleCCSession* cs = NULL;
+    Session* xfrin_session = NULL;
+    bool xfrin_session_established = false; // XXX (see Trac #287)
+    ModuleCCSession* config_session = NULL;
+    XfroutClient xfrout_client(UNIX_SOCKET_FILE);
     try {
         string specfile;
         if (getenv("B10_FROM_BUILD")) {
@@ -150,22 +163,46 @@ main(int argc, char* argv[]) {
             specfile = string(AUTH_SPECFILE_LOCATION);
         }
 
-        auth_server = new AuthSrv(cache);
+        auth_server = new AuthSrv(cache, xfrout_client);
         auth_server->setVerbose(verbose_mode);
         cout << "[b10-auth] Server created." << endl;
 
-        io_service = new asio_link::IOService(auth_server, address, port,
-                                              use_ipv4, use_ipv6);
+        if (address != NULL) {
+            // XXX: we can only specify at most one explicit address.
+            // This also means the server cannot run in the dual address
+            // family mode if explicit addresses need to be specified.
+            // We don't bother to fix this problem, however.  The -a option
+            // is a short term workaround until we support dynamic listening
+            // port allocation.
+            io_service = new asio_link::IOService(auth_server, *port,
+                                                  *address);
+        } else {
+            io_service = new asio_link::IOService(auth_server, *port,
+                                                  use_ipv4, use_ipv6);
+        }
         cout << "[b10-auth] IOService created." << endl;
 
         cc_session = new Session(io_service->get_io_service());
-        cout << "[b10-auth] Session channel created." << endl;
+        cout << "[b10-auth] Configuration session channel created." << endl;
 
-        cs = new ModuleCCSession(specfile, *cc_session, my_config_handler,
-                                 my_command_handler);
+        config_session = new ModuleCCSession(specfile, *cc_session,
+                                             my_config_handler,
+                                             my_command_handler);
         cout << "[b10-auth] Configuration channel established." << endl;
 
-        auth_server->setConfigSession(cs);
+        xfrin_session = new Session(io_service->get_io_service());
+        cout << "[b10-auth] Xfrin session channel created." << endl;
+        xfrin_session->establish(NULL);
+        xfrin_session_established = true;
+        cout << "[b10-auth] Xfrin session channel established." << endl;
+
+        // XXX: with the current interface to asio_link we have to create
+        // auth_server before io_service while Session needs io_service.
+        // In a next step of refactoring we should make asio_link independent
+        // from auth_server, and create io_service, auth_server, and
+        // sessions in that order.
+        auth_server->setXfrinSession(xfrin_session);
+        auth_server->setConfigSession(config_session);
         auth_server->updateConfig(ElementPtr());
 
         cout << "[b10-auth] Server started." << endl;
@@ -175,9 +212,15 @@ main(int argc, char* argv[]) {
         ret = 1;
     }
 
-    delete cs;
+    if (xfrin_session_established) {
+        xfrin_session->disconnect();
+    }
+
+    delete xfrin_session;
+    delete config_session;
     delete cc_session;
     delete io_service;
     delete auth_server;
+
     return (ret);
 }

src/bin/auth/tests/Makefile.am

@@ -14,6 +14,7 @@ run_unittests_SOURCES = $(top_srcdir)/src/lib/dns/tests/unittest_util.h
 run_unittests_SOURCES += $(top_srcdir)/src/lib/dns/tests/unittest_util.cc
 run_unittests_SOURCES += ../auth_srv.h ../auth_srv.cc
 run_unittests_SOURCES += auth_srv_unittest.cc
+run_unittests_SOURCES += asio_link_unittest.cc
 run_unittests_SOURCES += run_unittests.cc
 run_unittests_CPPFLAGS = $(AM_CPPFLAGS) $(GTEST_INCLUDES)
 run_unittests_LDFLAGS = $(AM_LDFLAGS) $(GTEST_LDFLAGS)
@@ -24,6 +25,8 @@ run_unittests_LDADD +=  $(top_builddir)/src/lib/dns/.libs/libdns++.a
 run_unittests_LDADD += $(top_builddir)/src/lib/config/.libs/libcfgclient.a
 run_unittests_LDADD += $(top_builddir)/src/lib/cc/.libs/libcc.a
 run_unittests_LDADD += $(top_builddir)/src/lib/exceptions/.libs/libexceptions.a
+run_unittests_LDADD += $(top_builddir)/src/bin/auth/libasio_link.a
+run_unittests_LDADD += $(top_builddir)/src/lib/xfr/.libs/libxfr.a
 endif
 
 noinst_PROGRAMS = $(TESTS)

src/bin/auth/tests/asio_link_unittest.cc

@@ -0,0 +1,351 @@
+// Copyright (C) 2010  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.
+
+// $Id$
+
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netdb.h>
+
+#include <stdint.h>
+
+#include <functional>
+#include <string>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include <exceptions/exceptions.h>
+
+#include <dns/tests/unittest_util.h>
+
+#include <auth/asio_link.h>
+
+using isc::UnitTestUtil;
+using namespace std;
+using namespace asio_link;
+
+namespace {
+const char* const TEST_PORT = "53535";
+const char* const TEST_IPV6_ADDR = "::1";
+const char* const TEST_IPV4_ADDR = "127.0.0.1";
+// This data is intended to be valid as a DNS/TCP-like message: the first
+// two octets encode the length of the rest of the data.  This is crucial
+// for the tests below.
+const uint8_t test_data[] = {0, 4, 1, 2, 3, 4};
+
+TEST(IOAddressTest, fromText) {
+    IOAddress io_address_v4("192.0.2.1");
+    EXPECT_EQ("192.0.2.1", io_address_v4.toText());
+
+    IOAddress io_address_v6("2001:db8::1234");
+    EXPECT_EQ("2001:db8::1234", io_address_v6.toText());
+
+    // bogus IPv4 address-like input
+    EXPECT_THROW(IOAddress("192.0.2.2.1"), IOError);
+
+    // bogus IPv6 address-like input
+    EXPECT_THROW(IOAddress("2001:db8:::1234"), IOError);
+}
+
+TEST(IOEndpointTest, create) {
+    const IOEndpoint* ep;
+    ep = IOEndpoint::create(IPPROTO_UDP, IOAddress("192.0.2.1"), 5300);
+    EXPECT_EQ("192.0.2.1", ep->getAddress().toText());
+    delete ep;
+
+    ep = IOEndpoint::create(IPPROTO_TCP, IOAddress("192.0.2.1"), 5300);
+    EXPECT_EQ("192.0.2.1", ep->getAddress().toText());
+    delete ep;
+
+    ep = IOEndpoint::create(IPPROTO_UDP, IOAddress("2001:db8::1234"), 5300);
+    EXPECT_EQ("2001:db8::1234", ep->getAddress().toText());
+    delete ep;
+
+    ep = IOEndpoint::create(IPPROTO_TCP, IOAddress("2001:db8::1234"), 5300);
+    EXPECT_EQ("2001:db8::1234", ep->getAddress().toText());
+    delete ep;
+
+    EXPECT_THROW(IOEndpoint::create(IPPROTO_IP, IOAddress("192.0.2.1"),
+                                    5300)->getAddress().toText(),
+                 IOError);
+}
+
+TEST(IOSocketTest, dummySockets) {
+    EXPECT_EQ(IPPROTO_UDP, IOSocket::getDummyUDPSocket().getProtocol());
+    EXPECT_EQ(IPPROTO_TCP, IOSocket::getDummyTCPSocket().getProtocol());
+    EXPECT_EQ(-1, IOSocket::getDummyUDPSocket().getNative());
+    EXPECT_EQ(-1, IOSocket::getDummyTCPSocket().getNative());
+}
+
+TEST(IOServiceTest, badPort) {
+    EXPECT_THROW(IOService(NULL, *"65536", true, false), IOError);
+    EXPECT_THROW(IOService(NULL, *"5300.0", true, false), IOError);
+    EXPECT_THROW(IOService(NULL, *"-1", true, false), IOError);
+    EXPECT_THROW(IOService(NULL, *"domain", true, false), IOError);
+}
+
+TEST(IOServiceTest, badAddress) {
+    EXPECT_THROW(IOService(NULL, *TEST_PORT, *"192.0.2.1.1"),
+                 IOError);
+    EXPECT_THROW(IOService(NULL, *TEST_PORT, *"2001:db8:::1"),
+                 IOError);
+    EXPECT_THROW(IOService(NULL, *TEST_PORT, *"localhost"),
+                 IOError);
+}
+
+TEST(IOServiceTest, unavailableAddress) {
+    // These addresses should generally be unavailable as a valid local
+    // address, although there's no guarantee in theory.
+    EXPECT_THROW(IOService(NULL, *TEST_PORT, *"ffff:ffff::"), IOError);
+    EXPECT_THROW(IOService(NULL, *TEST_PORT, *"255.255.0.0"), IOError);
+}
+
+TEST(IOServiceTest, duplicateBind) {
+    // In each sub test case, second attempt should fail due to duplicate bind
+
+    // IPv6, "any" address
+    IOService* io_service = new IOService(NULL, *TEST_PORT, false, true);
+    EXPECT_THROW(IOService(NULL, *TEST_PORT, false, true), IOError);
+    delete io_service;
+
+    // IPv6, specific address
+    io_service = new IOService(NULL, *TEST_PORT, *TEST_IPV6_ADDR);
+    EXPECT_THROW(IOService(NULL, *TEST_PORT, *TEST_IPV6_ADDR), IOError);
+    delete io_service;
+
+    // IPv4, "any" address
+    io_service = new IOService(NULL, *TEST_PORT, true, false);
+    EXPECT_THROW(IOService(NULL, *TEST_PORT, true, false), IOError);
+    delete io_service;
+
+    // IPv4, specific address
+    io_service = new IOService(NULL, *TEST_PORT, *TEST_IPV4_ADDR);
+    EXPECT_THROW(IOService(NULL, *TEST_PORT, *TEST_IPV4_ADDR), IOError);
+    delete io_service;
+}
+
+struct addrinfo*
+resolveAddress(const int family, const int sock_type, const int protocol) {
+    const char* const addr = (family == AF_INET6) ?
+        TEST_IPV6_ADDR : TEST_IPV4_ADDR;
+
+    struct addrinfo hints;
+    memset(&hints, 0, sizeof(hints));
+    hints.ai_family = family;
+    hints.ai_socktype = sock_type;
+    hints.ai_protocol = protocol;
+
+    struct addrinfo* res;
+    const int error = getaddrinfo(addr, TEST_PORT, &hints, &res);
+    if (error != 0) {
+        isc_throw(IOError, "getaddrinfo failed: " << gai_strerror(error));
+    }
+
+    return (res);
+}
+
+// This fixture is a framework for various types of network operations
+// using the ASIO interfaces.  Each test case creates an IOService object,
+// opens a local "client" socket for testing, sends data via the local socket
+// to the service that would run in the IOService object.
+// A mock callback function (an ASIOCallBack object) is registered with the
+// IOService object, so the test code should be able to examine the data
+// receives on the server side.  It then checks the received data matches
+// expected parameters.
+// If initialization parameters of the IOService should be modified, the test
+// case can do it using the setIOService() method.
+// Note: the set of tests in ASIOLinkTest use actual network services and may
+// involve undesirable side effect such as blocking.
+class ASIOLinkTest : public ::testing::Test {
+protected:
+    ASIOLinkTest();
+    ~ASIOLinkTest() {
+        if (res_ != NULL) {
+            freeaddrinfo(res_);
+        }
+        if (sock_ != -1) {
+            close(sock_);
+        }
+        delete io_service_;
+    }
+    void sendUDP(const int family) {
+        res_ = resolveAddress(family, SOCK_DGRAM, IPPROTO_UDP);
+
+        sock_ = socket(res_->ai_family, res_->ai_socktype, res_->ai_protocol);
+        if (sock_ < 0) {
+            isc_throw(IOError, "failed to open test socket");
+        }
+        const int cc = sendto(sock_, test_data, sizeof(test_data), 0,
+                              res_->ai_addr, res_->ai_addrlen);
+        if (cc != sizeof(test_data)) {
+            isc_throw(IOError, "unexpected sendto result: " << cc);
+        }
+        io_service_->run();
+    }
+    void sendTCP(const int family) {
+        res_ = resolveAddress(family, SOCK_STREAM, IPPROTO_TCP);
+
+        sock_ = socket(res_->ai_family, res_->ai_socktype, res_->ai_protocol);
+        if (sock_ < 0) {
+            isc_throw(IOError, "failed to open test socket");
+        }
+        if (connect(sock_, res_->ai_addr, res_->ai_addrlen) < 0) {
+            isc_throw(IOError, "failed to connect to the test server");
+        }
+        const int cc = send(sock_, test_data, sizeof(test_data), 0);
+        if (cc != sizeof(test_data)) {
+            isc_throw(IOError, "unexpected sendto result: " << cc);
+        }
+        io_service_->run();
+    }
+    void setIOService(const char& address) {
+        delete io_service_;
+        io_service_ = NULL;
+        io_service_ = new IOService(NULL, *TEST_PORT, address);
+        io_service_->setCallBack(ASIOCallBack(this));
+    }
+    void setIOService(const bool use_ipv4, const bool use_ipv6) {
+        delete io_service_;
+        io_service_ = NULL;
+        io_service_ = new IOService(NULL, *TEST_PORT, use_ipv4, use_ipv6);
+        io_service_->setCallBack(ASIOCallBack(this));
+    }
+    void doTest(const int family, const int protocol) {
+        if (protocol == IPPROTO_UDP) {
+            sendUDP(family);
+        } else {
+            sendTCP(family);
+        }
+
+        // There doesn't seem to be an effective test for the validity of
+        // 'native'.
+        // One thing we are sure is it must be different from our local socket.
+        EXPECT_NE(sock_, callback_native_);
+        EXPECT_EQ(protocol, callback_protocol_);
+        EXPECT_EQ(family == AF_INET6 ? TEST_IPV6_ADDR : TEST_IPV4_ADDR,
+                  callback_address_);
+
+        const uint8_t* expected_data =
+            protocol == IPPROTO_UDP ? test_data : test_data + 2;
+        const size_t expected_datasize =
+            protocol == IPPROTO_UDP ? sizeof(test_data) :
+            sizeof(test_data) - 2;
+        EXPECT_PRED_FORMAT4(UnitTestUtil::matchWireData, &callback_data_[0],
+                            callback_data_.size(),
+                            expected_data, expected_datasize);
+    }
+private:
+    class ASIOCallBack : public std::unary_function<IOMessage, void> {
+    public:
+        ASIOCallBack(ASIOLinkTest* test_obj) : test_obj_(test_obj) {}
+        void operator()(const IOMessage& io_message) const {
+            test_obj_->callBack(io_message);
+        }
+    private:
+        ASIOLinkTest* test_obj_;
+    };
+    void callBack(const IOMessage& io_message) {
+        callback_protocol_ = io_message.getSocket().getProtocol();
+        callback_native_ = io_message.getSocket().getNative();
+        callback_address_ =
+            io_message.getRemoteEndpoint().getAddress().toText();
+        callback_data_.assign(
+            static_cast<const uint8_t*>(io_message.getData()),
+            static_cast<const uint8_t*>(io_message.getData()) +
+            io_message.getDataSize());
+        io_service_->stop();
+    }
+protected:
+    IOService* io_service_;
+    int callback_protocol_;
+    int callback_native_;
+    string callback_address_;
+    vector<uint8_t> callback_data_;
+    int sock_;
+private:
+    struct addrinfo* res_;
+};
+
+ASIOLinkTest::ASIOLinkTest() :
+    io_service_(NULL), sock_(-1), res_(NULL)
+{
+    setIOService(true, true);
+}
+
+TEST_F(ASIOLinkTest, v6UDPSend) {
+    doTest(AF_INET6, IPPROTO_UDP);
+}
+
+TEST_F(ASIOLinkTest, v6TCPSend) {
+    doTest(AF_INET6, IPPROTO_TCP);
+}
+
+TEST_F(ASIOLinkTest, v4UDPSend) {
+    doTest(AF_INET, IPPROTO_UDP);
+}
+
+TEST_F(ASIOLinkTest, v4TCPSend) {
+    doTest(AF_INET, IPPROTO_TCP);
+}
+
+TEST_F(ASIOLinkTest, v6UDPSendSpecific) {
+    // Explicitly set a specific address to be bound to the socket.
+    // The subsequent test does not directly ensures the underlying socket
+    // is bound to the expected address, but the success of the tests should
+    // reasonably suggest it works as intended.
+    // Specifying an address also implicitly means the service runs in a
+    // single address-family mode.  In tests using TCP we can confirm that
+    // by trying to make a connection and seeing a failure.  In UDP, it'd be
+    // more complicated because we need to use a connected socket and catch
+    // an error on a subsequent read operation.  We could do it, but for
+    // simplicity we only tests the easier cases for now.
+
+    setIOService(*TEST_IPV6_ADDR);
+    doTest(AF_INET6, IPPROTO_UDP);
+}
+
+TEST_F(ASIOLinkTest, v6TCPSendSpecific) {
+    setIOService(*TEST_IPV6_ADDR);
+    doTest(AF_INET6, IPPROTO_TCP);
+
+    EXPECT_THROW(sendTCP(AF_INET), IOError);
+}
+
+TEST_F(ASIOLinkTest, v4UDPSendSpecific) {
+    setIOService(*TEST_IPV4_ADDR);
+    doTest(AF_INET, IPPROTO_UDP);
+}
+
+TEST_F(ASIOLinkTest, v4TCPSendSpecific) {
+    setIOService(*TEST_IPV4_ADDR);
+    doTest(AF_INET, IPPROTO_TCP);
+
+    EXPECT_THROW(sendTCP(AF_INET6), IOError);
+}
+
+TEST_F(ASIOLinkTest, v6TCPOnly) {
+    // Open only IPv6 TCP socket.  A subsequent attempt of establishing an
+    // IPv4/TCP connection should fail.  See above for why we only test this
+    // for TCP.
+    setIOService(false, true);
+    EXPECT_THROW(sendTCP(AF_INET), IOError);
+}
+
+TEST_F(ASIOLinkTest, v4TCPOnly) {
+    setIOService(true, false);
+    EXPECT_THROW(sendTCP(AF_INET6), IOError);
+}
+
+}

src/bin/auth/tests/auth_srv_unittest.cc

@@ -14,6 +14,8 @@
 
 // $Id$
 
+#include <config.h>
+
 #include <gtest/gtest.h>
 
 #include <dns/buffer.h>
@@ -24,33 +26,105 @@
 #include <dns/rrtype.h>
 
 #include <cc/data.h>
+#include <cc/session.h>
+
+#include <xfr/xfrout_client.h>
 
 #include <auth/auth_srv.h>
+#include <auth/asio_link.h>
 
 #include <dns/tests/unittest_util.h>
 
 using isc::UnitTestUtil;
 using namespace std;
+using namespace isc::cc;
 using namespace isc::dns;
 using namespace isc::data;
+using namespace isc::xfr;
+using namespace asio_link;
 
 namespace {
-const char* CONFIG_TESTDB =
+const char* const CONFIG_TESTDB =
     "{\"database_file\": \"" TEST_DATA_DIR "/example.sqlite3\"}";
 // The following file must be non existent and must be non"creatable" (see
 // the sqlite3 test).
-const char* BADCONFIG_TESTDB =
+const char* const BADCONFIG_TESTDB =
     "{ \"database_file\": \"" TEST_DATA_DIR "/nodir/notexist\"}";
+const char* const DEFAULT_REMOTE_ADDRESS = "192.0.2.1";
 
 class AuthSrvTest : public ::testing::Test {
+private:
+    class MockXfroutClient : public AbstractXfroutClient {
+    public:
+        MockXfroutClient() :
+            is_connected_(false), connect_ok_(true), send_ok_(true),
+            disconnect_ok_(true)
+        {}
+        virtual void connect();
+        virtual void disconnect();
+        virtual int sendXfroutRequestInfo(int tcp_sock, const void* msg_data,
+                                          uint16_t msg_len);
+        bool isConnected() const { return (is_connected_); }
+        void disableConnect() { connect_ok_ = false; }
+        void disableDisconnect() { disconnect_ok_ = false; }
+        void enableDisconnect() { disconnect_ok_ = true; }
+        void disableSend() { send_ok_ = false; }
+    private:
+        bool is_connected_;
+        bool connect_ok_;
+        bool send_ok_;
+        bool disconnect_ok_;
+    };
+
+    class MockSession : public AbstractSession {
+    public:
+        MockSession() :
+            // by default we return a simple "success" message.
+            msg_(Element::fromJSON("{\"result\": [0, \"SUCCESS\"]}")),
+            send_ok_(true), receive_ok_(true)
+        {}
+        virtual void establish(const char* socket_file);
+        virtual void disconnect();
+        virtual int group_sendmsg(ElementPtr msg, string group,
+                                  string instance, string to);
+        virtual bool group_recvmsg(ElementPtr& envelope, ElementPtr& msg,
+                                   bool nonblock, int seq);
+        virtual void subscribe(string group, string instance);
+        virtual void unsubscribe(string group, string instance);
+        virtual void startRead(boost::function<void()> read_callback);
+        virtual int reply(ElementPtr& envelope, ElementPtr& newmsg);
+        virtual bool hasQueuedMsgs();
+
+        void setMessage(ElementPtr msg) { msg_ = msg; }
+        void disableSend() { send_ok_ = false; }
+        void disableReceive() { receive_ok_ = false; }
+
+        ElementPtr sent_msg;
+        string msg_destination;
+    private:
+        ElementPtr msg_;
+        bool send_ok_;
+        bool receive_ok_;
+    };
+
 protected:
-    AuthSrvTest() : server(true), request_message(Message::RENDER),
+    AuthSrvTest() : server(true, xfrout),
+                    request_message(Message::RENDER),
                     parse_message(Message::PARSE), default_qid(0x1035),
                     opcode(Opcode(Opcode::QUERY())), qname("www.example.com"),
-                    qclass(RRClass::IN()), qtype(RRType::A()), ibuffer(NULL),
-                    request_obuffer(0), request_renderer(request_obuffer),
+                    qclass(RRClass::IN()), qtype(RRType::A()),
+                    io_message(NULL), endpoint(NULL), request_obuffer(0),
+                    request_renderer(request_obuffer),
                     response_obuffer(0), response_renderer(response_obuffer)
-    {}
+    {
+        server.setXfrinSession(&notify_session);
+    }
+    ~AuthSrvTest() {
+        delete io_message;
+        delete endpoint;
+    }
+    MockSession notify_session;
+    MockXfroutClient xfrout;
     AuthSrv server;
     Message request_message;
     Message parse_message;
@@ -59,16 +133,114 @@ protected:
     const Name qname;
     const RRClass qclass;
     const RRType qtype;
-    InputBuffer* ibuffer;
+    IOMessage* io_message;
+    const IOEndpoint* endpoint;
     OutputBuffer request_obuffer;
     MessageRenderer request_renderer;
     OutputBuffer response_obuffer;
     MessageRenderer response_renderer;
     vector<uint8_t> data;
 
-    void createDataFromFile(const char* const datafile);
+    void createDataFromFile(const char* const datafile, int protocol);
+    void createRequestMessage(const Opcode& opcode, const Name& request_name,
+                              const RRClass& rrclass, const RRType& rrtype);
+    void createRequestPacket(const Opcode& opcode, const Name& request_name,
+                             const RRClass& rrclass, const RRType& rrtype,
+                             int protocol);
+    void createRequestPacket(int protocol);
 };
 
+void
+AuthSrvTest::MockSession::establish(const char* socket_file UNUSED_PARAM) {}
+
+void
+AuthSrvTest::MockSession::disconnect() {}
+
+void
+AuthSrvTest::MockSession::subscribe(string group UNUSED_PARAM,
+                                    string instance UNUSED_PARAM)
+{}
+
+void
+AuthSrvTest::MockSession::unsubscribe(string group UNUSED_PARAM,
+                                      string instance UNUSED_PARAM)
+{}
+
+void
+AuthSrvTest::MockSession::startRead(
+    boost::function<void()> read_callback UNUSED_PARAM)
+{}
+
+int
+AuthSrvTest::MockSession::reply(ElementPtr& envelope UNUSED_PARAM,
+                                ElementPtr& newmsg UNUSED_PARAM)
+{
+    return (-1);
+}
+
+bool
+AuthSrvTest::MockSession::hasQueuedMsgs() {
+    return (false);
+}
+
+int
+AuthSrvTest::MockSession::group_sendmsg(ElementPtr msg, string group,
+                                        string instance UNUSED_PARAM,
+                                        string to UNUSED_PARAM)
+{
+    if (!send_ok_) {
+        isc_throw(XfroutError, "mock session send is disabled for test");
+    }
+
+    sent_msg = msg;
+    msg_destination = group;
+    return (0);
+}
+
+bool
+AuthSrvTest::MockSession::group_recvmsg(ElementPtr& envelope UNUSED_PARAM,
+                                        ElementPtr& msg,
+                                        bool nonblock UNUSED_PARAM,
+                                        int seq UNUSED_PARAM)
+{
+    if (!receive_ok_) {
+        isc_throw(XfroutError, "mock session receive is disabled for test");
+    }
+
+    msg = msg_;
+    return (true);
+}
+
+void
+AuthSrvTest::MockXfroutClient::connect() {
+    if (!connect_ok_) {
+        isc_throw(XfroutError, "xfrout connection disabled for test");
+    }
+    is_connected_ = true;
+}
+
+void
+AuthSrvTest::MockXfroutClient::disconnect() {
+    if (!disconnect_ok_) {
+        isc_throw(XfroutError,
+                  "closing xfrout connection is disabled for test");
+    }
+    is_connected_ = false;
+}
+
+int
+AuthSrvTest::MockXfroutClient::sendXfroutRequestInfo(
+    const int tcp_sock UNUSED_PARAM,
+    const void* msg_data UNUSED_PARAM,
+    const uint16_t msg_len UNUSED_PARAM)
+{
+    if (!send_ok_) {
+        isc_throw(XfroutError, "xfrout connection send is disabled for test");
+    }
+    return (0);
+}
+
+
 // These are flags to indicate whether the corresponding flag bit of the
 // DNS header is to be set in the test cases.  (Note that the flag values
 // is irrelevant to their wire-format values)
@@ -81,12 +253,56 @@ const unsigned int AD_FLAG = 0x20;
 const unsigned int CD_FLAG = 0x40;
 
 void
-AuthSrvTest::createDataFromFile(const char* const datafile) {
-    delete ibuffer;
+AuthSrvTest::createDataFromFile(const char* const datafile,
+                                const int protocol = IPPROTO_UDP)
+{
+    delete io_message;
     data.clear();
 
+    delete endpoint;
+    endpoint = IOEndpoint::create(protocol,
+                                  IOAddress(DEFAULT_REMOTE_ADDRESS), 5300);
     UnitTestUtil::readWireData(datafile, data);
-    ibuffer = new InputBuffer(&data[0], data.size());
+    io_message = new IOMessage(&data[0], data.size(),
+                               protocol == IPPROTO_UDP ?
+                               IOSocket::getDummyUDPSocket() :
+                               IOSocket::getDummyTCPSocket(), *endpoint);
+}
+
+void
+AuthSrvTest::createRequestMessage(const Opcode& opcode,
+                                  const Name& request_name,
+                                  const RRClass& rrclass,
+                                  const RRType& rrtype)
+{
+    request_message.clear(Message::RENDER);
+    request_message.setOpcode(opcode);
+    request_message.setQid(default_qid);
+    request_message.addQuestion(Question(request_name, rrclass, rrtype));
+}
+
+void
+AuthSrvTest::createRequestPacket(const Opcode& opcode,
+                                 const Name& request_name,
+                                 const RRClass& rrclass, const RRType& rrtype,
+                                 const int protocol = IPPROTO_UDP)
+{
+    createRequestMessage(opcode, request_name, rrclass, rrtype);
+    createRequestPacket(protocol);
+}
+
+void
+AuthSrvTest::createRequestPacket(const int protocol = IPPROTO_UDP) {
+    request_message.toWire(request_renderer);
+
+    delete io_message;
+    endpoint = IOEndpoint::create(protocol,
+                                  IOAddress(DEFAULT_REMOTE_ADDRESS), 5300);
+    io_message = new IOMessage(request_renderer.getData(),
+                               request_renderer.getLength(),
+                               protocol == IPPROTO_UDP ?
+                               IOSocket::getDummyUDPSocket() :
+                               IOSocket::getDummyTCPSocket(), *endpoint);
 }
 
 void
@@ -115,15 +331,19 @@ headerCheck(const Message& message, const qid_t qid, const Rcode& rcode,
 
 // Unsupported requests.  Should result in NOTIMP.
 TEST_F(AuthSrvTest, unsupportedRequest) {
-    for (unsigned int i = 1; i < 16; ++i) {
+    for (unsigned int i = 0; i < 16; ++i) {
         // set Opcode to 'i', which iterators over all possible codes except
-        // the standard query (0)
+        // the standard query and notify
+        if (i == Opcode::QUERY().getCode() ||
+            i == Opcode::NOTIFY().getCode()) {
+            continue;
+        }
         createDataFromFile("simplequery_fromWire");
         data[2] = ((i << 3) & 0xff);
 
         parse_message.clear(Message::PARSE);
-        EXPECT_EQ(true, server.processMessage(*ibuffer, parse_message,
-                                              response_renderer, true));
+        EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                              response_renderer));
         headerCheck(parse_message, default_qid, Rcode::NOTIMP(), i, QR_FLAG,
                     0, 0, 0, 0);
     }
@@ -141,8 +361,8 @@ TEST_F(AuthSrvTest, verbose) {
 // Multiple questions.  Should result in FORMERR.
 TEST_F(AuthSrvTest, multiQuestion) {
     createDataFromFile("multiquestion_fromWire");
-    EXPECT_EQ(true, server.processMessage(*ibuffer, parse_message,
-                                          response_renderer, true));
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
     headerCheck(parse_message, default_qid, Rcode::FORMERR(), opcode.getCode(),
                 QR_FLAG, 2, 0, 0, 0);
 
@@ -162,8 +382,8 @@ TEST_F(AuthSrvTest, multiQuestion) {
 // dropped.
 TEST_F(AuthSrvTest, shortMessage) {
     createDataFromFile("shortmessage_fromWire");
-    EXPECT_EQ(false, server.processMessage(*ibuffer, parse_message,
-                                           response_renderer, true));
+    EXPECT_EQ(false, server.processMessage(*io_message, parse_message,
+                                           response_renderer));
 }
 
 // Response messages.  Must be silently dropped, whether it's a valid response
@@ -171,26 +391,26 @@ TEST_F(AuthSrvTest, shortMessage) {
 TEST_F(AuthSrvTest, response) {
     // A valid (although unusual) response
     createDataFromFile("simpleresponse_fromWire");
-    EXPECT_EQ(false, server.processMessage(*ibuffer, parse_message,
-                                           response_renderer, true));
+    EXPECT_EQ(false, server.processMessage(*io_message, parse_message,
+                                           response_renderer));
 
     // A response with a broken question section.  must be dropped rather than
     // returning FORMERR.
     createDataFromFile("shortresponse_fromWire");
-    EXPECT_EQ(false, server.processMessage(*ibuffer, parse_message,
-                                           response_renderer, true));
+    EXPECT_EQ(false, server.processMessage(*io_message, parse_message,
+                                           response_renderer));
 
     // A response to iquery.  must be dropped rather than returning NOTIMP.
     createDataFromFile("iqueryresponse_fromWire");
-    EXPECT_EQ(false, server.processMessage(*ibuffer, parse_message,
-                                           response_renderer, true));
+    EXPECT_EQ(false, server.processMessage(*io_message, parse_message,
+                                           response_renderer));
 }
 
 // Query with a broken question
 TEST_F(AuthSrvTest, shortQuestion) {
     createDataFromFile("shortquestion_fromWire");
-    EXPECT_EQ(true, server.processMessage(*ibuffer, parse_message,
-                                          response_renderer, true));
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
     // Since the query's question is broken, the question section of the
     // response should be empty.
     headerCheck(parse_message, default_qid, Rcode::FORMERR(), opcode.getCode(),
@@ -200,8 +420,8 @@ TEST_F(AuthSrvTest, shortQuestion) {
 // Query with a broken answer section
 TEST_F(AuthSrvTest, shortAnswer) {
     createDataFromFile("shortanswer_fromWire");
-    EXPECT_EQ(true, server.processMessage(*ibuffer, parse_message,
-                                          response_renderer, true));
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
 
     // This is a bogus query, but question section is valid.  So the response
     // should copy the question section.
@@ -219,8 +439,8 @@ TEST_F(AuthSrvTest, shortAnswer) {
 // Query with unsupported version of EDNS.
 TEST_F(AuthSrvTest, ednsBadVers) {
     createDataFromFile("queryBadEDNS_fromWire");
-    EXPECT_EQ(true, server.processMessage(*ibuffer, parse_message,
-                                          response_renderer, true));
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
 
     // The response must have an EDNS OPT RR in the additional section.
     // Note that the DNSSEC DO bit is cleared even if this bit in the query
@@ -231,6 +451,242 @@ TEST_F(AuthSrvTest, ednsBadVers) {
     EXPECT_FALSE(parse_message.isDNSSECSupported());
 }
 
+TEST_F(AuthSrvTest, AXFROverUDP) {
+    // AXFR over UDP is invalid and should result in FORMERR.
+    createRequestPacket(opcode, Name("example.com"), RRClass::IN(),
+                        RRType::AXFR(), IPPROTO_UDP);
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
+    headerCheck(parse_message, default_qid, Rcode::FORMERR(), opcode.getCode(),
+                QR_FLAG, 1, 0, 0, 0);
+}
+
+TEST_F(AuthSrvTest, AXFRSuccess) {
+    EXPECT_FALSE(xfrout.isConnected());
+    createRequestPacket(opcode, Name("example.com"), RRClass::IN(),
+                        RRType::AXFR(), IPPROTO_TCP);
+    // On success, the AXFR query has been passed to a separate process,
+    // so we shouldn't have to respond.
+    EXPECT_EQ(false, server.processMessage(*io_message, parse_message,
+                                           response_renderer));
+    EXPECT_TRUE(xfrout.isConnected());
+}
+
+TEST_F(AuthSrvTest, AXFRConnectFail) {
+    EXPECT_FALSE(xfrout.isConnected()); // check prerequisite
+    xfrout.disableConnect();
+    createRequestPacket(opcode, Name("example.com"), RRClass::IN(),
+                        RRType::AXFR(), IPPROTO_TCP);
+    EXPECT_TRUE(server.processMessage(*io_message, parse_message,
+                                      response_renderer));
+    headerCheck(parse_message, default_qid, Rcode::SERVFAIL(),
+                opcode.getCode(), QR_FLAG, 1, 0, 0, 0);
+    EXPECT_FALSE(xfrout.isConnected());
+}
+
+TEST_F(AuthSrvTest, AXFRSendFail) {
+    // first send a valid query, making the connection with the xfr process
+    // open.
+    createRequestPacket(opcode, Name("example.com"), RRClass::IN(),
+                        RRType::AXFR(), IPPROTO_TCP);
+    server.processMessage(*io_message, parse_message, response_renderer);
+    EXPECT_TRUE(xfrout.isConnected());
+
+    xfrout.disableSend();
+    parse_message.clear(Message::PARSE);
+    response_renderer.clear();
+    createRequestPacket(opcode, Name("example.com"), RRClass::IN(),
+                        RRType::AXFR(), IPPROTO_TCP);
+    EXPECT_TRUE(server.processMessage(*io_message, parse_message,
+                                      response_renderer));
+    headerCheck(parse_message, default_qid, Rcode::SERVFAIL(),
+                opcode.getCode(), QR_FLAG, 1, 0, 0, 0);
+
+    // The connection should have been closed due to the send failure.
+    EXPECT_FALSE(xfrout.isConnected());
+}
+
+TEST_F(AuthSrvTest, AXFRDisconnectFail) {
+    // In our usage disconnect() shouldn't fail.  So we'll see the exception
+    // should it be thrown.
+    xfrout.disableSend();
+    xfrout.disableDisconnect();
+    createRequestPacket(opcode, Name("example.com"), RRClass::IN(),
+                        RRType::AXFR(), IPPROTO_TCP);
+    EXPECT_THROW(server.processMessage(*io_message, parse_message,
+                                       response_renderer),
+                 XfroutError);
+    EXPECT_TRUE(xfrout.isConnected());
+    // XXX: we need to re-enable disconnect.  otherwise an exception would be
+    // thrown via the destructor of the server.
+    xfrout.enableDisconnect();
+}
+
+TEST_F(AuthSrvTest, notify) {
+    createRequestMessage(Opcode::NOTIFY(), Name("example.com"), RRClass::IN(),
+                        RRType::SOA());
+    request_message.setHeaderFlag(MessageFlag::AA());
+    createRequestPacket(IPPROTO_UDP);
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
+
+    // An internal command message should have been created and sent to an
+    // external module.  Check them.
+    EXPECT_EQ("Xfrin", notify_session.msg_destination);
+    EXPECT_EQ("notify",
+              notify_session.sent_msg->get("command")->get(0)->stringValue());
+    ElementPtr notify_args = notify_session.sent_msg->get("command")->get(1);
+    EXPECT_EQ("example.com.", notify_args->get("zone_name")->stringValue());
+    EXPECT_EQ(DEFAULT_REMOTE_ADDRESS,
+              notify_args->get("master")->stringValue());
+    EXPECT_EQ("IN", notify_args->get("rrclass")->stringValue());
+
+    // On success, the server should return a response to the notify.
+    headerCheck(parse_message, default_qid, Rcode::NOERROR(),
+                Opcode::NOTIFY().getCode(), QR_FLAG | AA_FLAG, 1, 0, 0, 0);
+
+    // The question must be identical to that of the received notify
+    ConstQuestionPtr question = *parse_message.beginQuestion();
+    EXPECT_EQ(Name("example.com"), question->getName());
+    EXPECT_EQ(RRClass::IN(), question->getClass());
+    EXPECT_EQ(RRType::SOA(), question->getType());
+}
+
+TEST_F(AuthSrvTest, notifyForCHClass) {
+    // Same as the previous test, but for the CH RRClass.
+    createRequestMessage(Opcode::NOTIFY(), Name("example.com"), RRClass::CH(),
+                        RRType::SOA());
+    request_message.setHeaderFlag(MessageFlag::AA());
+    createRequestPacket(IPPROTO_UDP);
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
+
+    // Other conditions should be the same, so simply confirm the RR class is
+    // set correctly.
+    ElementPtr notify_args = notify_session.sent_msg->get("command")->get(1);
+    EXPECT_EQ("CH", notify_args->get("rrclass")->stringValue());
+}
+
+TEST_F(AuthSrvTest, notifyEmptyQuestion) {
+    request_message.clear(Message::RENDER);
+    request_message.setOpcode(Opcode::NOTIFY());
+    request_message.setHeaderFlag(MessageFlag::AA());
+    request_message.setQid(default_qid);
+    request_message.toWire(request_renderer);
+    createRequestPacket(IPPROTO_UDP);
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
+    headerCheck(parse_message, default_qid, Rcode::FORMERR(),
+                Opcode::NOTIFY().getCode(), QR_FLAG, 0, 0, 0, 0);
+}
+
+TEST_F(AuthSrvTest, notifyMultiQuestions) {
+    createRequestMessage(Opcode::NOTIFY(), Name("example.com"), RRClass::IN(),
+                        RRType::SOA());
+    // add one more SOA question
+    request_message.addQuestion(Question(Name("example.com"), RRClass::IN(),
+                                         RRType::SOA()));
+    request_message.setHeaderFlag(MessageFlag::AA());
+    createRequestPacket(IPPROTO_UDP);
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
+    headerCheck(parse_message, default_qid, Rcode::FORMERR(),
+                Opcode::NOTIFY().getCode(), QR_FLAG, 2, 0, 0, 0);
+}
+
+TEST_F(AuthSrvTest, notifyNonSOAQuestion) {
+    createRequestMessage(Opcode::NOTIFY(), Name("example.com"), RRClass::IN(),
+                        RRType::NS());
+    request_message.setHeaderFlag(MessageFlag::AA());
+    createRequestPacket(IPPROTO_UDP);
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
+    headerCheck(parse_message, default_qid, Rcode::FORMERR(),
+                Opcode::NOTIFY().getCode(), QR_FLAG, 1, 0, 0, 0);
+}
+
+TEST_F(AuthSrvTest, notifyWithoutAA) {
+    // implicitly leave the AA bit off.  our implementation will accept it.
+    createRequestPacket(Opcode::NOTIFY(), Name("example.com"), RRClass::IN(),
+                        RRType::SOA());
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
+    headerCheck(parse_message, default_qid, Rcode::NOERROR(),
+                Opcode::NOTIFY().getCode(), QR_FLAG | AA_FLAG, 1, 0, 0, 0);
+}
+
+TEST_F(AuthSrvTest, notifyWithErrorRcode) {
+    createRequestMessage(Opcode::NOTIFY(), Name("example.com"), RRClass::IN(),
+                        RRType::SOA());
+    request_message.setHeaderFlag(MessageFlag::AA());
+    request_message.setRcode(Rcode::SERVFAIL());
+    createRequestPacket(IPPROTO_UDP);
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
+    headerCheck(parse_message, default_qid, Rcode::NOERROR(),
+                Opcode::NOTIFY().getCode(), QR_FLAG | AA_FLAG, 1, 0, 0, 0);
+}
+
+TEST_F(AuthSrvTest, notifyWithoutSession) {
+    server.setXfrinSession(NULL);
+
+    createRequestMessage(Opcode::NOTIFY(), Name("example.com"), RRClass::IN(),
+                        RRType::SOA());
+    request_message.setHeaderFlag(MessageFlag::AA());
+    createRequestPacket(IPPROTO_UDP);
+
+    // we simply ignore the notify and let it be resent if an internal error
+    // happens.
+    EXPECT_FALSE(server.processMessage(*io_message, parse_message,
+                                       response_renderer));
+}
+
+TEST_F(AuthSrvTest, notifySendFail) {
+    notify_session.disableSend();
+
+    createRequestMessage(Opcode::NOTIFY(), Name("example.com"), RRClass::IN(),
+                        RRType::SOA());
+    request_message.setHeaderFlag(MessageFlag::AA());
+    createRequestPacket(IPPROTO_UDP);
+
+    EXPECT_FALSE(server.processMessage(*io_message, parse_message,
+                                       response_renderer));
+}
+
+TEST_F(AuthSrvTest, notifyReceiveFail) {
+    notify_session.disableReceive();
+
+    createRequestMessage(Opcode::NOTIFY(), Name("example.com"), RRClass::IN(),
+                        RRType::SOA());
+    request_message.setHeaderFlag(MessageFlag::AA());
+    createRequestPacket(IPPROTO_UDP);
+    EXPECT_FALSE(server.processMessage(*io_message, parse_message,
+                                       response_renderer));
+}
+
+TEST_F(AuthSrvTest, notifyWithBogusSessionMessage) {
+    notify_session.setMessage(Element::fromJSON("{\"foo\": 1}"));
+
+    createRequestMessage(Opcode::NOTIFY(), Name("example.com"), RRClass::IN(),
+                        RRType::SOA());
+    request_message.setHeaderFlag(MessageFlag::AA());
+    createRequestPacket(IPPROTO_UDP);
+    EXPECT_FALSE(server.processMessage(*io_message, parse_message,
+                                       response_renderer));
+}
+
+TEST_F(AuthSrvTest, notifyWithSessionMessageError) {
+    notify_session.setMessage(
+        Element::fromJSON("{\"result\": [1, \"FAIL\"]}"));
+
+    createRequestMessage(Opcode::NOTIFY(), Name("example.com"), RRClass::IN(),
+                        RRType::SOA());
+    request_message.setHeaderFlag(MessageFlag::AA());
+    createRequestPacket(IPPROTO_UDP);
+    EXPECT_FALSE(server.processMessage(*io_message, parse_message,
+                                       response_renderer));
+}
+
 void
 updateConfig(AuthSrv* server, const char* const dbfile,
              const bool expect_success)
@@ -253,8 +709,8 @@ TEST_F(AuthSrvTest, updateConfig) {
     // response should have the AA flag on, and have an RR in each answer
     // and authority section.
     createDataFromFile("examplequery_fromWire");
-    EXPECT_EQ(true, server.processMessage(*ibuffer, parse_message,
-                                          response_renderer, true));
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
     headerCheck(parse_message, default_qid, Rcode::NOERROR(), opcode.getCode(),
                 QR_FLAG | AA_FLAG, 1, 1, 1, 0);
 }
@@ -267,10 +723,10 @@ TEST_F(AuthSrvTest, datasourceFail) {
     // in a SERVFAIL response, and the answer and authority sections should
     // be empty.
     createDataFromFile("badExampleQuery_fromWire");
-    EXPECT_EQ(true, server.processMessage(*ibuffer, parse_message,
-                                          response_renderer, true));
-    headerCheck(parse_message, default_qid, Rcode::SERVFAIL(), opcode.getCode(),
-                QR_FLAG, 1, 0, 0, 0);
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
+    headerCheck(parse_message, default_qid, Rcode::SERVFAIL(),
+                opcode.getCode(), QR_FLAG, 1, 0, 0, 0);
 }
 
 TEST_F(AuthSrvTest, updateConfigFail) {
@@ -282,8 +738,8 @@ TEST_F(AuthSrvTest, updateConfigFail) {
 
     // The original data source should still exist.
     createDataFromFile("examplequery_fromWire");
-    EXPECT_EQ(true, server.processMessage(*ibuffer, parse_message,
-                                          response_renderer, true));
+    EXPECT_EQ(true, server.processMessage(*io_message, parse_message,
+                                          response_renderer));
     headerCheck(parse_message, default_qid, Rcode::NOERROR(), opcode.getCode(),
                 QR_FLAG | AA_FLAG, 1, 1, 1, 0);
 }

src/bin/xfrin/tests/xfrin_test.py

@@ -412,22 +412,31 @@ class TestXfrin(unittest.TestCase):
         return self.xfr._parse_cmd_params(self.args)
 
     def test_parse_cmd_params(self):
-        name, master_addrinfo, db_file = self._do_parse()
+        name, rrclass, master_addrinfo, db_file = self._do_parse()
         self.assertEqual(master_addrinfo[4][1], int(TEST_MASTER_PORT))
         self.assertEqual(name, TEST_ZONE_NAME)
+        self.assertEqual(rrclass, TEST_RRCLASS)
         self.assertEqual(master_addrinfo[4][0], TEST_MASTER_IPV4_ADDRESS)
         self.assertEqual(db_file, TEST_DB_FILE)
 
     def test_parse_cmd_params_default_port(self):
         del self.args['port']
-        master_addrinfo = self._do_parse()[1]
+        master_addrinfo = self._do_parse()[2]
         self.assertEqual(master_addrinfo[4][1], 53)
 
     def test_parse_cmd_params_ip6master(self):
         self.args['master'] = TEST_MASTER_IPV6_ADDRESS
-        master_addrinfo = self._do_parse()[1]
+        master_addrinfo = self._do_parse()[2]
         self.assertEqual(master_addrinfo[4][0], TEST_MASTER_IPV6_ADDRESS)
 
+    def test_parse_cmd_params_chclass(self):
+        self.args['rrclass'] = 'CH'
+        self.assertEqual(self._do_parse()[1], RRClass.CH())
+
+    def test_parse_cmd_params_bogusclass(self):
+        self.args['rrclass'] = 'XXX'
+        self.assertRaises(XfrinException, self._do_parse)
+
     def test_parse_cmd_params_nozone(self):
         # zone name is mandatory.
         del self.args['zone_name']
@@ -504,6 +513,13 @@ class TestXfrin(unittest.TestCase):
         self.assertEqual(self.xfr.command_handler("refresh",
                                                   self.args)['result'][0], 0)
 
+    def test_command_handler_notify(self):
+        # at this level, refresh is no different than retransfer.
+        self.args['master'] = TEST_MASTER_IPV6_ADDRESS
+        # ...but right now we disable the feature due to security concerns.
+        self.assertEqual(self.xfr.command_handler("notify",
+                                                  self.args)['result'][0], 1)
+
     def test_command_handler_unknown(self):
         self.assertEqual(self.xfr.command_handler("xxx", None)['result'][0], 1)
 

src/bin/xfrin/xfrin.py.in

@@ -404,14 +404,37 @@ a separate method for the convenience of unit tests.
             if command == 'shutdown':
                 self._shutdown_event.set()
             elif command == 'retransfer' or command == 'refresh':
-                # The default RR class is IN.  We should fix this so that
-                # the class is passed in the command arg (where we specify
-                # the default)
-                rrclass = RRClass.IN()
-                zone_name, master_addr, db_file = self._parse_cmd_params(args)
-                ret = self.xfrin_start(zone_name, rrclass, db_file, master_addr,
+                (zone_name, rrclass,
+                 master_addr, db_file) = self._parse_cmd_params(args)
+                ret = self.xfrin_start(zone_name, rrclass, db_file,
+                                       master_addr,
                                    False if command == 'retransfer' else True)
                 answer = create_answer(ret[0], ret[1])
+            elif command == 'notify':
+                # This is the temporary implementation for notify.
+                # actually the notfiy command should be sent to the
+                # Zone Manager module.  Being temporary, we separate this case
+                # from refresh/retransfer while we could (and should otherwise)
+                # share the code.
+                (zone_name, rrclass,
+                 master_addr, db_file) = self._parse_cmd_params(args)
+
+                # XXX: master_addr is the sender of the notify message.
+                # It's very dangerous to naively trust it as the source of
+                # subsequent zone transfer; any remote node can easily exploit
+                # it to mount zone poisoning or DoS attacks.  We should
+                # locally identify the appropriate set of master servers.
+                # For now, we disable the code below.
+                master_is_valid = False
+
+                if master_is_valid:
+                    ret = self.xfrin_start(zone_name, rrclass, db_file,
+                                           master_addr, True)
+                else:
+                    errmsg = 'Failed to validate the master address ('
+                    errmsg += args['master'] + '), ignoring notify'
+                    ret = [1, errmsg]
+                answer = create_answer(ret[0], ret[1])
             else:
                 answer = create_answer(1, 'unknown command: ' + command)
 
@@ -425,6 +448,18 @@ a separate method for the convenience of unit tests.
         if not zone_name:
             raise XfrinException('zone name should be provided')
 
+        rrclass = args.get('rrclass')
+        if not rrclass:
+            # The default RR class is IN.  We should fix this so that
+            # the class is always passed in the command arg (where we specify
+            # the default)
+            rrclass = RRClass.IN()
+        else:
+            try:
+                rrclass = RRClass(rrclass)
+            except InvalidRRClass as e:
+                raise XfrinException('invalid RRClass: ' + rrclass)
+
         master = args.get('master')
         if not master:
             raise XfrinException('master address should be provided')
@@ -450,7 +485,7 @@ a separate method for the convenience of unit tests.
                 db_file = os.environ["B10_FROM_BUILD"] + os.sep + "bind10_zones.sqlite3"
             self._cc.remove_remote_config(AUTH_SPECFILE_LOCATION)
 
-        return (zone_name, master_addrinfo, db_file)
+        return (zone_name, rrclass, master_addrinfo, db_file)
 
     def startup(self):
         while not self._shutdown_event.is_set():

src/lib/cc/session.h

@@ -40,13 +40,34 @@ namespace isc {
                 isc::Exception(file, line, what) {}
         };
 
+        /// \brief The AbstractSession class is an abstract base class that
+        /// defines the interfaces of Session.
+        /// The intended primary usage of abstraction is to allow tests for the
+        /// user class of Session without requiring actual communication
+        /// channels.
+        /// For simplicity we only define the methods that are necessary for
+        /// existing test cases that use this base class.  Eventually we'll
+        /// probably have to extend them.
         class AbstractSession {
+            ///
+            /// \name Constructors, Assignment Operator and Destructor.
+            ///
+            /// Note: The copy constructor and the assignment operator are
+            /// intentionally defined as private to make it explicit that
+            /// this is a pure base class.
+            //@{
         private:
             AbstractSession(const AbstractSession& source);
             AbstractSession& operator=(const AbstractSession& source);
         protected:
+            /// \brief The default constructor.
+            ///
+            /// This is intentionally defined as \c protected as this base
+            /// class should never be instantiated (except as part of a
+            /// derived class).
             AbstractSession() {}
         public:
+            /// \brief The destructor.
             virtual ~AbstractSession() {}
             //@}
             virtual void establish(const char* socket_file) = 0;
@@ -79,7 +100,7 @@ namespace isc {
 
         public:
             Session(asio::io_service& ioservice);
-            ~Session();
+            virtual ~Session();
 
             virtual void startRead(boost::function<void()> read_callback);
 

src/lib/xfr/xfrout_client.cc

@@ -54,16 +54,25 @@ XfroutClient::~XfroutClient()
 
 void
 XfroutClient::connect() {
-    impl_->socket_.connect(stream_protocol::endpoint(impl_->file_path_));
+    asio::error_code err;
+    impl_->socket_.connect(stream_protocol::endpoint(impl_->file_path_), err);
+    if (err) {
+        isc_throw(XfroutError, "socket connect failed: " << err.message());
+    }
 }
 
 void
 XfroutClient::disconnect() {
-    impl_->socket_.close();
+    asio::error_code err;
+    impl_->socket_.close(err);
+    if (err) {
+        isc_throw(XfroutError, "close socket failed: " << err.message());
+    }
 }
 
 int 
-XfroutClient::sendXfroutRequestInfo(const int tcp_sock, uint8_t* msg_data,
+XfroutClient::sendXfroutRequestInfo(const int tcp_sock,
+                                    const void* const msg_data,
                                     const uint16_t msg_len)
 {
     if (-1 == send_fd(impl_->socket_.native(), tcp_sock)) {
@@ -71,7 +80,8 @@ XfroutClient::sendXfroutRequestInfo(const int tcp_sock, uint8_t* msg_data,
                   "Fail to send the socket file descriptor to xfrout module");
     }
 
-    // XXX: this shouldn't be blocking send, even though it's unlikely to block.
+    // XXX: this shouldn't be blocking send, even though it's unlikely to
+    // block.
     const uint8_t lenbuf[2] = { msg_len >> 8, msg_len & 0xff };
     if (send(impl_->socket_.native(), lenbuf, sizeof(lenbuf), 0) !=
         sizeof(lenbuf)) {

src/lib/xfr/xfrout_client.h

@@ -34,7 +34,39 @@ public:
         isc::Exception(file, line, what) {}
 };
 
-class XfroutClient {
+/// \brief The AbstractXfroutClient class is an abstract base class that
+/// defines the interfaces of XfroutClient.
+///
+/// The intended primary usage of abstraction is to allow tests for the
+/// user class of XfroutClient without requiring actual communication.
+class AbstractXfroutClient {
+    ///
+    /// \name Constructors, Assignment Operator and Destructor.
+    ///
+    /// Note: The copy constructor and the assignment operator are
+    /// intentionally defined as private to make it explicit that this is a
+    /// pure base class.
+    //@{
+private:
+    AbstractXfroutClient(const AbstractXfroutClient& source);
+    AbstractXfroutClient& operator=(const AbstractXfroutClient& source);
+protected:
+    /// \brief The default constructor.
+    ///
+    /// This is intentionally defined as \c protected as this base class should
+    /// never be instantiated (except as part of a derived class).
+    AbstractXfroutClient() {}
+public:
+    /// \brief The destructor.
+    virtual ~AbstractXfroutClient() {}
+    //@}
+    virtual void connect() = 0;
+    virtual void disconnect() = 0;
+    virtual int sendXfroutRequestInfo(int tcp_sock, const void* msg_data,
+                                      uint16_t msg_len) = 0;
+};
+
+class XfroutClient : public AbstractXfroutClient {
 public:
     XfroutClient(const std::string& file);
     ~XfroutClient();
@@ -43,10 +75,10 @@ private:
     XfroutClient(const XfroutClient& source);
     XfroutClient& operator=(const XfroutClient& source);
 public:
-    void connect();
-    void disconnect();
-    int sendXfroutRequestInfo(int tcp_sock, uint8_t* msg_data,
-                              uint16_t msg_len);
+    virtual void connect();
+    virtual void disconnect();
+    virtual int sendXfroutRequestInfo(int tcp_sock, const void* msg_data,
+                                      uint16_t msg_len);
 private:
     XfroutClientImpl* impl_;
 };