consumer

Apr 22, 2024

Changes in this version

+ const MAX_ETCD_ERR_LEN

+ var ErrDeferToNextTransaction = fmt.Errorf("consumer application deferred message")

+ func PickFirstHints(hints *pc.GetHintsResponse, log pb.Journal) recoverylog.FSMHints

+ func ShardUnassign(ctx context.Context, srv *Service, req *pc.UnassignRequest) (*pc.UnassignResponse, error)

type Service

+ func (svc *Service) Unassign(ctx context.Context, req *pc.UnassignRequest) (*pc.UnassignResponse, error)

type Shard

+ PrimaryLoop func() client.OpFuture

Sep 2, 2021

Changes in this version

+ func ShardApply(ctx context.Context, srv *Service, req *pc.ApplyRequest) (*pc.ApplyResponse, error)

+ func ShardGetHints(ctx context.Context, srv *Service, req *pc.GetHintsRequest) (*pc.GetHintsResponse, error)

+ func ShardList(ctx context.Context, srv *Service, req *pc.ListRequest) (*pc.ListResponse, error)

+ func ShardStat(ctx context.Context, srv *Service, req *pc.StatRequest) (*pc.StatResponse, error)

+ type EnvelopeOrError struct

+ Error error

+ type MessageProducer interface

+ ReadThrough func(Shard, Store, ResolveArgs) (pb.Offsets, error)

+ ReplayRange func(_ Shard, _ Store, journal pb.Journal, begin, end pb.Offset) message.Iterator

+ StartReadingMessages func(_ Shard, _ Store, _ pc.Checkpoint, intoCh chan<- EnvelopeOrError)

type Resolver

+ func (r *Resolver) Collect(ch chan<- prometheus.Metric)

+ func (r *Resolver) Describe(ch chan<- *prometheus.Desc)

type Service

+ PublishClockDelta time.Duration

+ ShardAPI struct{ ... }

type Shard

+ RecoveredHints func() *pc.GetHintsResponse

Apr 24, 2020

Changes in this version

+ func ShardIsConsistent(shard allocator.Item, assignment keyspace.KeyValue, all keyspace.KeyValues) bool

Oct 30, 2019

Changes in this version

+ func VerifyReferencedJournals(ctx context.Context, jc pb.JournalClient, req *pc.ApplyRequest) error

type Resolver

+ func (r *Resolver) ShardsWithSource(journal pb.Journal) []*pc.ShardSpec

Oct 4, 2019

Changes in this version

type BeginFinisher

+ FinishedTxn func(Shard, Store, OpFuture)

type JSONFileStore

+ func (s *JSONFileStore) RestoreCheckpoint(Shard) (pc.Checkpoint, error)

+ func (s *JSONFileStore) StartCommit(_ Shard, cp pc.Checkpoint, waitFor OpFutures) OpFuture

+ type OpFuture = client.OpFuture

+ type OpFutures = client.OpFutures

type ResolveArgs

+ ReadThrough pb.Offsets

+ type SQLStore struct

+ Cache interface{}

+ DB *sql.DB

+ func NewSQLStore(db *sql.DB) *SQLStore

+ func (s *SQLStore) Destroy()

+ func (s *SQLStore) RestoreCheckpoint(shard Shard) (cp pc.Checkpoint, _ error)

+ func (s *SQLStore) StartCommit(shard Shard, cp pc.Checkpoint, waitFor OpFutures) OpFuture

+ func (s *SQLStore) Transaction(ctx context.Context, txOpts *sql.TxOptions) (_ *sql.Tx, err error)

type Service

+ App Application

type Shard

+ FQN func() string

+ Progress func() (readThrough, publishAt pb.Offsets)

type Store

+ RestoreCheckpoint func(Shard) (pc.Checkpoint, error)

+ StartCommit func(_ Shard, _ pc.Checkpoint, waitFor OpFutures) OpFuture

Sep 10, 2019

Changes in this version

+ var ErrResolverStopped = errors.New("resolver has stopped serving local replicas")

