[trac554] Now have UDPSocket and its unit test working

src/lib/asiolink/Makefile.am

@@ -26,8 +26,9 @@ libasiolink_la_SOURCES += io_socket.cc io_socket.h
 libasiolink_la_SOURCES += io_message.h
 libasiolink_la_SOURCES += io_address.cc io_address.h
 libasiolink_la_SOURCES += io_endpoint.cc io_endpoint.h
-libasiolink_la_SOURCES += udp_endpoint.h udp_socket.h
+libasiolink_la_SOURCES += udp_endpoint.h
 libasiolink_la_SOURCES += udp_server.h udp_server.cc
+libasiolink_la_SOURCES += udp_socket.h udp_socket.cc
 libasiolink_la_SOURCES += udp_query.h udp_query.cc
 libasiolink_la_SOURCES += tcp_endpoint.h tcp_socket.h
 libasiolink_la_SOURCES += tcp_server.h tcp_server.cc

src/lib/asiolink/asiolink.h

@@ -34,6 +34,9 @@
 #include <asiolink/io_socket.h>
 #include <asiolink/io_error.h>
 
+#include <asiolink/udp_endpoint.h>
+#include <asiolink/udp_socket.h>
+
 /// \namespace asiolink
 /// \brief A wrapper interface for the ASIO library.
 ///

src/lib/asiolink/io_completion_cb.h

@@ -15,9 +15,11 @@
 #ifndef __IO_COMPLETION_CB_H
 #define __IO_COMPLETION_CB_H
 
+#include <asio/error.hpp>
 #include <asio/error_code.hpp>
 #include <coroutine.h>
 
+namespace asiolink {
 
 /// \brief Asynchronous I/O Completion Callback
 ///
@@ -83,5 +85,6 @@ private:
     IOCompletionCallback*   self_;      ///< Pointer to real object
 };
 
+} // namespace asiolink
 
 #endif // __IO_COMPLETION_CB_H

src/lib/asiolink/io_fetch.cc

@@ -172,6 +172,9 @@ IOFetch::operator()(error_code ec, size_t length) {
         /// be unnecessary.)
         data_->buffer->writeData(data_->data.get(), length);
 
+        // Finished with this socket, so close it.
+        data_->socket->close();
+
         /// We are done
         stop(SUCCESS);
     }

src/lib/asiolink/tests/Makefile.am

@@ -17,13 +17,15 @@ if HAVE_GTEST
 TESTS += run_unittests
 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 += udp_query_unittest.cc
 run_unittests_SOURCES += io_address_unittest.cc
 run_unittests_SOURCES += io_endpoint_unittest.cc
 run_unittests_SOURCES += io_socket_unittest.cc
 run_unittests_SOURCES += io_service_unittest.cc
 run_unittests_SOURCES += interval_timer_unittest.cc
 run_unittests_SOURCES += recursive_query_unittest.cc
+run_unittests_SOURCES += udp_endpoint_unittest.cc
+run_unittests_SOURCES += udp_query_unittest.cc
+run_unittests_SOURCES += udp_socket_unittest.cc
 run_unittests_SOURCES += run_unittests.cc
 run_unittests_CPPFLAGS = $(AM_CPPFLAGS) $(GTEST_INCLUDES)
 run_unittests_LDFLAGS = $(AM_LDFLAGS) $(GTEST_LDFLAGS) $(LOG4CXX_LDFLAGS)

src/lib/asiolink/tests/io_address_unittest.cc

@@ -56,3 +56,8 @@ TEST(IOAddressTest, Equality) {
     EXPECT_TRUE(IOAddress("2001:db8::1234") != IOAddress("192.0.2.3"));
     EXPECT_FALSE(IOAddress("2001:db8::1234") == IOAddress("192.0.2.3"));
 }
+
+TEST(IOAddressTest, Family) {
+    EXPECT_EQ(AF_INET, IOAddress("192.0.2.1").getFamily());
+    EXPECT_EQ(AF_INET6, IOAddress("2001:0DB8:0:0::0012").getFamily());
+}
\ No newline at end of file

src/lib/asiolink/tests/recursive_query_unittest.cc

@@ -41,8 +41,13 @@
 // if we include asio.hpp unless we specify a special compiler option.
 // If we need to test something at the level of underlying ASIO and need
 // their definition, that test should go to asiolink/internal/tests.
