[master] Merge branch 'trac5200'

src/bin/agent/tests/ca_command_mgr_unittests.cc

@@ -171,15 +171,12 @@ public:
     ///
     /// @param response Stub response to be sent from the server socket to the
     /// client.
-    /// @param stop_after_count Number of received messages received over the
-    /// server socket after which the IO service should be stopped.
-    void bindServerSocket(const std::string& response,
-                          const unsigned int stop_after_count = 1) {
+    void bindServerSocket(const std::string& response) {
         server_socket_.reset(new test::TestServerUnixSocket(*getIOService(),
                                                             unixSocketFilePath(),
                                                             TEST_TIMEOUT,
                                                             response));
-        server_socket_->bindServerSocket(stop_after_count);
+        server_socket_->bindServerSocket();
     }
 
     /// @brief Creates command with no arguments.
@@ -214,27 +211,28 @@ public:
     /// @param expected_result0 Expected first result in response from the server.
     /// @param expected_result1 Expected second result in response from the server.
     /// @param expected_result2 Expected third result in response from the server.
-    /// @param stop_after_count Number of received messages received over the
     /// server socket after which the IO service should be stopped.
+    /// @param expected_responses Number of responses after which the test finishes.
     /// @param server_response Stub response to be sent by the server.
     void testForward(const CtrlAgentCfgContext::ServerType& server_type,
                      const std::string& service,
                      const int expected_result0,
                      const int expected_result1 = -1,
                      const int expected_result2 = -1,
-                     const unsigned stop_after_count = 1,
+                     const size_t expected_responses = 1,
                      const std::string& server_response = "{ \"result\": 0 }") {
         // Configure client side socket.
         configureControlSocket(server_type);
         // Create server side socket.
-        bindServerSocket(server_response, stop_after_count);
+        bindServerSocket(server_response);
 
         // The client side communication is synchronous. To be able to respond
         // to this we need to run the server side socket at the same time.
         // Running IO service in a thread guarantees that the server responds
         // as soon as it receives the control command.
-        isc::util::thread::Thread(boost::bind(&IOService::run,
-                                              getIOService().get()));
+        isc::util::thread::Thread(boost::bind(&CtrlAgentCommandMgrTest::runIO,
+                                              getIOService(), server_socket_,
+                                              expected_responses));
 
         ConstElementPtr command = createCommand("foo", service);
         ConstElementPtr answer = mgr_.handleCommand("foo", ConstElementPtr(),
@@ -243,6 +241,24 @@ public:
         checkAnswer(answer, expected_result0, expected_result1, expected_result2);
     }
 
+    /// @brief Runs IO service until number of sent responses is lower than
+    /// expected.
+    ///
+    /// @param server_socket Pointer to the server socket.
+    /// @param expected_responses Number of expected responses.
+    static void runIO(IOServicePtr& io_service,
+                      const test::TestServerUnixSocketPtr& server_socket,
+                      const size_t expected_responses) {
+        while (server_socket->getResponseNum() < expected_responses) {
+            io_service->run_one();
+        }
+    }
+
+
+    CtrlAgentCommandMgrTest* getTestSelf() {
+        return (this);
+    }
+
     /// @brief a convenience reference to control agent command manager
     CtrlAgentCommandMgr& mgr_;
 
@@ -270,19 +286,19 @@ TEST_F(CtrlAgentCommandMgrTest, listCommands) {
 };
 
 /// Check that control command is successfully forwarded to the DHCPv4 server.
-TEST_F(CtrlAgentCommandMgrTest, DISABLED_forwardToDHCPv4Server) {
+TEST_F(CtrlAgentCommandMgrTest, forwardToDHCPv4Server) {
     testForward(CtrlAgentCfgContext::TYPE_DHCP4, "dhcp4",
                 isc::config::CONTROL_RESULT_SUCCESS);
 }
 
 /// Check that control command is successfully forwarded to the DHCPv6 server.
-TEST_F(CtrlAgentCommandMgrTest, DISABLED_forwardToDHCPv6Server) {
+TEST_F(CtrlAgentCommandMgrTest, forwardToDHCPv6Server) {
     testForward(CtrlAgentCfgContext::TYPE_DHCP6, "dhcp6",
                 isc::config::CONTROL_RESULT_SUCCESS);
 }
 
 /// Check that the same command is forwarded to multiple servers.
-TEST_F(CtrlAgentCommandMgrTest, DISABLED_forwardToBothDHCPServers) {
+TEST_F(CtrlAgentCommandMgrTest, forwardToBothDHCPServers) {
     configureControlSocket(CtrlAgentCfgContext::TYPE_DHCP6);
 
     testForward(CtrlAgentCfgContext::TYPE_DHCP4, "dhcp4,dhcp6",
@@ -347,8 +363,8 @@ TEST_F(CtrlAgentCommandMgrTest, forwardListCommands) {
     // to this we need to run the server side socket at the same time.
     // Running IO service in a thread guarantees that the server responds
     // as soon as it receives the control command.
-    isc::util::thread::Thread(boost::bind(&IOService::run,
-                                          getIOService().get()));
+    isc::util::thread::Thread(boost::bind(&CtrlAgentCommandMgrTest::runIO,
+                                          getIOService(), server_socket_, 1));
 
     ConstElementPtr command = createCommand("list-commands", "dhcp4");
     ConstElementPtr answer = mgr_.handleCommand("list-commands", ConstElementPtr(),

src/lib/asiolink/tests/unix_domain_socket_unittest.cc

@@ -95,7 +95,9 @@ TEST_F(UnixDomainSocketTest, sendReceive) {
     ASSERT_EQ(outbound_data.size(), sent_size);
 
     // Run IO service to generate server's response.
-    io_service_.run();
+    while (test_socket_.getResponseNum() < 1) {
+        io_service_.run_one();
+    }
 
     // Receive response from the socket.
     std::array<char, 1024> read_buf;

src/lib/asiolink/testutils/test_server_unix_socket.cc

@@ -7,11 +7,191 @@
 #include <asiolink/asio_wrapper.h>
 #include <asiolink/testutils/test_server_unix_socket.h>
 #include <boost/bind.hpp>
+#include <boost/shared_ptr.hpp>
+#include <functional>
+#include <set>
+
+using namespace boost::asio::local;
 
 namespace isc {
 namespace asiolink {
 namespace test {
 
+/// @brief ASIO unix domain socket.
+typedef stream_protocol::socket UnixSocket;
+
+/// @brief Pointer to the ASIO unix domain socket.
+typedef boost::shared_ptr<UnixSocket> UnixSocketPtr;
+
+/// @brief Callback function invoked when response is sent from the server.
+typedef std::function<void()> SentResponseCallback;
+
+/// @brief Connection to the server over unix domain socket.
+///
+/// It reads the data over the socket, sends responses and closes a socket.
+class Connection {
+public:
+
+    /// @brief Constructor.
+    ///
+    /// It starts asynchronous read operation.
+    ///
+    /// @param unix_socket Pointer to the unix domain socket into which
+    /// connection has been accepted.
+    /// @param custom_response Custom response that the server should send.
+    /// @param sent_response_callback Callback function to be invoked when
+    /// server sends a response.
+    Connection(const UnixSocketPtr& unix_socket,
+               const std::string custom_response,
+               const SentResponseCallback& sent_response_callback)
+        : socket_(unix_socket), custom_response_(custom_response),
+          sent_response_callback_(sent_response_callback) {
+       socket_->async_read_some(boost::asio::buffer(&raw_buf_[0], raw_buf_.size()),
+           boost::bind(&Connection::readHandler, this, _1, _2));
+    }
+
+    /// @brief Handler invoked when data have been received over the socket.
+    ///
+    /// This is the handler invoked when the data have been received over the
+    /// socket. If custom response has been specified, this response is sent
+    /// back to the client. Otherwise, the handler echoes back the request
+    /// and prepends the word "received ". Finally, it calls a custom
+    /// callback function (specified in the constructor) to notify that the
+    /// response has been sent over the socket.
+    ///
+    /// @param bytes_transferred Number of bytes received.
+    void
+    readHandler(const boost::system::error_code&, size_t bytes_transferred) {
+        if (!custom_response_.empty()) {
+            boost::asio::write(*socket_,
+               boost::asio::buffer(custom_response_.c_str(), custom_response_.size()));
+
+        } else {
+            std::string received(&raw_buf_[0], bytes_transferred);
+            std::string response("received " + received);
+            boost::asio::write(*socket_,
+                boost::asio::buffer(response.c_str(), response.size()));
+        }
+
+        // Invoke callback function to notify that the response has been sent.
+        sent_response_callback_();
+    }
+
+    /// @brief Closes the socket.
+    void stop() {
+        socket_->close();
+    }
+
+private:
+
+    /// @brief Pointer to the unix domain socket.
+    UnixSocketPtr socket_;
+
+    /// @brief Custom response to be sent to the client.
+    std::string custom_response_;
+
+    /// @brief Receive buffer.
+    std::array<char, 1024> raw_buf_;
+
+    /// @brief Pointer to the callback function to be invoked when response
+    /// has been sent.
+    SentResponseCallback sent_response_callback_;
+
+};
+
+/// @brief Pointer to a Connection object.
+typedef boost::shared_ptr<Connection> ConnectionPtr;
+
+/// @brief Connection pool.
+///
+/// Holds all connections established with the server and gracefully
+/// terminates these connections.
+class ConnectionPool {
+public:
+
+    /// @brief Constructor.
+    ///
+    /// @param io_service Reference to the IO service.
+    ConnectionPool(IOService& io_service)
+        : io_service_(io_service), connections_(), next_socket_(),
+          response_num_(0) {
+    }
+
+    /// @brief Destructor.
+    ~ConnectionPool() {
+        stopAll();
+    }
+
+    /// @brief Creates new unix domain socket and returns it.
+    ///
+    /// This convenience method creates a socket which can be used to accept
+    /// new connections. If such socket already exists, it is returned.
+    ///
+    /// @return Pointer to the socket.
+    UnixSocketPtr getSocket() {
+        if (!next_socket_) {
+            next_socket_.reset(new UnixSocket(io_service_.get_io_service()));
+        }
+        return (next_socket_);
+    }
+
+    /// @brief Starts new connection.
+    ///
+    /// The socket returned by the @ref ConnectionPool::getSocket is used to
+    /// create new connection. Then, the @ref next_socket_ is reset, to force
+    /// the @ref ConnectionPool::getSocket to generate a new socket for a
+    /// next connection.
+    ///
+    /// @param custom_response Custom response to be sent to the client.
+    void start(const std::string& custom_response) {
+        ConnectionPtr conn(new Connection(next_socket_, custom_response, [this] {
+            ++response_num_;
+        }));
+
+        connections_.insert(conn);
+        next_socket_.reset();
+    }
+
+    /// @brief Stops the given connection.
+    ///
+    /// @param conn Pointer to the connection to be stopped.
+    void stop(const ConnectionPtr& conn) {
+        conn->stop();
+        connections_.erase(conn);
+    }
+
+    /// @brief Stops all connections.
+    void stopAll() {
+        for (auto conn = connections_.begin(); conn != connections_.end();
+             ++conn) {
+            (*conn)->stop();
+        }
+        connections_.clear();
+    }
+
+    /// @brief Returns number of responses sent so far.
+    size_t getResponseNum() const {
+        return (response_num_);
+    }
+
+private:
+
+    /// @brief Reference to the IO service.
+    IOService& io_service_;
+
+    /// @brief Container holding established connections.
+    std::set<ConnectionPtr> connections_;
+
+    /// @brief Holds pointer to the generated socket.
+    ///
+    /// This socket will be used by the next connection.
+    UnixSocketPtr next_socket_;
+
+    /// @brief Holds the number of sent responses.
+    size_t response_num_;
+};
+
+
 TestServerUnixSocket::TestServerUnixSocket(IOService& io_service,
                                            const std::string& socket_file_path,
                                            const long test_timeout,
@@ -19,67 +199,35 @@ TestServerUnixSocket::TestServerUnixSocket(IOService& io_service,
     : io_service_(io_service),
       server_endpoint_(socket_file_path),
       server_acceptor_(io_service_.get_io_service()),
-      server_socket_(io_service_.get_io_service()),
       test_timer_(io_service_),
       custom_response_(custom_response),
-      stop_after_count_(1),
-      read_count_(0) {
+      connection_pool_(new ConnectionPool(io_service)) {
     test_timer_.setup(boost::bind(&TestServerUnixSocket::timeoutHandler, this),
                       test_timeout, IntervalTimer::ONE_SHOT);
 }
 
+TestServerUnixSocket::~TestServerUnixSocket() {
+    connection_pool_->stopAll();
+}
+
 void
-TestServerUnixSocket::bindServerSocket(const unsigned int stop_after_count) {
+TestServerUnixSocket::bindServerSocket() {
     server_acceptor_.open();
     server_acceptor_.bind(server_endpoint_);
     server_acceptor_.listen();
     accept();
-
-    stop_after_count_ = stop_after_count;
 }
 
 void
 TestServerUnixSocket::acceptHandler(const boost::system::error_code&) {
-    server_socket_.async_read_some(boost::asio::buffer(&raw_buf_[0],
-                                                       raw_buf_.size()),
-                                   boost::bind(&TestServerUnixSocket::
-                                               readHandler, this, _1, _2));
+    connection_pool_->start(custom_response_);
+    accept();
 }
 
 void
 TestServerUnixSocket::accept() {
-    server_acceptor_.async_accept(server_socket_,
-                                  boost::bind(&TestServerUnixSocket::
-                                              acceptHandler, this, _1));
-}
-
-
-void
-TestServerUnixSocket::readHandler(const boost::system::error_code&,
-                                  size_t bytes_transferred) {
-    if (!custom_response_.empty()) {
-        boost::asio::write(server_socket_, boost::asio::buffer(custom_response_.c_str(),
-                                                               custom_response_.size()));
-
-    } else {
-        std::string received(&raw_buf_[0], bytes_transferred);
-        std::string response("received " + received);
-        boost::asio::write(server_socket_, boost::asio::buffer(response.c_str(),
-                                                               response.size()));
-    }
-
-    // Close the connection as we might be expecting another connection over the
-    // same socket.
-    server_socket_.close();
-
-    // Stop IO service if we have reached the maximum number of read messages.
-    if (++read_count_ >= stop_after_count_) {
-        io_service_.stop();
-
-    } else {
-        // Previous connection is done, so let's accept another connection.
-        accept();
-    }
+    server_acceptor_.async_accept(*(connection_pool_->getSocket()),
+        boost::bind(&TestServerUnixSocket::acceptHandler, this, _1));
 }
 
 void
@@ -88,6 +236,12 @@ TestServerUnixSocket::timeoutHandler() {
     io_service_.stop();
 }
 
+size_t
+TestServerUnixSocket::getResponseNum() const {
+    return (connection_pool_->getResponseNum());
+}
+
+
 } // end of namespace isc::asiolink::test
 } // end of namespace isc::asiolink
 } // end of namespace isc

src/lib/asiolink/testutils/test_server_unix_socket.h

@@ -12,14 +12,32 @@
 #include <asiolink/io_service.h>
 #include <boost/shared_ptr.hpp>
 #include <gtest/gtest.h>
-#include <array>
+#include <list>
 #include <string>
 
 namespace isc {
 namespace asiolink {
 namespace test {
 
+class ConnectionPool;
+
 /// @brief Provides unix domain socket functionality for unit tests.
+///
+/// This class represents a server side socket. It can be used to
+/// test client's transmission over the unix domain socket. By default,
+/// the server side socket echoes the client's message so the client's
+/// message (prefixed with the word "received").
+///
+/// It is also possible to specify a custom response from the server
+/// instead of eachoing back the request.
+///
+/// It is possible to make multiple connections to the server side
+/// socket simultaneously.
+///
+/// The test should perform IOService::run_one until it finds that
+/// the number of responses sent by the server is greater than
+/// expected. The number of responses sent so far can be retrieved
+/// using @ref TestServerUnixSocket::getResponseNum.
 class TestServerUnixSocket {
 public:
 
@@ -32,31 +50,29 @@ public:
     TestServerUnixSocket(IOService& io_service,
                          const std::string& socket_file_path,
                          const long test_timeout,
-                         const std::string& custom_respons_ = "");
+                         const std::string& custom_response = "");
 
-    /// @brief Creates and binds server socket.
+    /// @brief Destructor.
     ///
-    /// @param stop_after_count Number of received messages after which the
-    /// IO service should be stopped.
-    void bindServerSocket(const unsigned int stop_after_count = 1);
+    /// Closes active connections.
+    ~TestServerUnixSocket();
+
+    /// @brief Creates and binds server socket.
+    void bindServerSocket();
 
     /// @brief Server acceptor handler.
     ///
     /// @param ec Error code.
     void acceptHandler(const boost::system::error_code& ec);
 
-    /// @brief Server read handler.
-    ///
-    /// @param ec Error code.
-    /// @param bytes_transferred Number of bytes read.
-    void readHandler(const boost::system::error_code& ec,
-                     size_t bytes_transferred);
-
     /// @brief Callback function invoke upon test timeout.
     ///
     /// It stops the IO service and reports test timeout.
     void timeoutHandler();
 
+    /// @brief Return number of responses sent so far to the clients.
+    size_t getResponseNum() const;
+
 private:
 
     /// @brief Asynchronously accept new connections.
@@ -70,23 +86,14 @@ private:
     /// @brief Server acceptor.
     boost::asio::local::stream_protocol::acceptor server_acceptor_;
 
-    /// @brief Server side unix domain socket.
-    boost::asio::local::stream_protocol::socket server_socket_;
-
-    /// @brief Receive buffer.
-    std::array<char, 1024> raw_buf_;
-
     /// @brief Asynchronous timer service to detect timeouts.
     IntervalTimer test_timer_;
 
     /// @brief Holds custom response to be sent to the client.
     std::string custom_response_;
 
-    /// @brief Number of messages received after which IO service gets stopped.
-    unsigned int stop_after_count_;
-
-    /// @brief Number of messages received so far.
-    unsigned int read_count_;
+    /// @brief Pool of connections.
+    boost::shared_ptr<ConnectionPool> connection_pool_;
 };
 
 /// @brief Pointer to the @ref TestServerUnixSocket.