serve

Documentation

Index

• Constants
• type Config
  • func NewConfig() (*Config, error)
• type Flag
• type PublicServerAPI
  • func NewPublicServerAPI(w Server, client *rpc.Client) *PublicServerAPI
  • func (api *PublicServerAPI) Stream(ctx context.Context, params eth.SubscriptionSettings) (*rpc.Subscription, error)
  • func (api *PublicServerAPI) WatchAddress(operation types.OperationType, args []types.WatchAddressArg) error
• type Server
  • func NewServer(settings *Config) (Server, error)
• type Service
  • func (sap *Service) APIs() []rpc.API
  • func (sap *Service) Backend() *eth.Backend
  • func (sap *Service) Protocols() []p2p.Protocol
  • func (sap *Service) Serve(wg *sync.WaitGroup, screenAndServePayload <-chan eth.ConvertedPayload)
  • func (sap *Service) Start() error
  • func (sap *Service) Stop() error
  • func (sap *Service) Subscribe(id rpc.ID, sub chan<- SubscriptionPayload, quitChan chan<- bool, ...)
  • func (sap *Service) Unsubscribe(id rpc.ID)
• type Subscription
• type SubscriptionPayload
  • func (sp SubscriptionPayload) BackFillComplete() bool
  • func (sp SubscriptionPayload) Error() error

Constants

const (
	SERVER_WS_PATH   = "SERVER_WS_PATH"
	SERVER_IPC_PATH  = "SERVER_IPC_PATH"
	SERVER_HTTP_PATH = "SERVER_HTTP_PATH"

	SERVER_MAX_IDLE_CONNECTIONS = "SERVER_MAX_IDLE_CONNECTIONS"
	SERVER_MAX_OPEN_CONNECTIONS = "SERVER_MAX_OPEN_CONNECTIONS"
	SERVER_MAX_CONN_LIFETIME    = "SERVER_MAX_CONN_LIFETIME"

	ETH_DEFAULT_SENDER_ADDR = "ETH_DEFAULT_SENDER_ADDR"
	ETH_RPC_GAS_CAP         = "ETH_RPC_GAS_CAP"
	ETH_CHAIN_CONFIG        = "ETH_CHAIN_CONFIG"
	ETH_SUPPORTS_STATEDIFF  = "ETH_SUPPORTS_STATEDIFF"
	ETH_FORWARD_ETH_CALLS   = "ETH_FORWARD_ETH_CALLS"
	ETH_PROXY_ON_ERROR      = "ETH_PROXY_ON_ERROR"

	VALIDATOR_ENABLED         = "VALIDATOR_ENABLED"
	VALIDATOR_EVERY_NTH_BLOCK = "VALIDATOR_EVERY_NTH_BLOCK"
)

Env variables

const (
	HTTP_TIMEOUT = "HTTP_TIMEOUT"

	ETH_WS_PATH       = "ETH_WS_PATH"
	ETH_HTTP_PATH     = "ETH_HTTP_PATH"
	ETH_NODE_ID       = "ETH_NODE_ID"
	ETH_CLIENT_NAME   = "ETH_CLIENT_NAME"
	ETH_GENESIS_BLOCK = "ETH_GENESIS_BLOCK"
	ETH_NETWORK_ID    = "ETH_NETWORK_ID"
	ETH_CHAIN_ID      = "ETH_CHAIN_ID"

	DATABASE_NAME                 = "DATABASE_NAME"
	DATABASE_HOSTNAME             = "DATABASE_HOSTNAME"
	DATABASE_PORT                 = "DATABASE_PORT"
	DATABASE_USER                 = "DATABASE_USER"
	DATABASE_PASSWORD             = "DATABASE_PASSWORD"
	DATABASE_MAX_IDLE_CONNECTIONS = "DATABASE_MAX_IDLE_CONNECTIONS"
	DATABASE_MAX_OPEN_CONNECTIONS = "DATABASE_MAX_OPEN_CONNECTIONS"
	DATABASE_MAX_CONN_LIFETIME    = "DATABASE_MAX_CONN_LIFETIME"
)

Env variables

const APIName = "vdb"

APIName is the namespace used for the state diffing service API

const APIVersion = "0.0.1"

APIVersion is the version of the state diffing service API

const (
	PayloadChanBufferSize = 2000
)

Types

type Config

type Config struct {
	DB       *sqlx.DB
	DBConfig postgres.Config

	WSEnabled  bool
	WSEndpoint string

	HTTPEnabled  bool
	HTTPEndpoint string

	IPCEnabled  bool
	IPCEndpoint string

	EthGraphqlEnabled  bool
	EthGraphqlEndpoint string

	IpldGraphqlEnabled          bool
	IpldGraphqlEndpoint         string
	IpldPostgraphileEndpoint    string
	TracingHttpEndpoint         string
	TracingPostgraphileEndpoint string

	ChainConfig      *params.ChainConfig
	DefaultSender    *common.Address
	RPCGasCap        *big.Int
	EthHttpEndpoint  string
	Client           *rpc.Client
	SupportStateDiff bool
	ForwardEthCalls  bool
	ProxyOnError     bool
	NodeNetworkID    string

	// Cache configuration.
	GroupCache *ethServerShared.GroupCacheConfig

	StateValidationEnabled       bool
	StateValidationEveryNthBlock uint64
}

