managerstream

Documentation

Index

• Constants
• Variables
• func ExtractStringFromMapInterface(key string, m map[string]interface{}) (stringValue string, isString bool)
• type ClientConfig
• type IndexerClient
  • func NewIndexerClient(ctx context.Context, logger *zap.Logger, f flow.Flow, ...) *IndexerClient
  • func (ic *IndexerClient) CloseStream(ctx context.Context, streamID uuid.UUID) error
  • func (ic *IndexerClient) GetData(ctx context.Context, taskID uuid.UUID, heightRange structs.ExpandedHeightRange, ...)
  • func (ic *IndexerClient) GetLatestMark(ctx context.Context, taskID uuid.UUID, ldr structs.LatestDataRequest, ...)
  • func (ic *IndexerClient) Ingest(ctx context.Context, taskID uuid.UUID, heightRange structs.ExpandedHeightRange, ...)
  • func (ic *IndexerClient) LatestSequence(ctx context.Context, taskID uuid.UUID, ldr structs.LatestDataRequest, ...)
  • func (ic *IndexerClient) RegisterStream(ctx context.Context, stream *cStructs.StreamAccess) error
  • func (ic *IndexerClient) Run(ctx context.Context, stream *cStructs.StreamAccess)

Constants

const NumberOfWorkerProcesses = 20

Variables

var DefaultClientConfig = &ClientConfig{
	LatestMarkTimeout: time.Second * 5,
}

Functions

func ExtractStringFromMapInterface

func ExtractStringFromMapInterface(key string, m map[string]interface{}) (stringValue string, isString bool)

Types

type ClientConfig

type ClientConfig struct {
	LatestMarkTimeout time.Duration
}

type IndexerClient

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

func NewIndexerClient

func NewIndexerClient(
	ctx context.Context,
	logger *zap.Logger,
	f flow.Flow,
	inFlow flow.IngestionFlow,
	c *ClientConfig,
) *IndexerClient

func (*IndexerClient) CloseStream

func (ic *IndexerClient) CloseStream(ctx context.Context, streamID uuid.UUID) error

CloseStream removes stream from worker/client

func (*IndexerClient) GetData

func (ic *IndexerClient) GetData(
	ctx context.Context,
	taskID uuid.UUID,
	heightRange structs.ExpandedHeightRange,
	stream *cStructs.StreamAccess,
)

func (*IndexerClient) GetLatestMark

func (ic *IndexerClient) GetLatestMark(
	ctx context.Context,
	taskID uuid.UUID,
	ldr structs.LatestDataRequest,
	stream *cStructs.StreamAccess,
)

GetLatestMark gets latest block

func (*IndexerClient) Ingest

func (ic *IndexerClient) Ingest(
	ctx context.Context,
	taskID uuid.UUID,
	heightRange structs.ExpandedHeightRange,
	stream *cStructs.StreamAccess,
)

Ingest has a worker ingest data from a specified data source and leaves it up to the worker implementation to specify where transformed data gets persisted to.

func (*IndexerClient) LatestSequence

func (ic *IndexerClient) LatestSequence(
	ctx context.Context,
	taskID uuid.UUID,
	ldr structs.LatestDataRequest,
	stream *cStructs.StreamAccess,
)

func (*IndexerClient) RegisterStream

func (ic *IndexerClient) RegisterStream(ctx context.Context, stream *cStructs.StreamAccess) error

RegisterStream adds new listeners to the streams - currently fixed number per stream

func (*IndexerClient) Run

func (ic *IndexerClient) Run(ctx context.Context, stream *cStructs.StreamAccess)