add option to disable ws connection to rpc node

Author: jhunsaker

pc/manager.cpp

@@ -45,7 +45,8 @@ void manager_sub::on_add_symbol( manager *, price * )
 // manager
 
 manager::manager()
-: thost_( PC_RPC_HOST ),
+: wconn_{ nullptr },
+  thost_( PC_RPC_HOST ),
   rhost_( PC_RPC_HOST ),
   sub_( nullptr ),
   status_( 0 ),
@@ -260,7 +261,12 @@ void manager::teardown()
 
   // destroy rpc connections
   hconn_.close();
-  wconn_.close();
+  if ( wconn_ ) {
+    wconn_->close();
+    delete wconn_;
+    wconn_ = nullptr;
+    clnt_.set_ws_conn( nullptr );
+  }
 }
 
 bool manager::init()
@@ -305,15 +311,18 @@ bool manager::init()
   hconn_.set_host( rhost );
   hconn_.set_net_loop( &nl_ );
   clnt_.set_http_conn( &hconn_ );
-  wconn_.set_port( wport );
-  wconn_.set_host( rhost );
-  wconn_.set_net_loop( &nl_ );
-  clnt_.set_ws_conn( &wconn_ );
+  if ( get_do_ws() ) {
+    wconn_ = new ws_connect{};
+    wconn_->set_port( wport );
+    wconn_->set_host( rhost );
+    wconn_->set_net_loop( &nl_ );
+    clnt_.set_ws_conn( wconn_ );
+  }
   if ( !hconn_.init() ) {
     return set_err_msg( hconn_.get_err_msg() );
   }
