stream

Documentation

Index

• Constants
• type Checkpoint
  • func NewCheckpoint(stream Stream, vbIds []uint16, client couchbase.Client, ...) Checkpoint
• type CheckpointMetric
• type LeaderElection
  • func NewLeaderElection(config *config.Dcp, serviceDiscovery servicediscovery.ServiceDiscovery, ...) LeaderElection
• type Metric
• type Stream
  • func NewStream(client couchbase.Client, metadata metadata.Metadata, config *config.Dcp, ...) Stream
• type VBucketDiscovery
  • func NewVBucketDiscovery(client couchbase.Client, config *config.Dcp, vBucketNumber int, ...) VBucketDiscovery
• type VBucketDiscoveryMetric

Constants

const (
	CheckpointTypeAuto            = "auto"
	CheckpointAutoResetTypeLatest = "latest"
)

const (
	KubernetesLeaderElectionType = "kubernetes"
)

Types

type Checkpoint

type Checkpoint interface {
	Save()
	Load() (*wrapper.ConcurrentSwissMap[uint16, *models.Offset], *wrapper.ConcurrentSwissMap[uint16, bool], bool)
	Clear()
	StartSchedule()
	StopSchedule()
	GetMetric() *CheckpointMetric
}

func NewCheckpoint

func NewCheckpoint(
	stream Stream,
	vbIds []uint16,
	client couchbase.Client,
	metadata metadata.Metadata,
	config *config.Dcp,
) Checkpoint

type CheckpointMetric

type CheckpointMetric struct {
	OffsetWrite        int
	OffsetWriteLatency int64
}

type LeaderElection

type LeaderElection interface {
	Start()
	Stop()
}

func NewLeaderElection

func NewLeaderElection(
	config *config.Dcp,
	serviceDiscovery servicediscovery.ServiceDiscovery,
	bus EventBus.Bus,
) LeaderElection

type Metric

type Metric struct {
	ProcessLatency int64
	DcpLatency     int64
	Rebalance      int
}

type Stream

type Stream interface {
	Open()
	Rebalance()
	Save()
	Close(bool)
	GetOffsets() (*wrapper.ConcurrentSwissMap[uint16, *models.Offset], *wrapper.ConcurrentSwissMap[uint16, bool], bool)
	GetObserver() couchbase.Observer
	GetMetric() (*Metric, int)
	UnmarkDirtyOffsets()
	GetCheckpointMetric() *CheckpointMetric
}

func NewStream

func NewStream(client couchbase.Client,
	metadata metadata.Metadata,
	config *config.Dcp,
	version *couchbase.Version,
	bucketInfo *couchbase.BucketInfo,
	vBucketDiscovery VBucketDiscovery,
	listener models.Listener,
	collectionIDs map[uint32]string,
	stopCh chan struct{},
	bus EventBus.Bus,
	eventHandler models.EventHandler,
) Stream

type VBucketDiscovery

type VBucketDiscovery interface {
	Get() []uint16
	Close()
	GetMetric() *VBucketDiscoveryMetric
}

func NewVBucketDiscovery

func NewVBucketDiscovery(client couchbase.Client,
	config *config.Dcp,
	vBucketNumber int,
	bus EventBus.Bus,
) VBucketDiscovery

type VBucketDiscoveryMetric

type VBucketDiscoveryMetric struct {
	Type              string
	TotalMembers      int
	MemberNumber      int
	VBucketCount      int
	VBucketRangeStart uint16
	VBucketRangeEnd   uint16
}