MaybeSecureSocketAdaptor unit tests

This commit changes MaybeSecureSocketAdaptor to template a few critical
types which allows us to write unit tests against some of it's behaviours

In theory this could probably also replace the SocketIntercept things, which
were a nice solution when the implementation of this class passed straight through
to the asio types.

Since working around some asio bugs bloomberg#69
and adding support for data rate limits bloomberg#88
the MaybeSecureSocketAdaptor class has expanded enough that we should
be writing unit tests for it - and probably should have done so before now.

libamqpprox/amqpprox_maybesecuresocketadaptor.h

@@ -38,32 +38,36 @@ namespace amqpprox {
  * not secured, or the top level functions which pass through openssl if it is
  * secured.
  */
+template <typename StreamType =
+              boost::asio::ssl::stream<boost::asio::ip::tcp::socket>,
+          typename TimerType  = boost::asio::steady_timer,
+          typename IoContext  = boost::asio::io_context,
+          typename TlsContext = boost::asio::ssl::context>
 class MaybeSecureSocketAdaptor {
-    using stream_type = boost::asio::ssl::stream<boost::asio::ip::tcp::socket>;
-    using endpoint    = boost::asio::ip::tcp::endpoint;
+    using endpoint       = boost::asio::ip::tcp::endpoint;
     using handshake_type = boost::asio::ssl::stream_base::handshake_type;
 
-    boost::asio::io_context                               &d_ioContext;
+    IoContext                                             &d_ioContext;
     std::optional<std::reference_wrapper<SocketIntercept>> d_intercept;
-    std::unique_ptr<stream_type>                           d_socket;
+    std::unique_ptr<StreamType>                            d_socket;
     bool                                                   d_secured;
     bool                                                   d_handshook;
     char                                                   d_smallBuffer;
     bool                                                   d_smallBufferSet;
 
-    DataRateLimit             d_dataRateLimit;
-    DataRateLimit             d_dataRateAlarm;
-    boost::asio::steady_timer d_dataRateTimer;
-    bool                      d_alarmed;
-    bool                      d_dataRateTimerStarted;
+    DataRateLimit d_dataRateLimit;
+    DataRateLimit d_dataRateAlarm;
+    TimerType     d_dataRateTimer;
+    bool          d_alarmed;
+    bool          d_dataRateTimerStarted;
 
   public:
-    typedef typename stream_type::executor_type executor_type;
+    typedef typename StreamType::executor_type executor_type;
 
 #ifdef SOCKET_TESTING
-    MaybeSecureSocketAdaptor(boost::asio::io_context &ioContext,
-                             SocketIntercept         &intercept,
-                             bool                     secured)
+    MaybeSecureSocketAdaptor(IoContext       &ioContext,
+                             SocketIntercept &intercept,
+                             bool             secured)
     : d_ioContext(ioContext)
     , d_intercept(intercept)
     , d_socket()
@@ -80,12 +84,12 @@ class MaybeSecureSocketAdaptor {
     }
 #endif
 
-    MaybeSecureSocketAdaptor(boost::asio::io_context   &ioContext,
-                             boost::asio::ssl::context &context,
-                             bool                       secured)
+    MaybeSecureSocketAdaptor(IoContext  &ioContext,
+                             TlsContext &context,
+                             bool        secured)
     : d_ioContext(ioContext)
     , d_intercept()
-    , d_socket(std::make_unique<stream_type>(ioContext, context))
+    , d_socket(std::make_unique<StreamType>(ioContext, context))
     , d_secured(secured)
     , d_handshook(false)
     , d_smallBuffer(0)
@@ -112,7 +116,7 @@ class MaybeSecureSocketAdaptor {
     , d_alarmed(false)
     , d_dataRateTimerStarted(false)
     {
-        src.d_socket         = std::unique_ptr<stream_type>();
+        src.d_socket         = std::unique_ptr<StreamType>();
         src.d_secured        = false;
         src.d_handshook      = false;
         src.d_smallBuffer    = 0;

libamqpprox/amqpprox_server.cpp

@@ -199,15 +199,15 @@ void Server::doAccept(int port, bool secure)
         return;
     }
 
-    std::shared_ptr<MaybeSecureSocketAdaptor> incomingSocket =
-        std::make_shared<MaybeSecureSocketAdaptor>(
+    std::shared_ptr<MaybeSecureSocketAdaptor<>> incomingSocket =
+        std::make_shared<MaybeSecureSocketAdaptor<>>(
             d_ioContext, d_ingressTlsContext, secure);
 
     it->second.async_accept(
         incomingSocket->socket(),
         [this, port, secure, incomingSocket](error_code ec) {
             if (!ec) {
-                MaybeSecureSocketAdaptor clientSocket(
+                MaybeSecureSocketAdaptor<> clientSocket(
                     d_ioContext, d_egressTlsContext, false);
 
                 auto session =

libamqpprox/amqpprox_session.cpp

@@ -112,8 +112,8 @@ void logException(const std::string_view error,
 }
 
 Session::Session(boost::asio::io_context               &ioContext,
-                 MaybeSecureSocketAdaptor             &&serverSocket,
-                 MaybeSecureSocketAdaptor             &&clientSocket,
+                 MaybeSecureSocketAdaptor<>           &&serverSocket,
+                 MaybeSecureSocketAdaptor<>           &&clientSocket,
                  ConnectionSelectorInterface           *connectionSelector,
                  EventSource                           *eventSource,
                  BufferPool                            *bufferPool,
@@ -684,9 +684,9 @@ void Session::backendDisconnect()
     });
 }
 
-void Session::handleWriteData(FlowType                  direction,
-                              MaybeSecureSocketAdaptor &writeSocket,
-                              Buffer                    data)
+void Session::handleWriteData(FlowType                    direction,
+                              MaybeSecureSocketAdaptor<> &writeSocket,
+                              Buffer                      data)
 {
     auto self(shared_from_this());
     auto writeHandler = [this, self, direction](error_code ec, std::size_t) {

libamqpprox/amqpprox_session.h

@@ -61,8 +61,8 @@ class Session : public std::enable_shared_from_this<Session> {
         std::chrono::time_point<std::chrono::high_resolution_clock>;
 
     boost::asio::io_context     &d_ioContext;
-    MaybeSecureSocketAdaptor     d_serverSocket;
-    MaybeSecureSocketAdaptor     d_clientSocket;
+    MaybeSecureSocketAdaptor<>   d_serverSocket;
+    MaybeSecureSocketAdaptor<>   d_clientSocket;
     BufferHandle                 d_serverDataHandle;
     BufferHandle                 d_serverWriteDataHandle;
     BufferHandle                 d_clientDataHandle;
@@ -90,8 +90,8 @@ class Session : public std::enable_shared_from_this<Session> {
   public:
     // CREATORS
     Session(boost::asio::io_context                       &ioContext,
-            MaybeSecureSocketAdaptor                     &&serverSocket,
-            MaybeSecureSocketAdaptor                     &&clientSocket,
+            MaybeSecureSocketAdaptor<>                   &&serverSocket,
+            MaybeSecureSocketAdaptor<>                   &&clientSocket,
             ConnectionSelectorInterface                   *connectionSelector,
             EventSource                                   *eventSource,
             BufferPool                                    *bufferPool,
@@ -238,9 +238,9 @@ class Session : public std::enable_shared_from_this<Session> {
      * \param writeSocket to write the data
      * \param data to be written onto the outgoing socket
      */
-    void handleWriteData(FlowType                  direction,
-                         MaybeSecureSocketAdaptor &writeSocket,
-                         Buffer                    data);
+    void handleWriteData(FlowType                    direction,
+                         MaybeSecureSocketAdaptor<> &writeSocket,
+                         Buffer                      data);
 
     /**
      * \brief Handle errors on an established connection
@@ -280,7 +280,7 @@ class Session : public std::enable_shared_from_this<Session> {
      * \param direction specifies direction of the data flow (ingress/egress)
      * \return a mutable reference to the socket to read from
      */
-    inline MaybeSecureSocketAdaptor &readSocket(FlowType direction);
+    inline MaybeSecureSocketAdaptor<> &readSocket(FlowType direction);
 
     /**
      * \param direction specifies direction of the data flow (ingress/egress)
@@ -329,7 +329,7 @@ class Session : public std::enable_shared_from_this<Session> {
     inline bool &currentlyReading(FlowType direction);
 };
 
-inline MaybeSecureSocketAdaptor &Session::readSocket(FlowType direction)
+inline MaybeSecureSocketAdaptor<> &Session::readSocket(FlowType direction)
 {
     return (direction == FlowType::INGRESS) ? d_serverSocket : d_clientSocket;
 }

tests/CMakeLists.txt

@@ -18,14 +18,19 @@ add_executable(amqpprox_tests
     amqpprox_bufferhandle.t.cpp
     amqpprox_bufferpool.t.cpp
     amqpprox_buffersource.t.cpp
+    amqpprox_connectionlimitermanager.t.cpp
+    amqpprox_connectionselector.t.cpp
     amqpprox_connectionstats.t.cpp
     amqpprox_dataratelimit.t.cpp
+    amqpprox_defaultauthintercept.t.cpp
     amqpprox_dnsresolver.t.cpp
     amqpprox_eventsourcesignal.t.cpp
     amqpprox_farmstore.t.cpp
-    amqpprox_frame.t.cpp
+    amqpprox_fixedwindowconnectionratelimiter.t.cpp
     amqpprox_flowtype.t.cpp
-    amqpprox_connectionselector.t.cpp
+    amqpprox_frame.t.cpp
+    amqpprox_httpauthintercept.t.cpp
+    amqpprox_maybesecuresocketadaptor.t.cpp
     amqpprox_methods_start.t.cpp
     amqpprox_packetprocessor.t.cpp
     amqpprox_partitionpolicystore.t.cpp
@@ -38,10 +43,6 @@ add_executable(amqpprox_tests
     amqpprox_statsnapshot.t.cpp
     amqpprox_types.t.cpp
     amqpprox_vhoststate.t.cpp
-    amqpprox_defaultauthintercept.t.cpp
-    amqpprox_httpauthintercept.t.cpp
-    amqpprox_fixedwindowconnectionratelimiter.t.cpp
-    amqpprox_connectionlimitermanager.t.cpp
     )
 
 target_include_directories(amqpprox_tests PRIVATE ${PROTO_HDR_PATH})