consumer

Documentation

Index

• Variables
• type Consumer
  • func NewConsumer(ctx context.Context, socketURL string, progPath string, ...) (*Consumer, error)
  • func (c *Consumer) HandleRepoCommit(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) error
  • func (c *Consumer) HandleStreamEvent(ctx context.Context, xe *events.XRPCStreamEvent) error
  • func (c *Consumer) ReadCursor(ctx context.Context) error
  • func (c *Consumer) WriteCursor(ctx context.Context) error
• type Event
• type Progress
  • func (p *Progress) Get() (int64, time.Time)
  • func (p *Progress) Update(seq int64, processedAt time.Time)
• type URI
  • func GetURI(uri string) (*URI, error)

Variables

var (
	EvtCreateRecord = "c"
	EvtUpdateRecord = "u"
	EvtDeleteRecord = "d"
)

Types

type Consumer

type Consumer struct {
	SocketURL string
	Progress  *Progress
	Emit      func(context.Context, Event) error
}

Consumer is the consumer of the firehose

func NewConsumer

func NewConsumer(
	ctx context.Context,
	socketURL string,
	progPath string,
	emit func(context.Context, Event) error,
) (*Consumer, error)

NewConsumer creates a new consumer

func (*Consumer) HandleRepoCommit

func (c *Consumer) HandleRepoCommit(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) error

HandleRepoCommit handles a repo commit event from the firehose and processes the records

func (*Consumer) HandleStreamEvent

func (c *Consumer) HandleStreamEvent(ctx context.Context, xe *events.XRPCStreamEvent) error

HandleStreamEvent handles a stream event from the firehose

func (*Consumer) ReadCursor

func (c *Consumer) ReadCursor(ctx context.Context) error

ReadCursor reads the cursor from file

func (*Consumer) WriteCursor

func (c *Consumer) WriteCursor(ctx context.Context) error

WriteCursor writes the cursor to file

type Event

type Event struct {
	Did        string `json:"did"`
	Seq        int64  `json:"seq"`
	OpType     string `json:"opType"`
	Collection string `json:"collection,omitempty"`
	RKey       string `json:"rkey,omitempty"`

	Record any `json:"record,omitempty"`
}

type Progress

type Progress struct {
	LastSeq            int64     `json:"last_seq"`
	LastSeqProcessedAt time.Time `json:"last_seq_processed_at"`
	// contains filtered or unexported fields
}

Progress is the cursor for the consumer

func (*Progress) Get

func (p *Progress) Get() (int64, time.Time)

func (*Progress) Update

func (p *Progress) Update(seq int64, processedAt time.Time)

type URI

type URI struct {
	Did        string
	RKey       string
	Collection string
}

func GetURI

func GetURI(uri string) (*URI, error)

URI: at://{did}/{namespace}/{rkey}