-  if ( !wconn_.init() ) {
-    return set_err_msg( wconn_.get_err_msg() );
+  if ( wconn_ && !wconn_->init() ) {
+    return set_err_msg( wconn_->get_err_msg() );
   }
   // connect to pyth_tx server
   if ( do_tx_ ) {
@@ -358,7 +367,7 @@ bool manager::get_is_tx_send() const
 
 bool manager::get_is_rpc_send() const
 {
-  return hconn_.get_is_send() || wconn_.get_is_send();
+  return hconn_.get_is_send() || ( wconn_ && wconn_->get_is_send() );
 }
 
 bool manager::bootstrap()
@@ -420,7 +429,9 @@ void manager::poll( bool do_wait )
   } else {
     if ( has_status( PC_PYTH_RPC_CONNECTED ) ) {
       hconn_.poll();
-      wconn_.poll();
+      if ( wconn_ ) {
+        wconn_->poll();
+      }
     }
     if ( do_tx_ ) {
       tconn_.poll();
@@ -476,7 +487,7 @@ void manager::poll( bool do_wait )
   // submit new quotes while connected
   if ( has_status( PC_PYTH_RPC_CONNECTED ) &&
        !hconn_.get_is_err() &&
-       !wconn_.get_is_err() ) {
+       ( !wconn_ || !wconn_->get_is_err() ) ) {
     poll_schedule();
   } else {
     reconnect_rpc();
@@ -506,15 +517,15 @@ void manager::reconnect_rpc()
   if ( hconn_.get_is_wait() ) {
     hconn_.check();
   }
-  if ( wconn_.get_is_wait() ) {
-    wconn_.check();
+  if ( wconn_ && wconn_->get_is_wait() ) {
+    wconn_->check();
   }
-  if ( hconn_.get_is_wait() || wconn_.get_is_wait() ) {
+  if ( hconn_.get_is_wait() || ( wconn_ && wconn_->get_is_wait() ) ) {
     return;
   }
 
   // check for successful (re)connect
-  if ( !hconn_.get_is_err() && !wconn_.get_is_err() ) {
+  if ( !hconn_.get_is_err() && ( !wconn_ || !wconn_->get_is_err() ) ) {
     PC_LOG_INF( "rpc_connected" ).end();
     set_status( PC_PYTH_RPC_CONNECTED );
 
@@ -613,7 +624,9 @@ void manager::reconnect_rpc()
   ctimeout_ = std::min( ctimeout_, PC_RECONNECT_TIMEOUT );
   wait_conn_ = true;
   hconn_.init();
-  wconn_.init();
+  if ( wconn_ ) {
+    wconn_->init();
+  }
 }
 
 void manager::log_disconnect()
@@ -626,11 +639,11 @@ void manager::log_disconnect()
       .end();
     return;
   }
-  if ( wconn_.get_is_err() ) {
+  if ( wconn_ && wconn_->get_is_err() ) {
     PC_LOG_ERR( "rpc_websocket_reset" )
-      .add( "error", wconn_.get_err_msg() )
+      .add( "error", wconn_->get_err_msg() )
       .add( "host", rhost_ )
-      .add( "port", wconn_.get_port() )
+      .add( "port", wconn_->get_port() )
       .end();
     return;
   }

pc/manager.hpp

@@ -215,7 +215,7 @@ namespace pc
 
     net_loop     nl_;       // epoll loop
     tcp_connect  hconn_;    // rpc http connection
-    ws_connect   wconn_;    // rpc websocket sonnection
+    ws_connect  *wconn_;    // rpc websocket sonnection
     tcp_listen   lsvr_;     // listening socket
     rpc_client   clnt_;     // rpc api
     tx_connect   tconn_;    // tx proxy connection

pc/rpc_client.cpp

@@ -123,7 +123,9 @@ net_connect *rpc_client::get_http_conn() const
 void rpc_client::set_ws_conn( net_connect *wptr )
 {
   wptr_ = wptr;
-  wptr_->set_net_parser( &wp_ );
+  if ( wptr_ ) {
+    wptr_->set_net_parser( &wp_ );
+  }
   wp_.set_net_connect( wptr_ );
 }
 
@@ -171,12 +173,15 @@ void rpc_client::send( rpc_request *rptr )
     msg.add_hdr( "Content-Type", "application/json" );
     msg.commit( jw );
     hptr_->add_send( msg );
-  } else {
+  } else if ( wptr_ ) {
     // submit websocket message
     ws_wtr msg;
     msg.commit( ws_wtr::text_id, jw, true );
     wptr_->add_send( msg );
   }
+  else {
+    PC_LOG_WRN( "no ws connection to send msg" ).end();
+  }
 }
 
 void rpc_client::rpc_http::parse_content( const char *txt, size_t len )

pcapps/pythd.cpp

@@ -62,6 +62,9 @@ int usage()
   std::cerr << "  -x" << std::endl;
   std::cerr << "     Disable connection to pyth_tx transaction proxy server"
                "\n" << std::endl;
+  std::cerr << "  -z" << std::endl;
+  std::cerr << "     Disable WebSocket connection to Solana RPC node"
+               "\n" << std::endl;
   std::cerr << "  -m <commitment_level>" << std::endl;
   std::cerr << "     Subscription commitment level: processed, confirmed or "
                "finalized\n" << std::endl;
@@ -98,8 +101,8 @@ int main(int argc, char **argv)
   std::string tx_host  = get_tx_host();
   int pyth_port = get_port();
   int opt = 0;
-  bool do_wait = true, do_tx = true, do_debug = false;
-  while( (opt = ::getopt(argc,argv, "r:t:p:k:w:c:l:m:dnxh" )) != -1 ) {
+  bool do_wait = true, do_tx = true, do_ws = true, do_debug = false;
+  while( (opt = ::getopt(argc,argv, "r:t:p:k:w:c:l:m:dnxhz" )) != -1 ) {
     switch(opt) {
       case 'r': rpc_host = optarg; break;
       case 't': tx_host = optarg; break;
@@ -111,6 +114,7 @@ int main(int argc, char **argv)
       case 'm': cmt = str_to_commitment(optarg); break;
       case 'n': do_wait = false; break;
       case 'x': do_tx = false; break;
+      case 'z': do_ws = false; break;
       case 'd': do_debug = true; break;
       default: return usage();
     }
@@ -138,6 +142,7 @@ int main(int argc, char **argv)
   mgr.set_content_dir( cnt_dir );
   mgr.set_capture_file( cap_file );
   mgr.set_do_tx( do_tx );
+  mgr.set_do_ws( do_ws );
   mgr.set_do_capture( !cap_file.empty() );
   mgr.set_commitment( cmt );
   if ( !mgr.init() ) {