history

Documentation

Index

• type Emitter
  • func NewEmitter[T any]() *Emitter[T]
  • func (s *Emitter[T]) Emit(value T)
  • func (s *Emitter[T]) Subscribe() <-chan T
• type HistoryProcessor
• type HistoryRetriever
  • func NewHistoryRetriever(store common.StorenodeRequestor, historyProcessor HistoryProcessor, ...) *HistoryRetriever
  • func (hr *HistoryRetriever) Query(ctx context.Context, criteria store.FilterCriteria, storenodeID peer.ID, ...) error
• type OneShotEmitter
  • func NewOneshotEmitter[T any]() *OneShotEmitter[T]
  • func (s *OneShotEmitter[T]) Emit(value T)
• type SortedStorenode
• type StorenodeConfigProvider
• type StorenodeCycle
  • func NewStorenodeCycle(logger *zap.Logger, pinger common.Pinger) *StorenodeCycle
  • func (m *StorenodeCycle) Cycle(ctx context.Context)
  • func (m *StorenodeCycle) DisconnectActiveStorenode(backoff time.Duration)
  • func (m *StorenodeCycle) GetActiveStorenode() peer.ID
  • func (m *StorenodeCycle) IsStorenodeAvailable(peerID peer.ID) bool
  • func (m *StorenodeCycle) PerformStorenodeTask(fn func() error, options ...StorenodeTaskOption) error
  • func (m *StorenodeCycle) SetStorenodeConfigProvider(provider StorenodeConfigProvider)
  • func (m *StorenodeCycle) Start(ctx context.Context)
  • func (m *StorenodeCycle) WaitForAvailableStoreNode(ctx context.Context) bool
• type StorenodeTaskOption
  • func WithPeerID(peerID peer.ID) StorenodeTaskOption

Types

type Emitter

type Emitter[T any] struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewEmitter

func NewEmitter[T any]() *Emitter[T]

func (*Emitter[T]) Emit

func (s *Emitter[T]) Emit(value T)

func (*Emitter[T]) Subscribe

func (s *Emitter[T]) Subscribe() <-chan T

type HistoryProcessor

type HistoryProcessor interface {
	OnEnvelope(env *protocol.Envelope, processEnvelopes bool) error
	OnRequestFailed(requestID []byte, peerID peer.ID, err error)
}

type HistoryRetriever

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

func NewHistoryRetriever

func NewHistoryRetriever(store common.StorenodeRequestor, historyProcessor HistoryProcessor, logger *zap.Logger) *HistoryRetriever

func (*HistoryRetriever) Query

func (hr *HistoryRetriever) Query(
	ctx context.Context,
	criteria store.FilterCriteria,
	storenodeID peer.ID,
	pageLimit uint64,
	shouldProcessNextPage func(int) (bool, uint64),
	processEnvelopes bool,
) error

type OneShotEmitter

type OneShotEmitter[T any] struct {
	Emitter[T]
}

func NewOneshotEmitter

func NewOneshotEmitter[T any]() *OneShotEmitter[T]

func (*OneShotEmitter[T]) Emit

func (s *OneShotEmitter[T]) Emit(value T)

type SortedStorenode

type SortedStorenode struct {
	Storenode       peer.AddrInfo
	RTT             time.Duration
	CanConnectAfter time.Time
}

type StorenodeConfigProvider

type StorenodeConfigProvider interface {
	UseStorenodes() (bool, error)
	GetPinnedStorenode() (peer.AddrInfo, error)
	Storenodes() ([]peer.AddrInfo, error)
}

type StorenodeCycle

type StorenodeCycle struct {
	sync.RWMutex

	StorenodeAvailableOneshotEmitter *OneShotEmitter[struct{}]
	StorenodeChangedEmitter          *Emitter[peer.ID]
	StorenodeNotWorkingEmitter       *Emitter[struct{}]
	StorenodeAvailableEmitter        *Emitter[peer.ID]
	// contains filtered or unexported fields
}

func NewStorenodeCycle

func NewStorenodeCycle(logger *zap.Logger, pinger common.Pinger) *StorenodeCycle

func (*StorenodeCycle) Cycle

func (m *StorenodeCycle) Cycle(ctx context.Context)

func (*StorenodeCycle) DisconnectActiveStorenode

func (m *StorenodeCycle) DisconnectActiveStorenode(backoff time.Duration)

func (*StorenodeCycle) GetActiveStorenode

func (m *StorenodeCycle) GetActiveStorenode() peer.ID

func (*StorenodeCycle) IsStorenodeAvailable

func (m *StorenodeCycle) IsStorenodeAvailable(peerID peer.ID) bool

func (*StorenodeCycle) PerformStorenodeTask

func (m *StorenodeCycle) PerformStorenodeTask(fn func() error, options ...StorenodeTaskOption) error

func (*StorenodeCycle) SetStorenodeConfigProvider

func (m *StorenodeCycle) SetStorenodeConfigProvider(provider StorenodeConfigProvider)

func (*StorenodeCycle) Start

func (m *StorenodeCycle) Start(ctx context.Context)

func (*StorenodeCycle) WaitForAvailableStoreNode

func (m *StorenodeCycle) WaitForAvailableStoreNode(ctx context.Context) bool

type StorenodeTaskOption

type StorenodeTaskOption func(*storenodeTaskParameters)

func WithPeerID

func WithPeerID(peerID peer.ID) StorenodeTaskOption