ingest

Documentation

Index

• Variables
• func Unfreeze(unfrozen map[peer.ID]cid.Cid, dstore datastore.Datastore) error
• type Ingester
  • func NewIngester(cfg config.Ingest, h host.Host, idxr indexer.Interface, reg *registry.Registry, ...) (*Ingester, error)
  • func (ing *Ingester) Announce(ctx context.Context, nextCid cid.Cid, addrInfo peer.AddrInfo) error
  • func (ing *Ingester) BatchSize() int
  • func (ing *Ingester) Close() error
  • func (ing *Ingester) GetLatestSync(publisherID peer.ID) (cid.Cid, error)
  • func (ing *Ingester) RunWorkers(n int)
  • func (ing *Ingester) SetBatchSize(batchSize int)
  • func (ing *Ingester) SetRateLimit(cfgRateLimit config.RateLimit) error
  • func (ing *Ingester) Sync(ctx context.Context, peerID peer.ID, peerAddr multiaddr.Multiaddr, depth int, ...) (cid.Cid, error)
• type Option
  • func WithAdsDatastore(ds datastore.Batching) Option
  • func WithIndexCounts(ic *counter.IndexCounts) Option
• type Queue
  • func NewPriorityQueue() *Queue
  • func (q *Queue) Close()
  • func (q *Queue) Has(p providerID) bool
  • func (q *Queue) Length() int
  • func (q *Queue) Pop() providerID
  • func (q *Queue) PopChan() chan providerID
  • func (q *Queue) Push(p providerID) uint32

Variables

var Selectors struct {
	// One selects a single node in a DAG.
	One ipld.Node
	// All selects all the nodes in a DAG recursively traversing all the edges with no recursion
	// limit.
	All ipld.Node
	// AdSequence selects an schema.Advertisement sequence with recursive edge exploration of
	// PreviousID field.
	AdSequence ipld.Node
	// EntriesWithLimit selects schema.EntryChunk nodes  recursively traversing the edge at Next
	// filed with the given recursion limit.
	EntriesWithLimit func(limit selector.RecursionLimit) ipld.Node
}

Selectors captures a collection of IPLD selectors commonly used by the Ingester.

Functions

func Unfreeze

func Unfreeze(unfrozen map[peer.ID]cid.Cid, dstore datastore.Datastore) error

Types

type Ingester

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

Ingester is a type that uses dagsync for the ingestion protocol.

func NewIngester

func NewIngester(cfg config.Ingest, h host.Host, idxr indexer.Interface, reg *registry.Registry, ds datastore.Batching, options ...Option) (*Ingester, error)

NewIngester creates a new Ingester that uses a dagsync Subscriber to handle communication with providers.

func (*Ingester) Announce

func (ing *Ingester) Announce(ctx context.Context, nextCid cid.Cid, addrInfo peer.AddrInfo) error

Announce send an announce message to directly to dagsync, instead of through pubsub.

func (*Ingester) BatchSize

func (ing *Ingester) BatchSize() int

func (*Ingester) Close

func (ing *Ingester) Close() error

func (*Ingester) GetLatestSync

func (ing *Ingester) GetLatestSync(publisherID peer.ID) (cid.Cid, error)

Get the latest CID synced for the peer.

func (*Ingester) RunWorkers

func (ing *Ingester) RunWorkers(n int)

func (*Ingester) SetBatchSize

func (ing *Ingester) SetBatchSize(batchSize int)

func (*Ingester) SetRateLimit

func (ing *Ingester) SetRateLimit(cfgRateLimit config.RateLimit) error

func (*Ingester) Sync

func (ing *Ingester) Sync(ctx context.Context, peerID peer.ID, peerAddr multiaddr.Multiaddr, depth int, resync bool) (cid.Cid, error)

Sync syncs advertisements, up to the the latest advertisement, from a publisher. This channel returns the final CID that was synced by the call to Sync.

Sync works by first fetching each advertisement from the specified peer starting at the most recent and traversing to the advertisement last seen by the indexer, or until the advertisement depth limit is reached. Then the entries in each advertisement are synced and the multihashes in each entry are indexed.

The selector used to sync the advertisement is controlled by the following parameters: depth, and resync.

The depth argument specifies the recursion depth limit to use during sync. Its value may less than -1 for no limit, 0 to use the indexer's configured value, or greater than 1 for an explicit limit.

The resync argument specifies whether to stop the traversal at the latest known advertisement that is already synced. If set to true, the traversal will continue until either there are no more advertisements left or the recursion depth limit is reached.

The reference to the latest synced advertisement returned by GetLatestSync is only updated if the given depth is zero and resync is set to false. Otherwise, a custom selector with the given depth limit and stop link is constructed and used for traversal. See dagsync.Subscriber.Sync.

The Context argument controls the lifetime of the sync. Canceling it cancels the sync and causes the multihash channel to close without any data.

type Option

type Option func(*configIngest) error

Option is a function that sets a value in a config.

func WithAdsDatastore

func WithAdsDatastore(ds datastore.Batching) Option

WithAdsDatastore configures a separate datastore for advertizements.

func WithIndexCounts

func WithIndexCounts(ic *counter.IndexCounts) Option

WithIndexCounts configures counting indexes using an IndexCounts instance.

type Queue

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

func NewPriorityQueue

func NewPriorityQueue() *Queue

func (*Queue) Close

func (q *Queue) Close()

func (*Queue) Has

func (q *Queue) Has(p providerID) bool

func (*Queue) Length

func (q *Queue) Length() int

func (*Queue) Pop

func (q *Queue) Pop() providerID

func (*Queue) PopChan

func (q *Queue) PopChan() chan providerID

Returns a channel yielding the next provider to be pulled with 'at most once' semantics before the channel is closed.

func (*Queue) Push

func (q *Queue) Push(p providerID) uint32

Push a provider into the set. returns the number of pushes this provider has had since last popped.