-#include <asiolink/asiolink.h>
+#include <asiolink/recursive_query.h>
 #include <asiolink/io_socket.h>
+#include <asiolink/io_service.h>
+#include <asiolink/io_message.h>
+#include <asiolink/io_error.h>
+#include <asiolink/dns_lookup.h>
+#include <asiolink/simple_callback.h>
 
 using isc::UnitTestUtil;
 using namespace std;

src/lib/asiolink/tests/udp_endpoint_unittest.cc

+// Copyright (C) 2011  Internet Systems Consortium, Inc. ("ISC")
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
+// OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+#include <config.h>
+
+#include <string>
+
+#include <gtest/gtest.h>
+
+#include <asio.hpp>
+#include <asiolink/io_address.h>
+#include <asiolink/udp_endpoint.h>
+
+using namespace asiolink;
+using namespace std;
+
+// This test checks that the endpoint can manage its own internal
+// asio::ip::udp::endpoint object.
+
+TEST(UDPEndpointTest, v4Address) {
+    const string test_address("192.0.2.1");
+    const unsigned short test_port = 5301;
+
+    IOAddress address(test_address);
+    UDPEndpoint endpoint(address, test_port);
+
+    EXPECT_TRUE(address == endpoint.getAddress());
+    EXPECT_EQ(test_port, endpoint.getPort());
+    EXPECT_EQ(IPPROTO_UDP, endpoint.getProtocol());
+    EXPECT_EQ(AF_INET, endpoint.getFamily());
+}
+
+TEST(UDPEndpointTest, v6Address) {
+    const string test_address("2001:db8::1235");
+    const unsigned short test_port = 5302;
+
+    IOAddress address(test_address);
+    UDPEndpoint endpoint(address, test_port);
+
+    EXPECT_TRUE(address == endpoint.getAddress());
+    EXPECT_EQ(test_port, endpoint.getPort());
+    EXPECT_EQ(IPPROTO_UDP, endpoint.getProtocol());
+    EXPECT_EQ(AF_INET6, endpoint.getFamily());
+}

src/lib/asiolink/tests/udp_socket_unittest.cc

