client

Documentation

Index

• Variables
• func RunPriceUpdaterTaskLoop(ctx context.Context, exchangeToMarketPrices types.ExchangeToMarketPrices, ...) error
• type Client
  • func StartNewClient(ctx context.Context, daemonFlags flags.DaemonFlags, appFlags appflags.Flags, ...) (client *Client)
  • func (c *Client) Stop()
• type SubTaskRunner
• type SubTaskRunnerImpl
  • func (s *SubTaskRunnerImpl) StartPriceEncoder(exchangeId types.ExchangeId, configs types.PricefeedMutableMarketConfigs, ...)
  • func (s *SubTaskRunnerImpl) StartPriceFetcher(ticker *time.Ticker, stop <-chan bool, ...)
  • func (s *SubTaskRunnerImpl) StartPriceUpdater(c *Client, ctx context.Context, ticker *time.Ticker, stop <-chan bool, ...)

Variables

var HttpClient = http.Client{
	Transport: &http.Transport{MaxConnsPerHost: constants.MaxConnectionsPerExchange},
}

Functions

func RunPriceUpdaterTaskLoop

func RunPriceUpdaterTaskLoop(
	ctx context.Context,
	exchangeToMarketPrices types.ExchangeToMarketPrices,
	priceFeedServiceClient servertypes.PriceFeedServiceClient,
	logger log.Logger,
) error

RunPriceUpdaterTaskLoop copies the map of current `exchangeId -> MarketPriceTimestamp`, transforms the map values into a market price update request and sends the request to the socket where the pricefeed server is listening.

Types

type Client

type Client struct {
	// include HealthCheckable to track the health of the daemon.
	daemontypes.HealthCheckable
	// contains filtered or unexported fields
}

Client encapsulates the logic for executing and cleanly stopping all subtasks associated with the pricefeed client daemon. Access to the client's internal state is synchronized. The pricefeed daemon is a job that periodically queries external exchanges and transmits price data to the pricefeed service, which is then used by the application to compute index prices for proposing and validating oracle price updates on the blockchain. Note: price fetchers manage their own subtasks by blocking on their completion on every subtask run. When the price fetcher is stopped, it will wait for all of its own subtasks to complete before returning.

func StartNewClient

func StartNewClient(
	ctx context.Context,
	daemonFlags flags.DaemonFlags,
	appFlags appflags.Flags,
	logger log.Logger,
	grpcClient daemontypes.GrpcClient,
	marketParams []types.MarketParam,
	exchangeIdToQueryConfig map[types.ExchangeId]*types.ExchangeQueryConfig,
	exchangeIdToExchangeDetails map[types.ExchangeId]types.ExchangeQueryDetails,
	subTaskRunner SubTaskRunner,
) (client *Client)

StartNewClient initializes and starts a new pricefeed daemon as a subtask of the calling process. The pricefeed daemon is a job that periodically queries external exchanges and transmits price data to the pricefeed service, which is then used by the application to compute index prices for proposing and validating oracle price updates on the blockchain. Note: the daemon will panic if it fails to start up.

func (*Client) Stop

func (c *Client) Stop()

Stop stops the daemon and all running subtasks. This method is synchronized by the daemonStartup WaitGroup.

type SubTaskRunner

type SubTaskRunner interface {
	StartPriceUpdater(
		c *Client,
		ctx context.Context,
		ticker *time.Ticker,
		stop <-chan bool,
		exchangeToMarketPrices types.ExchangeToMarketPrices,
		priceFeedServiceClient servertypes.PriceFeedServiceClient,
		logger log.Logger,
	)
	StartPriceEncoder(
		exchangeId types.ExchangeId,
		configs types.PricefeedMutableMarketConfigs,
		exchangeToMarketPrices types.ExchangeToMarketPrices,
		logger log.Logger,
		bCh <-chan *price_fetcher.PriceFetcherSubtaskResponse,
	)
	StartPriceFetcher(
		ticker *time.Ticker,
		stop <-chan bool,
		configs types.PricefeedMutableMarketConfigs,
		exchangeQueryConfig types.ExchangeQueryConfig,
		exchangeDetails types.ExchangeQueryDetails,
		queryHandler handler.ExchangeQueryHandler,
		logger log.Logger,
		bCh chan<- *price_fetcher.PriceFetcherSubtaskResponse,
	)
}

SubTaskRunner is the interface for running pricefeed client task functions.

type SubTaskRunnerImpl

type SubTaskRunnerImpl struct{}

SubTaskRunnerImpl is the struct that implements the `SubTaskRunner` interface.

func (*SubTaskRunnerImpl) StartPriceEncoder

func (s *SubTaskRunnerImpl) StartPriceEncoder(
	exchangeId types.ExchangeId,
	configs types.PricefeedMutableMarketConfigs,
	exchangeToMarketPrices types.ExchangeToMarketPrices,
	logger log.Logger,
	bCh <-chan *price_fetcher.PriceFetcherSubtaskResponse,
)

StartPriceEncoder continuously reads from a buffered channel, reading encoded API responses for exchange requests and inserting them into an `ExchangeToMarketPrices` cache, performing currency conversions based on the index price of other markets as necessary. StartPriceEncoder reads price fetcher responses from a shared channel, and does not need a ticker or stop signal from the daemon to exit. It marks itself as done in the daemon's wait group when the price fetcher closes the shared channel.

func (*SubTaskRunnerImpl) StartPriceFetcher

func (s *SubTaskRunnerImpl) StartPriceFetcher(
	ticker *time.Ticker,
	stop <-chan bool,
	configs types.PricefeedMutableMarketConfigs,
	exchangeQueryConfig types.ExchangeQueryConfig,
	exchangeDetails types.ExchangeQueryDetails,
	queryHandler handler.ExchangeQueryHandler,
	logger log.Logger,
	bCh chan<- *price_fetcher.PriceFetcherSubtaskResponse,
)

StartPriceFetcher periodically starts goroutines to "fetch" market prices from a specific exchange. Each goroutine does the following: 1) query a single market price from a specific exchange 2) transform response to `MarketPriceTimestamp` 3) send transformed response to a buffered channel that's shared across multiple goroutines NOTE: the subtask response shared channel has a buffer size and goroutines will block if the buffer is full. NOTE: the price fetcher kicks off 1 to n go routines every time the subtask loop runs, but the subtask loop blocks until all go routines are done. This means that these go routines are not tracked by the wait group.

func (*SubTaskRunnerImpl) StartPriceUpdater

func (s *SubTaskRunnerImpl) StartPriceUpdater(
	c *Client,
	ctx context.Context,
	ticker *time.Ticker,
	stop <-chan bool,
	exchangeToMarketPrices types.ExchangeToMarketPrices,
	priceFeedServiceClient servertypes.PriceFeedServiceClient,
	logger log.Logger,
)

StartPriceUpdater periodically runs a task loop to send price updates to the pricefeed server via: 1) Get `MarketPriceTimestamps` for all exchanges in an `ExchangeToMarketPrices` struct. 2) Transform `MarketPriceTimestamps` and exchange ids into an `UpdateMarketPricesRequest` struct. StartPriceUpdater runs in the daemon's main goroutine and does not need access to the daemon's wait group to signal task completion.