More logging

include/tao/pq/connection.hpp

@@ -198,6 +198,8 @@ namespace tao::pq
 
       [[nodiscard]] auto flush() -> bool;
 
+      void consume_input();
+
       [[nodiscard]] auto direct() -> std::shared_ptr< pq::transaction >;
 
       [[nodiscard]] auto transaction() -> std::shared_ptr< pq::transaction >;

include/tao/pq/log.hpp

@@ -48,6 +48,14 @@ namespace tao::pq
          using get_result_t = std::function< void( connection&, std::chrono::steady_clock::time_point end ) >;
          using get_result_result_t = std::function< void( connection&, PGresult* ) >;
 
+         using enter_pipeline_mode_result_t = std::function< void( connection&, int result ) >;
+
+         using exit_pipeline_mode_t = std::function< void( connection& ) >;
+         using exit_pipeline_mode_result_t = std::function< void( connection&, int result ) >;
+
+         using pipeline_sync_t = std::function< void( connection& ) >;
+         using pipeline_sync_result_t = std::function< void( connection&, int result ) >;
+
          struct : send_query_t
          {
             send_query_result_t result;
@@ -91,6 +99,23 @@ namespace tao::pq
             using get_result_t::operator=;
          } get_result;
 
+         struct
+         {
+            enter_pipeline_mode_result_t result;
+         } enter_pipeline_mode;
+
+         struct : exit_pipeline_mode_t
+         {
+            exit_pipeline_mode_result_t result;
+            using exit_pipeline_mode_t::operator=;
+         } exit_pipeline_mode;
+
+         struct : pipeline_sync_t
+         {
+            pipeline_sync_result_t result;
+            using pipeline_sync_t::operator=;
+         } pipeline_sync;
+
       } connection;
 
       struct transaction_t

src/lib/pq/connection.cpp

@@ -504,21 +504,39 @@ namespace tao::pq
 
    void connection::enter_pipeline_mode()
    {
-      if( PQenterPipelineMode( m_pgconn.get() ) == 0 ) {
+      const auto result = PQenterPipelineMode( m_pgconn.get() );
+      if( m_log && m_log->connection.enter_pipeline_mode.result ) {
+         m_log->connection.enter_pipeline_mode.result( *this, result );
+      }
+      if( result == 0 ) {
          throw pq::connection_error( "unable to enter pipeline mode" );
       }
    }
 
    void connection::exit_pipeline_mode()
    {
-      if( PQexitPipelineMode( m_pgconn.get() ) == 0 ) {
+      if( m_log && m_log->connection.exit_pipeline_mode ) {
+         m_log->connection.exit_pipeline_mode( *this );
+      }
+      const auto result = PQexitPipelineMode( m_pgconn.get() );
+      if( m_log && m_log->connection.exit_pipeline_mode.result ) {
+         m_log->connection.exit_pipeline_mode.result( *this, result );
+      }
+      if( result == 0 ) {
          throw pq::connection_error( error_message() );
       }
    }
 
    void connection::pipeline_sync()
    {
-      if( PQpipelineSync( m_pgconn.get() ) == 0 ) {
+      if( m_log && m_log->connection.pipeline_sync ) {
+         m_log->connection.pipeline_sync( *this );
+      }
+      const auto result = PQpipelineSync( m_pgconn.get() );
+      if( m_log && m_log->connection.pipeline_sync.result ) {
+         m_log->connection.pipeline_sync.result( *this, result );
+      }
+      if( result == 0 ) {
          throw pq::connection_error( "unable to sync pipeline" );
       }
    }
@@ -553,6 +571,20 @@ namespace tao::pq
       }
    }
 
+   void connection::consume_input()
+   {
+      if( m_log && m_log->connection.consume_input ) {
+         m_log->connection.consume_input( *this );
+      }
+      const auto result = PQconsumeInput( m_pgconn.get() );
+      if( m_log && m_log->connection.consume_input.result ) {
+         m_log->connection.consume_input.result( *this, result );
+      }
+      if( result == 0 ) {
+         throw pq::connection_error( error_message() );
+      }
+   }
+
    auto connection::direct() -> std::shared_ptr< pq::transaction >
    {
       return std::make_shared< internal::autocommit_transaction >( shared_from_this() );
@@ -660,16 +692,7 @@ namespace tao::pq
 
    void connection::get_notifications()
    {
-      if( m_log && m_log->connection.consume_input ) {
-         m_log->connection.consume_input( *this );
-      }
-      const auto result = PQconsumeInput( m_pgconn.get() );
-      if( m_log && m_log->connection.consume_input.result ) {
-         m_log->connection.consume_input.result( *this, result );
-      }
-      if( result == 0 ) {
-         throw pq::connection_error( error_message() );
-      }
+      consume_input();
       handle_notifications();
    }