store

package
v0.8.1 Latest Latest
Warning

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

 Go to latest Published: Oct 7, 2023 License: MIT Imports: 23 Imported by: 5

Documentation

Index

Constants

View Source 
const MaxContentFilters = 10

MaxContentFilters is the maximum number of allowed content filters in a query

View Source 
const MaxPageSize = 100

MaxPageSize is the maximum number of waku messages to return per page

View Source 
const StoreID_v20beta4 = libp2pProtocol.ID("/vac/waku/store/2.0.0-beta4")

StoreID_v20beta4 is the current Waku Store protocol identifier

Variables

View Source 
var (
	// ErrMaxContentFilters is returned when the number of content topics in the query
	// exceeds the limit
	ErrMaxContentFilters = errors.New("exceeds the maximum number of content filters allowed")

	// ErrNoPeersAvailable is returned when there are no store peers in the peer store
	// that could be used to retrieve message history
	ErrNoPeersAvailable = errors.New("no suitable remote peers")

	// ErrInvalidID is returned when no RequestID is given
	ErrInvalidID = errors.New("invalid request id")

	// ErrFailedToResumeHistory is returned when the node attempted to retrieve historic
	// messages to fill its own message history but for some reason it failed
	ErrFailedToResumeHistory = errors.New("failed to resume the history")

	// ErrFailedQuery is emitted when the query fails to return results
	ErrFailedQuery = errors.New("failed to resolve the query")
)

Functions

This section is empty.

Types

type HistoryRequestOption 

type HistoryRequestOption func(*HistoryRequestParameters)

func DefaultOptions 

func DefaultOptions() []HistoryRequestOption

Default options to be used when querying a store node for results

func WithAutomaticPeerSelection 

func WithAutomaticPeerSelection(fromThesePeers ...peer.ID) HistoryRequestOption

WithAutomaticPeerSelection is an option used to randomly select a peer from the peer store to request the message history. If a list of specific peers is passed, the peer will be chosen from that list assuming it supports the chosen protocol, otherwise it will chose a peer from the node peerstore

func WithAutomaticRequestID added in v0.8.0 

func WithAutomaticRequestID() HistoryRequestOption

WithAutomaticRequestID is an option to automatically generate a request ID when creating a store request

func WithCursor 

func WithCursor(c *pb.Index) HistoryRequestOption

func WithFastestPeerSelection 

func WithFastestPeerSelection(ctx context.Context, fromThesePeers ...peer.ID) HistoryRequestOption

WithFastestPeerSelection is an option used to select a peer from the peer store with the lowest ping. If a list of specific peers is passed, the peer will be chosen from that list assuming it supports the chosen protocol, otherwise it will chose a peer from the node peerstore

func WithLocalQuery added in v0.5.0 

func WithLocalQuery() HistoryRequestOption

func WithPaging 

func WithPaging(asc bool, pageSize uint64) HistoryRequestOption

WithPaging is an option used to specify the order and maximum number of records to return

func WithPeer 

func WithPeer(p peer.ID) HistoryRequestOption

WithPeer is an option used to specify the peerID to request the message history

func WithRequestID added in v0.8.0 

func WithRequestID(requestID []byte) HistoryRequestOption

WithRequestID is an option to set a specific request ID to be used when creating a store request

type HistoryRequestParameters 

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

type MessageProvider 

type MessageProvider interface {
	GetAll() ([]persistence.StoredMessage, error)
	Query(query *pb.HistoryQuery) (*pb.Index, []persistence.StoredMessage, error)
	Validate(env *protocol.Envelope) error
	Put(env *protocol.Envelope) error
	MostRecentTimestamp() (int64, error)
	Start(ctx context.Context, timesource timesource.Timesource) error
	Stop()
	Count() (int, error)
}

type Metrics added in v0.8.0 

type Metrics interface {
	RecordQuery()
	RecordError(err metricsErrCategory)
}

Metrics exposes the functions required to update prometheus metrics for store protocol

type Query 

type Query struct {
	Topic         string
	ContentTopics []string
	StartTime     int64
	EndTime       int64
}

type Result 

type Result struct {
	Messages []*wpb.WakuMessage
	// contains filtered or unexported fields
}