+// Copyright (C) 2011  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.
+
+// Copyright (C) 2011  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.
+
+
+/// \brief Test of UDPSocket
+///
+/// Tests the fuctionality of a UDPSocket by working through an open-send-
+/// receive-close sequence and checking that the asynchronous notifications
+/// work.
+
+#include <string>
+
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+
+#include <algorithm>
+#include <cstdlib>
+#include <cstddef>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include <boost/bind.hpp>
+#include <boost/shared_ptr.hpp>
+
+#include <asio.hpp>
+
+#include <asiolink/io_completion_cb.h>
+#include <asiolink/io_service.h>
+#include <asiolink/udp_endpoint.h>
+#include <asiolink/udp_socket.h>
+
+using namespace asio;
+using namespace asiolink;
+using asio::ip::udp;
+using namespace std;
+
+namespace {
+
+const char* SERVER_ADDRESS = "127.0.0.1";
+const unsigned short SERVER_PORT = 5301;
+
+// FIXME Shouldn't we send something that is real message?
+const char OUTBOUND_DATA[] = "Data sent from client to server";
+const char INBOUND_DATA[] = "Returned data from server to client";
+}
+
+///
+/// An instance of this object is passed to the asynchronous I/O functions
+/// and the operator() method is called when when an asynchronous I/O
+/// completes.  The arguments to the completion callback are stored for later
+/// retrieval.
+class UDPCallback : public IOCompletionCallback {
+public:
+
+    struct PrivateData {
+        PrivateData() :
+            error_code_(), length_(0), called_(false), name_("")
+        {}
+
+        asio::error_code    error_code_;    ///< Completion error code
+        size_t              length_;        ///< Number of bytes transferred
+        bool                called_;        ///< Set true when callback called
+        std::string         name_;          ///< Which of the objects this is
+    };
+
+    /// \brief Constructor
+    ///
+    /// Constructs the object.  It also creates the data member pointed to by
+    /// a shared pointer.  When used as a callback object, this is copied as it
+    /// is passed into the asynchronous function.  This means that there are two
+    /// objects and inspecting the one we passed in does not tell us anything.
+    ///
+    /// Therefore we use a boost::shared_ptr.  When the object is copied, the
+    /// shared pointer is copied, which leaves both objects pointing to the same
+    /// data.
+    ///
+    /// \param which Which of the two callback objects this is
+    UDPCallback(std::string which) : ptr_(new PrivateData())
+    {
+        setName(which);
+    }
+
+    /// \brief Destructor
+    ///
+    /// No code needed, destroying the shared pointer destroys the private data.
+    virtual ~UDPCallback()
+    {}
+
+    /// \brief Callback Function
+    ///
+    /// Called when an asynchronous I/O completes, this stores the
+    /// completion error code and the number of bytes transferred.
+    ///
+    /// \param ec I/O completion error code passed to callback function.
+    /// \param length Number of bytes transferred
+    virtual void operator()(asio::error_code ec, size_t length = 0) {
+        ptr_->error_code_ = ec;
+        setLength(length);
+        setCalled(true);
+    }
+
+    /// \brief Get I/O completion error code
+    int getCode() {
+        return (ptr_->error_code_.value());
+    }
+
+    /// \brief Set I/O completion code
+    ///
+    /// \param code New value of completion code
+    void setCode(int code) {
+        ptr_->error_code_ = asio::error_code(code, asio::error_code().category());
+    }
+
+    /// \brief Get number of bytes transferred in I/O
+    size_t getLength() {
+        return (ptr_->length_);
+    }
+
+    /// \brief Set number of bytes transferred in I/O
+    ///
+    /// \param length New value of length parameter
+    void setLength(size_t length) {
+        ptr_->length_ = length;
+    }
+
+    /// \brief Get flag to say when callback was called
+    bool getCalled() {
+        return (ptr_->called_);
+    }
+
+    /// \brief Set flag to say when callback was called
+    ///
+    /// \param called New value of called parameter
+    void setCalled(bool called) {
+        ptr_->called_ = called;
+    }
+
+    /// \brief Return instance of callback name
+    std::string getName() {
+        return (ptr_->name_);
+    }
+
+    /// \brief Set callback name
+    ///
+    /// \param name New value of the callback name
+    void setName(const std::string& name) {
+        ptr_->name_ = name;
+    }
+
+private:
+    boost::shared_ptr<PrivateData>  ptr_;   ///< Pointer to private data
+};
+
+
+// Tests the operation of a UDPSocket by opening it, sending an asynchronous
+// message to a server, receiving an asynchronous message from the server and
+// closing.
+TEST(UDPSocket, SequenceTest) {
+
+    // Common objects.
+    IOAddress   server_address(SERVER_ADDRESS);         // Address of target server
+    UDPEndpoint endpoint(server_address, SERVER_PORT);  // Endpoint of target server
+    IOService   service;                    // Service object for async control
+
+    // The client - the UDPSocket being tested
+    UDPSocket   client(service);            // Socket under test
+    UDPCallback client_cb("Client");        // Async I/O callback function
+
+    // The server - with which the client communicates.  For convenience, we
+    // use the same io_service, and use the endpoint object created for
+    // the client to send to as the endpoint object in the constructor.
+    UDPCallback server_cb("Server");
+    udp::socket server(service.get_io_service(), endpoint.getASIOEndpoint());
+    server.set_option(socket_base::reuse_address(true));
+
+    // Assertion to ensure that the server buffer is large enough
+    char data[UDPSocket::MAX_SIZE];
+    ASSERT_GT(sizeof(data), sizeof(OUTBOUND_DATA));
+
+    // Open the client socket - the operation should be synchronous
+    EXPECT_FALSE(client.open(&endpoint, client_cb));
+
+    // Issue read on the server.  Completion callback should not have run.
+    server_cb.setCalled(false);
+    server_cb.setCode(42); // Answer to Life, the Universe and Everything!
+    UDPEndpoint server_remote_endpoint;
+    server.async_receive_from(buffer(data, sizeof(data)),
+        server_remote_endpoint.getASIOEndpoint(), server_cb);
+    EXPECT_FALSE(server_cb.getCalled());
+
+    // Write something to the server using the client - the callback should not
+    // be called until we call the io_service.run() method.
+    client_cb.setCalled(false);
+    client_cb.setCode(7);  // Arbitrary number
+    client.async_send(OUTBOUND_DATA, sizeof(OUTBOUND_DATA), &endpoint, client_cb);
+    EXPECT_FALSE(client_cb.getCalled());
+
+    // Execute the two callbacks.
+    service.run_one();
+    service.run_one();
+
+    EXPECT_TRUE(client_cb.getCalled());
+    EXPECT_EQ(0, client_cb.getCode());
+    EXPECT_EQ(sizeof(OUTBOUND_DATA), client_cb.getLength());
+
+    EXPECT_TRUE(server_cb.getCalled());
+    EXPECT_EQ(0, server_cb.getCode());
+    EXPECT_EQ(sizeof(OUTBOUND_DATA), server_cb.getLength());
+
+    EXPECT_TRUE(equal(&data[0], &data[server_cb.getLength() - 1], OUTBOUND_DATA));
+
+    // Now return data from the server to the client.  Issue the read on the
+    // client.
+    client_cb.setLength(12345);             // Arbitrary number
+    client_cb.setCalled(false);
+    client_cb.setCode(32);                  // Arbitrary number
+    UDPEndpoint client_remote_endpoint;     // To receive address of remote system
+    client.async_receive(data, sizeof(data), &client_remote_endpoint, client_cb);
+
+    // Issue the write on the server side to the source of the data it received.
+    server_cb.setLength(22345);             // Arbitrary number
+    server_cb.setCalled(false);
+    server_cb.setCode(232);                 // Arbitrary number
+    server.async_send_to(buffer(INBOUND_DATA, sizeof(INBOUND_DATA)),
+        server_remote_endpoint.getASIOEndpoint(), server_cb);
+
+
+    // Expect two callbacks to run
+    service.run_one();
+    service.run_one();
+
+    EXPECT_TRUE(client_cb.getCalled());
+    EXPECT_EQ(0, client_cb.getCode());
+    EXPECT_EQ(sizeof(INBOUND_DATA), client_cb.getLength());
+
+    EXPECT_TRUE(server_cb.getCalled());
+    EXPECT_EQ(0, server_cb.getCode());
+    EXPECT_EQ(sizeof(INBOUND_DATA), server_cb.getLength());
+
+    EXPECT_TRUE(equal(&data[0], &data[server_cb.getLength() - 1], INBOUND_DATA));
+
+    // Check that the address/port received by the client corresponds to the
+    // address and port the server is listening on.
+    EXPECT_TRUE(server_address == client_remote_endpoint.getAddress());
+    EXPECT_EQ(SERVER_PORT, client_remote_endpoint.getPort());
+
+    // Close client and server.
+    EXPECT_NO_THROW(client.close());
+    EXPECT_NO_THROW(server.close());
+}

