[trac554] Added IOFetch

IOFetch is a general upstream "fetch" class that should be able to
operate over TCP or UDP.  Related changes have been made in the
associated classes.  So far, only the unit tests for a UDP fetch
have been made (and passed).

src/lib/asiolink/udp_socket.h

@@ -19,9 +19,19 @@
 #error "asio.hpp must be included before including this, see asiolink.h as to why"
 #endif
 
+#include <log/dummylog.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <unistd.h>             // for some IPC/network system calls
+
 #include <cstddef>
-#include <asiolink/io_socket.h>
+
+#include <config.h>
+
+#include <asiolink/io_asio_socket.h>
+#include <asiolink/io_endpoint.h>
 #include <asiolink/io_service.h>
+#include <asiolink/udp_endpoint.h>
 
 namespace asiolink {
 
@@ -29,7 +39,10 @@ namespace asiolink {
 /// \c IOSocket that represents a UDP socket.
 ///
 /// Other notes about \c TCPSocket applies to this class, too.
-class UDPSocket : public IOSocket {
+///
+/// \param C Callback type
+template <typename C>
+class UDPSocket : public IOAsioSocket<C> {
 private:
     /// \brief Class is non-copyable
     UDPSocket(const UDPSocket&);
@@ -42,15 +55,17 @@ public:
     
     /// \brief Constructor from an ASIO UDP socket.
     ///
-    /// \param socket The ASIO representation of the UDP socket.
-    UDPSocket(asio::ip::udp::socket& socket) :
-        socket_ptr_(NULL), socket_(socket)
-    {}
+    /// \param socket The ASIO representation of the UDP socket.  It
+    /// is assumed that the caller will open and close the socket, so
+    /// these operations are a no-op for that socket.
+    UDPSocket(asio::ip::udp::socket& socket);
 
     /// \brief Constructor
     ///
     /// Used when the UDPSocket is being asked to manage its own internal
-    /// socket.
+    /// socket.  It is assumed that open() and close() will not be used.
+    ///
+    /// \param service I/O Service object used to manage the socket.
     UDPSocket(IOService& service);
 
     /// \brief Destructor
@@ -70,7 +85,7 @@ public:
     /// \param callback Unused.
     ///
     /// \return false to indicate that the "operation" completed synchronously.
-    virtual bool open(const IOEndpoint* endpoint, IOCompletionCallback&);
+    virtual bool open(const IOEndpoint* endpoint, C&);
 
     /// \brief Send Asynchronously
     ///
@@ -83,7 +98,7 @@ public:
     /// \param endpoint Target of the send
     /// \param callback Callback object.
     virtual void asyncSend(const void* data, size_t length,
-        const IOEndpoint* endpoint, IOCompletionCallback& callback);
+        const IOEndpoint* endpoint, C& callback);
 
     /// \brief Receive Asynchronously
     ///
@@ -99,7 +114,7 @@ public:
     /// \param endpoint Source of the communication
     /// \param callback Callback object
     virtual void asyncReceive(void* data, size_t length, size_t cumulative,
-        IOEndpoint* endpoint, IOCompletionCallback& callback);
+        IOEndpoint* endpoint, C& callback);
 
     /// \brief Checks if the data received is complete.
     ///
@@ -130,9 +145,133 @@ private:
     // Two variables to hold the socket - a socket and a pointer to it.  This
     // handles the case where a socket is passed to the UDPSocket on
     // construction, or where it is asked to manage its own socket.
-    asio::ip::udp::socket*      socket_ptr_;    ///< Pointer to the socket
+    asio::ip::udp::socket*      socket_ptr_;    ///< Pointer to own socket
     asio::ip::udp::socket&      socket_;        ///< Socket
+    bool                        isopen_;        ///< true when socket is open
 };
 
-}      // namespace asiolink
+// Constructor - caller manages socket
+
+template <typename C>
+UDPSocket<C>::UDPSocket(asio::ip::udp::socket& socket) :
+    socket_ptr_(NULL), socket_(socket), isopen_(true)
+{
+}
+
+// Constructor - create socket on the fly
+
+template <typename C>
+UDPSocket<C>::UDPSocket(IOService& service) :
+    socket_ptr_(new asio::ip::udp::socket(service.get_io_service())),
+    socket_(*socket_ptr_), isopen_(false)
+{
+}
+
+// Destructor.  Only delete the socket if we are managing it.
+
+template <typename C>
+UDPSocket<C>::~UDPSocket()
+{
+    delete socket_ptr_;
+}
+
+// Open the socket.  Throws an error on failure
+// TODO: Make the open more resilient
+
+template <typename C> bool
+UDPSocket<C>::open(const IOEndpoint* endpoint, C&) {
+
+    // Ignore opens on already-open socket.  Don't throw a failure because
+    // of uncertainties as to what precedes whan when using asynchronous I/O.
+    // At also allows us a treat a passed-in socket as a self-managed socket.
+
+    if (!isopen_) {
+        if (endpoint->getFamily() == AF_INET) {
+            socket_.open(asio::ip::udp::v4());
+        }
+        else {
+            socket_.open(asio::ip::udp::v6());
+        }
+        isopen_ = true;
+
+        // 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.  Should never do this if the socket is not open, so throw
+// an exception if this is the case.
+
+template <typename C> void
+UDPSocket<C>::asyncSend(const void* data, size_t length,
+    const IOEndpoint* endpoint, C& callback)
+{
+    if (isopen_) {
+
+        // Upconvert to a UDPEndpoint.  We need to do this because although
+        // IOEndpoint is the base class of UDPEndpoint and TCPEndpoint, it
+        // doing cont contain a method for getting at the underlying endpoint
+        // type - those are in the derived class and the two classes differ on
+        // return type.
+
+        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);
+    } else {
+        isc_throw(SocketNotOpen,
+            "attempt to send on a UDP socket that is not open");
+    }
+}
+
+// Receive a message. Note that the "cumulative" argument is ignored - every UDP
+// receive is put into the buffer beginning at the start - there is no concept
+// receiving a subsequent part of a message.  Same critera as before concerning
+// the need for the socket to be open.
+
+template <typename C> void
+UDPSocket<C>::asyncReceive(void* data, size_t length, size_t,
+    IOEndpoint* endpoint, C& callback)
+{
+    if (isopen_) {
+
+        // 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);
+    } else {
+        isc_throw(SocketNotOpen,
+            "attempt to receive from a UDP socket that is not open");
+    }
+}
+
+// Cancel I/O on the socket.  No-op if the socket is not open.
+template <typename C> void
+UDPSocket<C>::cancel() {
+    if (isopen_) {
+        socket_.cancel();
+    }
+}
+
+// Close the socket down.  Can only do this if the socket is open and we are
+// managing it ourself.
+
+template <typename C> void
+UDPSocket<C>::close() {
+    if (isopen_ && socket_ptr_) {
+        socket_.close();
+        isopen_ = false;
+    }
+}
+
+} // namespace asiolink
+
 #endif // __UDP_SOCKET_H