+ func ApplyShards(ctx context.Context, sc pc.ShardClient, req *pc.ApplyRequest) (*pc.ApplyResponse, error)

+ func ApplyShardsInBatches(ctx context.Context, sc pc.ShardClient, req *pc.ApplyRequest, size int) (*pc.ApplyResponse, error)

+ func FetchHints(ctx context.Context, sc pc.ShardClient, req *pc.GetHintsRequest) (*pc.GetHintsResponse, error)

+ func ListShards(ctx context.Context, sc pc.ShardClient, req *pc.ListRequest) (*pc.ListResponse, error)

+ func NewKeySpace(prefix string) *keyspace.KeySpace

+ func StatShard(ctx context.Context, rc pc.RoutedShardClient, req *pc.StatRequest) (*pc.StatResponse, error)

+ type Application interface

+ ConsumeMessage func(Shard, Store, message.Envelope) error

+ FinalizeTxn func(shard Shard, store Store) error

+ NewMessage func(*pb.JournalSpec) (message.Message, error)

+ NewStore func(shard Shard, dir string, rec *recoverylog.Recorder) (Store, error)

+ type BeginFinisher interface

+ BeginTxn func(Shard, Store) error

+ FinishTxn func(Shard, Store, error) error

+ type JSONFileStore struct

+ State interface{}

+ func NewJSONFileStore(rec *recoverylog.Recorder, dir string, state interface{}) (*JSONFileStore, error)

+ func (s *JSONFileStore) Destroy()

+ func (s *JSONFileStore) FetchJournalOffsets() (map[pb.Journal]int64, error)

+ func (s *JSONFileStore) Flush(offsets map[pb.Journal]int64) error

+ func (s *JSONFileStore) Recorder() *recoverylog.Recorder

+ type Replica struct

+ func NewReplica(app Application, ks *keyspace.KeySpace, etcd *clientv3.Client, ...) *Replica

+ func (r *Replica) Assignment() keyspace.KeyValue

+ func (r *Replica) Context() context.Context

+ func (r *Replica) JournalClient() client.AsyncJournalClient

+ func (r *Replica) Spec() *pc.ShardSpec

+ type Resolution struct

+ Done func()

+ Header pb.Header

+ Shard Shard

+ Spec *pc.ShardSpec

+ Status pc.Status

+ Store Store

+ type ResolveArgs struct

+ Context context.Context

+ MayProxy bool

+ ProxyHeader *pb.Header

+ ShardID pc.ShardID

+ type Resolver struct

+ func NewResolver(state *allocator.State, newReplica func() *Replica) *Resolver

+ func (r *Resolver) Resolve(args ResolveArgs) (res Resolution, err error)

+ type Service struct

+ Etcd *clientv3.Client

+ Journals pb.RoutedJournalClient

+ Loopback *grpc.ClientConn

+ Resolver *Resolver

+ State *allocator.State

+ func NewService(app Application, state *allocator.State, rjc pb.RoutedJournalClient, ...) *Service

+ func (srv *Service) Apply(ctx context.Context, req *pc.ApplyRequest) (*pc.ApplyResponse, error)

+ func (srv *Service) GetHints(ctx context.Context, req *pc.GetHintsRequest) (*pc.GetHintsResponse, error)

+ func (srv *Service) List(ctx context.Context, req *pc.ListRequest) (*pc.ListResponse, error)

+ func (srv *Service) Stat(ctx context.Context, req *pc.StatRequest) (*pc.StatResponse, error)

+ func (svc *Service) QueueTasks(tasks *task.Group, server *server.Server)

+ func (svc *Service) Stopping() <-chan struct{}

+ type Shard interface

+ Assignment func() keyspace.KeyValue

+ Context func() context.Context

+ JournalClient func() client.AsyncJournalClient

+ Spec func() *pc.ShardSpec

+ type Store interface

+ Destroy func()

+ FetchJournalOffsets func() (map[pb.Journal]int64, error)

+ Flush func(offsets map[pb.Journal]int64) error

+ Recorder func() *recoverylog.Recorder