src/lib/asiolink/udp_endpoint.h

@@ -33,6 +33,16 @@ public:
     /// \name Constructors and Destructor.
     ///
     //@{
+
+    /// \brief Default Constructor
+    ///
+    /// Creates an internal endpoint.  This is expected to be set by some
+    /// external call.
+    UDPEndpoint() :
+        asio_endpoint_placeholder_(new asio::ip::udp::endpoint()),
+        asio_endpoint_(*asio_endpoint_placeholder_)
+    {}
+
     /// \brief Constructor from a pair of address and port.
     ///
     /// \param address The IP address of the endpoint.
@@ -50,27 +60,27 @@ public:
     /// corresponding ASIO class, \c udp::endpoint.
     ///
     /// \param asio_endpoint The ASIO representation of the UDP endpoint.
-    UDPEndpoint(const asio::ip::udp::endpoint& asio_endpoint) :
+    UDPEndpoint(asio::ip::udp::endpoint& asio_endpoint) :
         asio_endpoint_placeholder_(NULL), asio_endpoint_(asio_endpoint)
     {}
 
     /// \brief The destructor.
-    ~UDPEndpoint() { delete asio_endpoint_placeholder_; }
+    virtual ~UDPEndpoint() { delete asio_endpoint_placeholder_; }
     //@}
 
