offset

Documentation

Index

• type Core
  • func (c *Core) DeleteOffset(ctx context.Context, m *Model) error
  • func (c *Core) DeleteOffsetStatus(ctx context.Context, m *Status) error
  • func (c *Core) Exists(ctx context.Context, m *Model) (bool, error)
  • func (c *Core) GetOffset(ctx context.Context, m *Model) (*Model, error)
  • func (c *Core) GetOffsetStatus(ctx context.Context, m *Status) (*Status, error)
  • func (c *Core) OffsetStatusExists(ctx context.Context, m *Status) (bool, error)
  • func (c *Core) RollBackOffset(ctx context.Context, m *Model) error
  • func (c *Core) SetOffset(ctx context.Context, m *Model) error
  • func (c *Core) SetOffsetStatus(ctx context.Context, m *Status) error
• type ICore
  • func NewCore(repo IRepo) ICore
• type IRepo
  • func NewRepo(registry registry.IRegistry) IRepo
• type Model
  • func (m *Model) Key() string
  • func (m *Model) Prefix() string
• type Repo
• type Status
  • func (m *Status) Key() string
  • func (m *Status) Prefix() string

Types

type Core

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

Core implements all business logic for offset

func (*Core) DeleteOffset

func (c *Core) DeleteOffset(ctx context.Context, m *Model) error

DeleteOffset deletes an offset

func (*Core) DeleteOffsetStatus

func (c *Core) DeleteOffsetStatus(ctx context.Context, m *Status) error

DeleteOffsetStatus deletes an offset status

func (*Core) Exists

func (c *Core) Exists(ctx context.Context, m *Model) (bool, error)

Exists to check if the offset exists with fully qualified consul key

func (*Core) GetOffset

func (c *Core) GetOffset(ctx context.Context, m *Model) (*Model, error)

GetOffset returns offset with the given key

func (*Core) GetOffsetStatus

func (c *Core) GetOffsetStatus(ctx context.Context, m *Status) (*Status, error)

GetOffsetStatus returns offset with the given key

func (*Core) OffsetStatusExists

func (c *Core) OffsetStatusExists(ctx context.Context, m *Status) (bool, error)

OffsetStatusExists to check if the offset status exists with fully qualified consul key

func (*Core) RollBackOffset

func (c *Core) RollBackOffset(ctx context.Context, m *Model) error

RollBackOffset rolls back to the previous offset in-memory

func (*Core) SetOffset

func (c *Core) SetOffset(ctx context.Context, m *Model) error

SetOffset sets/updates an offset

func (*Core) SetOffsetStatus

func (c *Core) SetOffsetStatus(ctx context.Context, m *Status) error

SetOffsetStatus sets an offset status

type ICore

type ICore interface {
	SetOffset(ctx context.Context, m *Model) error
	RollBackOffset(ctx context.Context, m *Model) error
	GetOffset(ctx context.Context, m *Model) (*Model, error)
	DeleteOffset(ctx context.Context, m *Model) error
	Exists(ctx context.Context, m *Model) (bool, error)

	SetOffsetStatus(ctx context.Context, m *Status) error
	OffsetStatusExists(ctx context.Context, m *Status) (bool, error)
	GetOffsetStatus(ctx context.Context, m *Status) (*Status, error)
	DeleteOffsetStatus(ctx context.Context, m *Status) error
}

ICore is an interface over offset core

func NewCore

func NewCore(repo IRepo) ICore

NewCore returns an instance of Core

type IRepo

type IRepo interface {
	common.IRepo
}

IRepo interface over database repository

func NewRepo

func NewRepo(registry registry.IRegistry) IRepo

NewRepo returns IRepo

type Model

type Model struct {
	common.BaseModel
	Topic        string `json:"topic_id"`
	Subscription string `json:"subscription_id"`
	Partition    int32  `json:"partition_id"`
	LatestOffset int32  `json:"latest_offset_value"`
	OrderingKey  string `json:"ordering_key"`
	// contains filtered or unexported fields
}

Model for an offset

func (*Model) Key

func (m *Model) Key() string

Key returns the key for storing in the registry

func (*Model) Prefix

func (m *Model) Prefix() string

Prefix returns the key prefix

type Repo

type Repo struct {
	common.BaseRepo
}

Repo implements various repository methods

type Status

type Status struct {
	Model
	OffsetStatus string `json:"status,omitempty"`
}

Status denotes the delivery status for an offset

func (*Status) Key

func (m *Status) Key() string

Key returns the key for storing in the registry

func (*Status) Prefix

func (m *Status) Prefix() string

Prefix returns the key prefix