collector

package
v0.0.0-...-28b05a4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 20, 2024 License: GPL-2.0 Imports: 24 Imported by: 0

Documentation

Index

Constants

View Source 
const ANCHOR_TIME_RANGE_SECONDS = 2
View Source 
const WORKER_COUNT = 2

Odd workers look at current. Even workers look ahead? Otherwise I can do a treadmill system.

Variables

View Source 
var Sources = []Source{}

The global table with all our sources.

View Source 
var SourcesBlockchainRPC = []Source{
	{
		"ethereum-ankr-rpc",
		joinEthereumRPC,
		nil,
		"https://rpc.ankr.com/eth",
		[]string{""},
		"",
	},
	{
		"polygon-ankr-rpc",
		joinEthereumRPC,
		nil,
		"https://rpc.ankr.com/polygon",
		[]string{""},
		"",
	},
}

Should handle these as a separate case, since blockchain information can exactly be consistent from different sources.

View Source 
var SourcesCEX = []Source{

	{
		"binance",
		defaultJoinCEX,
		panicStubFunctionSeeBodyForExplanation,
		"wss://stream.binance.com:9443/ws",
		[]string{"btcusdt", "ethusdt", "solusdt"},
		"{ \"method\": \"SUBSCRIBE\", \"params\": [ \"{{topic}}@aggTrade\" ], \"id\": 1 }"},
	{
		"coinbase",
		defaultJoinCEX,
		panicStubFunctionSeeBodyForExplanation,
		"wss://ws-feed.pro.coinbase.com",
		[]string{"BTC-USD", "ETH-USD", "BTC-ETH"},
		"{\"type\": \"subscribe\", \"product_ids\": [ \"{{topic}}\" ], \"channels\": [ \"ticker\" ]}"},

	{
		"dydx",
		defaultJoinCEX,
		panicStubFunctionSeeBodyForExplanation,
		"wss://api.dydx.exchange/v3/ws",
		[]string{"MATIC-USD", "LINK-USD", "SOL-USD", "ETH-USD", "BTC-USD"},
		"{\"type\": \"subscribe\", \"id\": \"{{topic}}\", \"channel\": \"v3_trades\"}",
	},
	{
		"bybit",
		defaultJoinCEX,
		panicStubFunctionSeeBodyForExplanation,
		"wss://stream.bybit.com/v5/public/spot",
		[]string{"orderbook.50.BTCUSDT", "publicTrade.BTCUSDT", "tickers.BTCUSDT", "kline.M.BTCUSDT"},
		`{"op": "subscribe","args": ["{{topic}}"]}`,
	},

	{
		"okx",
		defaultJoinCEX,
		panicStubFunctionSeeBodyForExplanation,
		"wss://ws.okx.com:8443/ws/v5/business",
		[]string{"sprd-bbo-tbt", "sprd-books5", "sprd-public-trades", "sprd-tickers"},

		`{"op": "subscribe","args": [{"channel": "{{topic}}","sprdId": "BTC-USDT_BTC-USDT-SWAP"}]}`,
	},
}
View Source 
var SourcesNFTExchange = []Source{}

Functions

func CalculateDataSize 

func CalculateDataSize(t *testing.T, ctx context.Context, dataWriter *DataWriter, timeToCollect int, timeFrameWindowSize int)

Handler to subscribe to each source and symbol & measure data size over a set period from current time.

func Subscribe 

func Subscribe(ctx context.Context, source Source, topic string) (chan []byte, error)

Subscribe will connect to the chosen source and create a channel which will return every message from it.

Types

type CollectorInstance 

type CollectorInstance struct {
	// contains filtered or unexported fields
}

func NewInstance 

func NewInstance(rpInstance *resourcepool.Instance) *CollectorInstance

func (*CollectorInstance) Start 

func (ci *CollectorInstance) Start(ctx context.Context)

func (*CollectorInstance) Stop 

func (ci *CollectorInstance) Stop()

func (*CollectorInstance) SubmitRequests 

func (ci *CollectorInstance) SubmitRequests(requests []Request, requestsNext []Request, targetAnchorTime time.Time, targetAnchorValue uint64) []Summary

type CollectorWorker 

type CollectorWorker struct {
	// contains filtered or unexported fields
}

type DataCollector 

type DataCollector struct {
	Source        Source
	TimeWindow    time.Duration
	DataWindows   map[string]*DataWindow
	BusiestWindow *DataWindow
}

func NewDataCollector 

func NewDataCollector(source Source, windowTimeSize time.Duration) *DataCollector

type DataWindow 

type DataWindow struct {
	SourceName   string
	Symbol       string
	StartTime    string
	DataSize     int64
	MessageCount int
	Throughput   float64
}

type DataWriter 

type DataWriter struct {
	// contains filtered or unexported fields
}

func NewDataWriter 

func NewDataWriter(filename string) (*DataWriter, error)

func (*DataWriter) Close 

func (dw *DataWriter) Close() error

func (*DataWriter) Write 

func (dw *DataWriter) Write(data []string) error

type MessageTimeHash 

type MessageTimeHash struct {
	// contains filtered or unexported fields
}

type Request 

type Request struct {
	Source Source
	Topic  int
}

type Source 

type Source struct {
	Name      string
	JoinFunc  func(ctx context.Context, source Source, topic string) (chan []byte, <-chan error, error)
	ParseFunc func(source Source, data []byte) []byte
	ApiURL    string // To-do: Add support for multiple endpoints.
	Topics    []string
	Request   string
}

Defines a "source" of data, all supported sources are laid out in the Sources array. We opt for this approach over oop for clarity and extensibility.

type SourceMetric 

type SourceMetric struct {
	// contains filtered or unexported fields
}

func NewSourceMetric 

func NewSourceMetric(sourceName string) *SourceMetric

type Summary 

type Summary struct {
	DataHashes []cid.Cid
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.