-    inline IOAddress getAddress() const {
+    virtual IOAddress getAddress() const {
         return (asio_endpoint_.address());
     }
 
-    inline uint16_t getPort() const {
+    virtual uint16_t getPort() const {
         return (asio_endpoint_.port());
     }
 
-    inline short getProtocol() const {
+    virtual short getProtocol() const {
         return (asio_endpoint_.protocol().protocol());
     }
 
-    inline short getFamily() const {
+    virtual short getFamily() const {
         return (asio_endpoint_.protocol().family());
     }
 
@@ -79,10 +89,13 @@ public:
     inline const asio::ip::udp::endpoint& getASIOEndpoint() const {
         return (asio_endpoint_);
     }
+    inline asio::ip::udp::endpoint& getASIOEndpoint() {
+        return (asio_endpoint_);
+    }
 
 private:
-    const asio::ip::udp::endpoint* asio_endpoint_placeholder_;
-    const asio::ip::udp::endpoint& asio_endpoint_;
+    asio::ip::udp::endpoint* asio_endpoint_placeholder_;
+    asio::ip::udp::endpoint& asio_endpoint_;
 };
 
 }      // namespace asiolink

src/lib/asiolink/udp_socket.cc

+// 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.
+
+#include <config.h>
+#include <iostream>
+#include <unistd.h>             // for some IPC/network system calls
+#include <sys/socket.h>
+#include <netinet/in.h>
+
+#include <boost/bind.hpp>
+
+#include <asio.hpp>
+#include <asio/deadline_timer.hpp>
+
+#include <boost/shared_ptr.hpp>
+#include <boost/date_time/posix_time/posix_time_types.hpp>
+
+#include <dns/buffer.h>
+#include <dns/message.h>
+#include <dns/messagerenderer.h>
+#include <log/dummylog.h>
+#include <dns/opcode.h>
+#include <dns/rcode.h>
+
+#include <coroutine.h>
+#include <asiolink/asiolink.h>
+
+using namespace asio;
+using asio::ip::udp;
+
+using namespace std;
+using namespace isc::dns;
+
+namespace asiolink {
+
+// Constructor - create socket on the fly
+
+UDPSocket::UDPSocket(IOService& service) :
+    socket_ptr_(new asio::ip::udp::socket(service.get_io_service())),
+    socket_(*socket_ptr_)
+{
+}
+
+// Destructor
+
+UDPSocket::~UDPSocket()
+{
+    delete socket_ptr_;
+}
+
+// Open the socket.  Throws an error on failure
+// TODO: Make the open more resolient
+
+bool
+UDPSocket::open(const IOEndpoint* endpoint, IOCompletionCallback&) {
+    if (endpoint->getFamily() == AF_INET) {
+        socket_.open(asio::ip::udp::v4());
+    }
+    else {
+        socket_.open(asio::ip::udp::v6());
+    }
+
+    // Ensure it can send and receive 4K buffers.
+    socket_.set_option(asio::socket_base::send_buffer_size(MAX_SIZE));
+    socket_.set_option(asio::socket_base::receive_buffer_size(MAX_SIZE));
+;
+    // Allow reuse of an existing port/address
+    socket_.set_option(asio::socket_base::reuse_address(true));
+
+    return (false);
+}
+
+// Send a message.
+
+void
+UDPSocket::async_send(const void* data, size_t length,
+    const IOEndpoint* endpoint, IOCompletionCallback& callback)
+{
+    // Upconverting.  Not nice, but we have the problem that in the abstract
+    // layer we are given an IOEndpoint.  For UDP code it is a UDPEndpoint
+    // and for TCP code a TCPEndpoint.  However the member that we are
+    // after - the asio endpoint - is different for UPD and TCP and there is
+    // no common ancestor.  Hence the promotion here.
+    assert(endpoint->getProtocol() == IPPROTO_UDP);
+    const UDPEndpoint* udp_endpoint = static_cast<const UDPEndpoint*>(endpoint);
+
+    socket_.async_send_to(buffer(data, length), udp_endpoint->getASIOEndpoint(),
+        callback);
+}
+
+// UDPSocket::receive_from
+
+void
+UDPSocket::async_receive(void* data, size_t length, IOEndpoint* endpoint,
+    IOCompletionCallback& callback)
+{
+    // Upconvert the endpoint again.
+    assert(endpoint->getProtocol() == IPPROTO_UDP);
+    UDPEndpoint* udp_endpoint = static_cast<UDPEndpoint*>(endpoint);
+
+    socket_.async_receive_from(buffer(data, length),
+        udp_endpoint->getASIOEndpoint(), callback);
+}
+
+// Cancel I/O on the socket
+void
+UDPSocket::cancel() {
+    socket_.cancel();
+}