Config struct

func NewConfig

func NewConfig() (*Config, error)

NewConfig is used to initialize a watcher config from a .toml file Separate chain watcher instances need to be ran with separate ipfs path in order to avoid lock contention on the ipfs repository lockfile

type Flag

type Flag int32

const (
	EmptyFlag Flag = iota
	BackFillCompleteFlag
)

type PublicServerAPI

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

PublicServerAPI is the public api for the watcher

func NewPublicServerAPI

func NewPublicServerAPI(w Server, client *rpc.Client) *PublicServerAPI

NewPublicServerAPI creates a new PublicServerAPI with the provided underlying Server process

func (*PublicServerAPI) Stream

func (api *PublicServerAPI) Stream(ctx context.Context, params eth.SubscriptionSettings) (*rpc.Subscription, error)

Stream is the public method to setup a subscription that fires off IPLD payloads as they are processed

func (*PublicServerAPI) WatchAddress

func (api *PublicServerAPI) WatchAddress(operation types.OperationType, args []types.WatchAddressArg) error

WatchAddress makes a geth WatchAddress API call with the given operation and args

type Server

type Server interface {
	// Start() and Stop()
	ethnode.Lifecycle
	APIs() []rpc.API
	Protocols() []p2p.Protocol
	// Pub-Sub handling event loop
	Serve(wg *sync.WaitGroup, screenAndServePayload <-chan eth.ConvertedPayload)
	// Method to subscribe to the service
	Subscribe(id rpc.ID, sub chan<- SubscriptionPayload, quitChan chan<- bool, params eth.SubscriptionSettings)
	// Method to unsubscribe from the service
	Unsubscribe(id rpc.ID)
	// Backend exposes the server's backend
	Backend() *eth.Backend
}

Server is the top level interface for streaming, converting to IPLDs, publishing, and indexing all chain data; screening this data; and serving it up to subscribed clients This service is compatible with the Ethereum service interface (node.Service)

func NewServer

func NewServer(settings *Config) (Server, error)

NewServer creates a new Server using an underlying Service struct

type Service

type Service struct {
	// Used to sync access to the Subscriptions
	sync.Mutex
	// Interface for filtering and serving data according to subscribed clients according to their specification
	Filterer eth.Filterer
	// Interface for fetching IPLD objects from IPFS
	IPLDFetcher eth.Fetcher
	// Interface for searching and retrieving CIDs from Postgres index
	Retriever eth.Retriever
	// Used to signal shutdown of the service
	QuitChan chan bool
	// A mapping of rpc.IDs to their subscription channels, mapped to their subscription type (hash of the StreamFilters)
	Subscriptions map[common.Hash]map[rpc.ID]Subscription
	// A mapping of subscription params hash to the corresponding subscription params
	SubscriptionTypes map[common.Hash]eth.SubscriptionSettings
	// contains filtered or unexported fields
}

Service is the underlying struct for the watcher

func (*Service) APIs

func (sap *Service) APIs() []rpc.API

APIs returns the RPC descriptors the watcher service offers

func (*Service) Backend

func (sap *Service) Backend() *eth.Backend

Backend exposes the server's backend

func (*Service) Protocols

func (sap *Service) Protocols() []p2p.Protocol

Protocols exports the services p2p protocols, this service has none

func (*Service) Serve

func (sap *Service) Serve(wg *sync.WaitGroup, screenAndServePayload <-chan eth.ConvertedPayload)

Serve listens for incoming converter data off the screenAndServePayload from the Sync process It filters and sends this data to any subscribers to the service This process can also be stood up alone, without an screenAndServePayload attached to a Sync process and it will hang on the WaitGroup indefinitely, allowing the Service to serve historical data requests only

func (*Service) Start

func (sap *Service) Start() error

Start is used to begin the service This is mostly just to satisfy the node.Service interface

func (*Service) Stop

func (sap *Service) Stop() error

Stop is used to close down the service This is mostly just to satisfy the node.Service interface

func (*Service) Subscribe

func (sap *Service) Subscribe(id rpc.ID, sub chan<- SubscriptionPayload, quitChan chan<- bool, params eth.SubscriptionSettings)

Subscribe is used by the API to remotely subscribe to the service loop The params must be rlp serializable and satisfy the SubscriptionSettings() interface

func (*Service) Unsubscribe

func (sap *Service) Unsubscribe(id rpc.ID)

Unsubscribe is used by the API to remotely unsubscribe to the StateDiffingService loop

type Subscription

type Subscription struct {
	ID          rpc.ID
	PayloadChan chan<- SubscriptionPayload
	QuitChan    chan<- bool
}

Subscription holds the information for an individual client subscription to the watcher

type SubscriptionPayload

type SubscriptionPayload struct {
	Data   []byte `json:"data"` // e.g. for Ethereum rlp serialized eth.StreamPayload
	Height int64  `json:"height"`
	Err    string `json:"err"`  // field for error
	Flag   Flag   `json:"flag"` // field for message
}

SubscriptionPayload is the struct for a watcher data subscription payload It carries data of a type specific to the chain being supported/queried and an error message

func (SubscriptionPayload) BackFillComplete

func (sp SubscriptionPayload) BackFillComplete() bool

func (SubscriptionPayload) Error

func (sp SubscriptionPayload) Error() error