Result represents a valid response from a store node

func (*Result) Cursor 

func (r *Result) Cursor() *pb.Index

func (*Result) GetMessages added in v0.4.0 

func (r *Result) GetMessages() []*wpb.WakuMessage

func (*Result) IsComplete 

func (r *Result) IsComplete() bool

func (*Result) Next added in v0.4.0 

func (r *Result) Next(ctx context.Context) (bool, error)

func (*Result) PeerID 

func (r *Result) PeerID() peer.ID

func (*Result) Query 

func (r *Result) Query() *pb.HistoryQuery

type Store 

type Store interface {
	SetHost(h host.Host)
	Start(context.Context, relay.Subscription) error
	Query(ctx context.Context, query Query, opts ...HistoryRequestOption) (*Result, error)
	Find(ctx context.Context, query Query, cb criteriaFN, opts ...HistoryRequestOption) (*wpb.WakuMessage, error)
	Next(ctx context.Context, r *Result) (*Result, error)
	Resume(ctx context.Context, pubsubTopic string, peerList []peer.ID) (int, error)
	Stop()
}

type WakuStore 

type WakuStore struct {
	MsgC relay.Subscription
	// contains filtered or unexported fields
}

func NewWakuStore 

func NewWakuStore(p MessageProvider, pm *peermanager.PeerManager, timesource timesource.Timesource, reg prometheus.Registerer, log *zap.Logger) *WakuStore

NewWakuStore creates a WakuStore using an specific MessageProvider for storing the messages Takes an optional peermanager if WakuStore is being created along with WakuNode. If using libp2p host, then pass peermanager as nil

func (*WakuStore) Find 

func (store *WakuStore) Find(ctx context.Context, query Query, cb criteriaFN, opts ...HistoryRequestOption) (*wpb.WakuMessage, error)

Find the first message that matches a criteria. criteriaCB is a function that will be invoked for each message and returns true if the message matches the criteria

func (*WakuStore) FindMessages 

func (store *WakuStore) FindMessages(query *pb.HistoryQuery) *pb.HistoryResponse

func (*WakuStore) Next 

func (store *WakuStore) Next(ctx context.Context, r *Result) (*Result, error)

Next is used with to retrieve the next page of rows from a query response. If no more records are found, the result will not contain any messages. This function is useful for iterating over results without having to manually specify the cursor and pagination order and max number of results

func (*WakuStore) Query 

func (store *WakuStore) Query(ctx context.Context, query Query, opts ...HistoryRequestOption) (*Result, error)

func (*WakuStore) Resume 

func (store *WakuStore) Resume(ctx context.Context, pubsubTopic string, peerList []peer.ID) (int, error)

Resume retrieves the history of waku messages published on the default waku pubsub topic since the last time the waku store node has been online messages are stored in the store node's messages field and in the message db the offline time window is measured as the difference between the current time and the timestamp of the most recent persisted waku message an offset of 20 second is added to the time window to count for nodes asynchrony the history is fetched from one of the peers persisted in the waku store node's peer manager unit peerList indicates the list of peers to query from. The history is fetched from the first available peer in this list. Such candidates should be found through a discovery method (to be developed). if no peerList is passed, one of the peers in the underlying peer manager unit of the store protocol is picked randomly to fetch the history from. The history gets fetched successfully if the dialed peer has been online during the queried time window. the resume proc returns the number of retrieved messages if no error occurs, otherwise returns the error string

func (*WakuStore) SetHost added in v0.6.0 

func (store *WakuStore) SetHost(h host.Host)

Sets the host to be able to mount or consume a protocol

func (*WakuStore) SetMessageProvider 

func (store *WakuStore) SetMessageProvider(p MessageProvider)

SetMessageProvider allows switching the message provider used with a WakuStore

func (*WakuStore) Start 

func (store *WakuStore) Start(ctx context.Context, sub relay.Subscription) error

Start initializes the WakuStore by enabling the protocol and fetching records from a message provider

func (*WakuStore) Stop 

func (store *WakuStore) Stop()

Stop closes the store message channel and removes the protocol stream handler

type WakuSwap added in v0.4.0 

type WakuSwap interface {
}

Source Files

View all Source files

Directories

Path Synopsis

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.