Merge remote-tracking branch 'origin/main' into devnet

Author: dragonfrond

doc/websocket_api.md

@@ -1,10 +1,12 @@
 # pyth-client websocket API
-The pythd daemon supports a websocket interface based on the jsoni-rpc 2.0 standard.  Methods include:
+The pythd daemon supports a websocket interface based on the json-rpc 2.0 standard.  Methods include:
 
 - [get_product_list](#get_product_list)
 - [update_price](#update_price)
 - [subscribe_price](#subscribe_price)
 - [subscribe_price_sched](#subscribe_price_sched)
+- [get_product](#get_product)
+- [get_all_products](#get_all_products)
 
 Batch requests are processed in the order the requests appear within the batch.
 
@@ -196,3 +198,178 @@ Where the result is an integer corresponding to a subscription identifier. All s
   }
 }
 ```
+
+## get_product
+
+Get full set of data for the given product.
+
+Request looks like:
+```
+{
+  "jsonrpc": "2.0",
+  "method": "get_product",
+  "params": {
+    "account": "4aDoSXJ5o3AuvL7QFeR6h44jALQfTmUUCTVGDD6aoJTM"
+  },
+  "id" : 1
+}
+```
+
+A successful response looks something like:
+```
+{
+  "jsonrpc": "2.0",
+  "result": {
+    "account": "4aDoSXJ5o3AuvL7QFeR6h44jALQfTmUUCTVGDD6aoJTM",
+    "attr_dict": {
+      "asset_type": "Crypto",
+      "symbol": "BTC/USD",
+      "country": "US",
+      "quote_currency": "USD",
+      "description": "BTC/USD",
+      "tenor": "Spot",
+      "generic_symbol": "BTCUSD"
+    },
+    "price_accounts": [
+      {
+        "account": "GVXRSBjFk6e6J3NbVPXohDJetcTjaeeuykUpbQF8UoMU",
+        "price_type": "price",
+        "price_exponent": -8,
+        "status": "trading",
+        "price": 4426101900000,
+        "conf": 4271150000,
+        "twap": 4433467600000,
+        "twac": 1304202670,
+        "valid_slot": 91402257,
+        "pub_slot": 91402259,
+        "prev_slot": 91402256,
+        "prev_price": 4425895500000,
+        "prev_conf": 3315350000,
+        "publisher_accounts": [
+          {
+            "account": "HekM1hBawXQu6wK6Ah1yw1YXXeMUDD2bfCHEzo25vnEB",
+            "status": "trading",
+            "price": 4426958500000,
+            "conf": 1492500000,
+            "slot": 91402255
+          },
+          {
+            "account": "GKNcUmNacSJo4S2Kq3DuYRYRGw3sNUfJ4tyqd198t6vQ",
+            "status": "trading",
+            "price": 4424690000000,
+            "conf": 3690000000,
+            "slot": 91402256
+          }
+        ]
+      }
+    ]
+  },
+  "id": 1
+}
+```
+
+## get_all_products
+
+Get full set of data for all products.
+
+Request looks like:
+```
+{
+  "jsonrpc": "2.0",
+  "method": "get_all_products",
+  "id" : 1
+}
+```
+
+A successful response looks something like:
+```
+{
+  "jsonrpc": "2.0",
+  "result": [
+    {
+      "account": "5uKdRzB3FzdmwyCHrqSGq4u2URja617jqtKkM71BVrkw",
+      "attr_dict": {
+        "asset_type": "Crypto",
+        "symbol": "BCH/USD",
+        "country": "US",
+        "quote_currency": "USD",
+        "description": "BCH/USD",
+        "tenor": "Spot",
+        "generic_symbol": "BCHUSD"
+      },
+      "price_accounts": [
+        {
+          "account": "5ALDzwcRJfSyGdGyhP3kP628aqBNHZzLuVww7o9kdspe",
+          "price_type": "price",
+          "price_exponent": -8,
+          "status": "trading",
+          "price": 60282000000,
+          "conf": 26000000,
+          "twap": 60321475000,
+          "twac": 22504746,
+          "valid_slot": 91402601,
+          "pub_slot": 91402604,
+          "prev_slot": 91402600,
+          "prev_price": 60282000000,
+          "prev_conf": 26000000,
+          "publisher_accounts": [
+            {
+              "account": "HekM1hBawXQu6wK6Ah1yw1YXXeMUDD2bfCHEzo25vnEB",
+              "status": "trading",
+              "price": 60282000000,
+              "conf": 26000000,
+              "slot": 91402599
+            },
+            {
+              "account": "2V7t5NaKY7aGkwytCWQgvUYZfEr9XMwNChhJEakTExk6",
+              "status": "unknown",
+              "price": 0,
+              "conf": 0,
+              "slot": 0
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "account": "3nuELNFBkbXqsXtnCzphRPCX6toKKYxVDnkyr9pTwB1K",
+      "attr_dict": {
+        "asset_type": "Crypto",
+        "symbol": "SABER/USD",
+        "country": "US",
+        "quote_currency": "USD",
+        "description": "SABER/USD",
+        "tenor": "Spot",
+        "generic_symbol": "SABERUSD"
+      },
+      "price_accounts": [
+        {
+          "account": "8Td9VML1nHxQK6M8VVyzsHo32D7VBk72jSpa9U861z2A",
+          "price_type": "price",
+          "price_exponent": -8,
+          "status": "trading",
+          "price": 5785000,
+          "conf": 5000,
+          "twap": 5856365,
+          "twac": 10241,
+          "valid_slot": 91402601,
+          "pub_slot": 91402604,
+          "prev_slot": 91402600,
+          "prev_price": 5785000,
+          "prev_conf": 5000,
+          "publisher_accounts": [
+            {
+              "account": "GKNcUmNacSJo4S2Kq3DuYRYRGw3sNUfJ4tyqd198t6vQ",
+              "status": "trading",
+              "price": 5785000,
+              "conf": 5000,
+              "slot": 91402601
+            }
+          ]
+        }
+      ]
+    }
+  ],
+  "id": 1
+}
+```

pc/attr_id.cpp

@@ -107,7 +107,7 @@ void attr_dict::write_account( net_wtr& wtr )
   }
 }
 
-void attr_dict::write_json( json_wtr& wtr )
+void attr_dict::write_json( json_wtr& wtr ) const
 {
   str vstr, kstr;
   for( unsigned id=1; id < avec_.size(); ++id ) {

pc/attr_id.hpp

@@ -59,7 +59,7 @@ namespace pc
     void write_account( net_wtr& );
 
     // serialize to json
-    void write_json( json_wtr& );
+    void write_json( json_wtr& ) const;
 
   private:
 

pc/manager.cpp

@@ -55,6 +55,7 @@ manager::manager()
   ctimeout_( PC_NSECS_IN_SEC ),
   slot_( 0UL ),
   slot_cnt_( 0UL ),
+  slot_ts_{ 0UL },
   curr_ts_( 0L ),
   pub_ts_( 0L ),
   pub_int_( PC_PUB_INTERVAL ),
@@ -69,6 +70,7 @@ manager::manager()
   breq_->set_sub( this );
   sreq_->set_sub( this );
   preq_->set_sub( this );
+  areq_->set_sub( this );
   tconn_.set_net_parser( &txp_ );
   txp_.mgr_ = this;
 }
@@ -130,6 +132,16 @@ std::string manager::get_tx_host() const
   return thost_;
 }
 
+void manager::set_do_ws( bool do_ws )
+{
+  do_ws_ = do_ws;
+}
+
+bool manager::get_do_ws() const
+{
+  return do_ws_;
+}
+
 void manager::set_do_tx( bool do_tx )
 {
   do_tx_ = do_tx;
@@ -343,6 +355,11 @@ bool manager::get_is_tx_send() const
   return tconn_.get_is_send();
 }
 
+bool manager::get_is_rpc_send() const
+{
+  return hconn_.get_is_send() || wconn_.get_is_send();
+}
+
 bool manager::bootstrap()
 {
   int status = PC_PYTH_RPC_CONNECTED | PC_PYTH_HAS_BLOCK_HASH;
@@ -441,6 +458,13 @@ void manager::poll( bool do_wait )
         clnt_.send( sreq_ );
       }
     }
+    if ( ! get_do_ws() ) {
+      if ( areq_->get_is_recv() ) {
+        if ( has_status( PC_PYTH_RPC_CONNECTED ) ) {
+          clnt_.send( areq_ );
+        }
+      }
+    }
   }
 
   // try to (re)connect to tx proxy
@@ -523,9 +547,15 @@ void manager::reconnect_rpc()
       set_err_msg( "missing or invalid program public key [" +
           get_program_pub_key_file() + "]" );
     } else {
-      preq_->set_commitment( get_commitment() );
-      preq_->set_program( gpub );
-      clnt_.send( preq_ );
+      if ( get_do_ws() ) {
+        preq_->set_commitment( get_commitment() );
+        preq_->set_program( get_program_pub_key() );
+        clnt_.send( preq_ );
+      }
+      else {
+        areq_->set_commitment( get_commitment() );
+        areq_->set_program( get_program_pub_key() );
+      }
     }
 
     // gather latest info on mapping accounts
@@ -673,7 +703,7 @@ void manager::on_response( rpc::get_slot *res )
 
   PC_LOG_DBG( "received get_slot" )
     .add( "slot", slot_ )
-    .add( "rount_trip_time(ms)", 1e-6*ack_ts )
+    .add( "round_trip_time(ms)", 1e-6*ack_ts )
     .end();
 
   // submit block hash every N slots
@@ -710,18 +740,34 @@ void manager::on_response( rpc::get_recent_block_hash *m )
   PC_LOG_INF( "received_recent_block_hash" )
     .add( "curr_slot", slot_ )
     .add( "hash_slot", m->get_slot() )
-    .add( "rount_trip_time(ms)", 1e-6*ack_ts )
+    .add( "round_trip_time(ms)", 1e-6*ack_ts )
     .end();
 
 }
 
-void manager::on_response( rpc::program_subscribe *m )
+void manager::on_response( rpc::account_update *m )
 {
   if ( m->get_is_err() ) {
-    set_err_msg( "failed to program_subscribe ["
+    set_err_msg( "account update failed ["
         + m->get_err_msg()  + "]" );
     return;
   }
+
+  if ( m->get_is_http() ) {
+    int64_t ack_ts = m->get_recv_time() - m->get_sent_time();
+    PC_LOG_DBG( "received account_update" )
+      .add( "account", *m->get_account() )
+      .add( "slot", slot_ )
+      .add( "round_trip_time(ms)", 1e-6*ack_ts )
+      .end();
+  }
+  else {
+    PC_LOG_DBG( "received account_update" )
+      .add( "account", *m->get_account() )
+      .add( "slot", slot_ )
+      .end();
+  }
+
   // look up by account and dispatch update
   acc_map_t::iter_t it = amap_.find( *m->get_account() );
   if ( it ) {

pc/manager.hpp

@@ -50,7 +50,7 @@ namespace pc
                   public rpc_sub,
                   public rpc_sub_i<rpc::get_slot>,
                   public rpc_sub_i<rpc::get_recent_block_hash>,
-                  public rpc_sub_i<rpc::program_subscribe>
+                  public rpc_sub_i<rpc::account_update>
   {
   public:
 
@@ -65,6 +65,10 @@ namespace pc
     void set_tx_host( const std::string& );
     std::string get_tx_host() const;
 
+    // turn on/off ws subscriptions
+    void set_do_ws( bool );
+    bool get_do_ws() const;
+
     // turn on/off tx proxy mode
     void set_do_tx( bool );
     bool get_do_tx() const;
@@ -166,9 +170,10 @@ namespace pc
     // rpc callbacks
     void on_response( rpc::get_slot * ) override;
     void on_response( rpc::get_recent_block_hash * ) override;
-    void on_response( rpc::program_subscribe * ) override;
+    void on_response( rpc::account_update * ) override;
     void set_status( int );
     get_mapping *get_last_mapping() const;
+    bool get_is_rpc_send() const;
 
   private:
 
@@ -235,6 +240,7 @@ namespace pc
     kpx_vec_t    kvec_;     // symbol price scheduling
     bool         wait_conn_;// waiting on connection
     bool         do_cap_;   // do capture flag
+    bool         do_ws_;    // do ws subscriptions
     bool         do_tx_;    // do tx proxy connectivity
     bool         is_pub_;   // is publishing mode
     capture      cap_;      // aggregate price capture
@@ -245,6 +251,7 @@ namespace pc
     rpc::get_slot              sreq_[1]; // slot subscription
     rpc::get_recent_block_hash breq_[1]; // block hash request
     rpc::program_subscribe     preq_[1]; // program account subscription
+    rpc::get_program_accounts  areq_[1]; // alternative to program_subscribe
   };
 
   inline bool manager::get_is_tx_connect() const

pc/misc.hpp

@@ -4,7 +4,8 @@
 #include <string>
 
 #define PC_PACKED        __attribute__((__packed__))
-#define PC_UNLIKELY(ARG) __builtin_expect((ARG),1)
+#define PC_LIKELY(ARG)   __builtin_expect((ARG),1)
+#define PC_UNLIKELY(ARG) __builtin_expect((ARG),0)
 #define PC_NSECS_IN_SEC  1000000000L
 #define PC_NSECS_IN_MSEC 1000000L