serve

Documentation

Index

• Constants
• type Config
  • func NewConfig() (*Config, error)
• type Flag
• type InfoAPI
  • func NewInfoAPI() *InfoAPI
  • func (iapi *InfoAPI) Modules() map[string]string
  • func (iapi *InfoAPI) NodeInfo() *p2p.NodeInfo
  • func (iapi *InfoAPI) Version() string
• type PublicServerAPI
  • func NewPublicServerAPI(w Server) *PublicServerAPI
  • func (api *PublicServerAPI) Chain() shared.ChainType
  • func (api *PublicServerAPI) Stream(ctx context.Context, params eth.SubscriptionSettings) (*rpc.Subscription, 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 eth2.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"
)

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               *postgres.DB
	DBConfig         postgres.Config
	WSEndpoint       string
	HTTPEndpoint     string
	IPCEndpoint      string
	ChainConfig      *params.ChainConfig
	DefaultSender    *common.Address
	RPCGasCap        *big.Int
	Client           *rpc.Client
	SupportStateDiff bool
}

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 InfoAPI

type InfoAPI struct{}

Struct for holding watcher meta data

func NewInfoAPI

func NewInfoAPI() *InfoAPI

NewInfoAPI creates a new InfoAPI

func (*InfoAPI) Modules

func (iapi *InfoAPI) Modules() map[string]string

Modules returns modules supported by this api

func (*InfoAPI) NodeInfo

func (iapi *InfoAPI) NodeInfo() *p2p.NodeInfo

NodeInfo gathers and returns a collection of metadata for the watcher

func (*InfoAPI) Version

func (iapi *InfoAPI) Version() string

Version returns the version of the watcher

type PublicServerAPI

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

PublicServerAPI is the public api for the watcher

func NewPublicServerAPI

func NewPublicServerAPI(w Server) *PublicServerAPI

NewPublicServerAPI creates a new PublicServerAPI with the provided underlying Server process

func (*PublicServerAPI) Chain

func (api *PublicServerAPI) Chain() shared.ChainType

Chain returns the chain type that this watcher instance supports

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

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 eth2.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 eth2.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