store

Documentation

Index

• Constants
• Variables
• type HistoryRequestOption
  • func DefaultOptions() []HistoryRequestOption
  • func WithAutomaticPeerSelection() HistoryRequestOption
  • func WithAutomaticRequestId() HistoryRequestOption
  • func WithCursor(c *pb.Index) HistoryRequestOption
  • func WithFastestPeerSelection(ctx context.Context) HistoryRequestOption
  • func WithPaging(asc bool, pageSize uint64) HistoryRequestOption
  • func WithPeer(p peer.ID) HistoryRequestOption
  • func WithRequestId(requestId []byte) HistoryRequestOption
• type HistoryRequestParameters
• type MessageProvider
• type Query
• type Result
  • func (r *Result) Cursor() *pb.Index
  • func (r *Result) PeerID() peer.ID
  • func (r *Result) Query() *pb.HistoryQuery
• type Store
• type WakuStore
  • func NewWakuStore(host host.Host, swap *swap.WakuSwap, p MessageProvider, log *zap.Logger) *WakuStore
  • func (store *WakuStore) FindMessages(query *pb.HistoryQuery) *pb.HistoryResponse
  • func (store *WakuStore) MessageChannel() chan *protocol.Envelope
  • func (store *WakuStore) Next(ctx context.Context, r *Result) (*Result, error)
  • func (store *WakuStore) Query(ctx context.Context, query Query, opts ...HistoryRequestOption) (*Result, error)
  • func (store *WakuStore) Resume(ctx context.Context, pubsubTopic string, peerList []peer.ID) (int, error)
  • func (store *WakuStore) SetMessageProvider(p MessageProvider)
  • func (store *WakuStore) Start(ctx context.Context)
  • func (store *WakuStore) Stop()

Constants

const MaxPageSize = 100

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

const MaxTimeVariance = time.Duration(20) * time.Second

MaxTimeVariance is the maximum duration in the future allowed for a message timestamp

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

StoreID_v20beta4 is the current Waku Store protocol identifier

Variables

var (
	// 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")

	ErrFutureMessage = errors.New("message timestamp in the future")
)

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() HistoryRequestOption

WithAutomaticPeerSelection is an option used to randomly select a peer from the peer store to request the message history

func WithAutomaticRequestId

func WithAutomaticRequestId() HistoryRequestOption

func WithCursor

func WithCursor(c *pb.Index) HistoryRequestOption

func WithFastestPeerSelection

func WithFastestPeerSelection(ctx context.Context) 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

func WithRequestId(requestId []byte) HistoryRequestOption

type HistoryRequestParameters

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

type MessageProvider

type MessageProvider interface {
	GetAll() ([]persistence.StoredMessage, error)
	Query(query *pb.HistoryQuery) ([]persistence.StoredMessage, error)
	Put(env *protocol.Envelope) error
	MostRecentTimestamp() (int64, error)
	Stop()
	Count() (int, error)
}

type Query

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

type Result

type Result struct {
	Messages []*pb.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) PeerID

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

func (*Result) Query

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

type Store

type Store interface {
	Start(ctx context.Context)
	Query(ctx context.Context, query Query, opts ...HistoryRequestOption) (*Result, error)
	Next(ctx context.Context, r *Result) (*Result, error)
	Resume(ctx context.Context, pubsubTopic string, peerList []peer.ID) (int, error)
	MessageChannel() chan *protocol.Envelope
	Stop()
}

type WakuStore

type WakuStore struct {
	MsgC chan *protocol.Envelope
	// contains filtered or unexported fields
}

func NewWakuStore

func NewWakuStore(host host.Host, swap *swap.WakuSwap, p MessageProvider, log *zap.Logger) *WakuStore

NewWakuStore creates a WakuStore using an specific MessageProvider for storing the messages

func (*WakuStore) FindMessages

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

func (*WakuStore) MessageChannel

func (store *WakuStore) MessageChannel() chan *protocol.Envelope

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) 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)

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