cos

Documentation

Index

• Constants
• func URL(host string, port int) string
• type ReadSideManager
  • func NewReadSideManager(ctx context.Context, cosHost string, cosPort int) ReadSideManager

Constants

const (
	// KeepAliveTime is the period after which a keepalive ping is sent on the
	// transport
	KeepAliveTime = 1200 * time.Second
)

Functions

func URL

func URL(host string, port int) string

URL create a http connection address

Types

type ReadSideManager

type ReadSideManager interface {
	// GetLatestOffset retrieves the latest offset across all shards
	GetLatestOffset(ctx context.Context, in *connect.Request[cospb.GetLatestOffsetRequest]) (*connect.Response[cospb.GetLatestOffsetResponse], error)
	// GetLatestOffsetByShard retrieves the latest offset given a shard
	GetLatestOffsetByShard(ctx context.Context, in *connect.Request[cospb.GetLatestOffsetByShardRequest]) (*connect.Response[cospb.GetLatestOffsetByShardResponse], error)
	// RestartReadSide will clear the read side offset and start it over again
	// from the first offset and this across all shards
	RestartReadSide(ctx context.Context, in *connect.Request[cospb.RestartReadSideRequest]) (*connect.Response[cospb.RestartReadSideResponse], error)
	// RestartReadSideByShard will clear the read side offset for the given shard and start it over again from the first offset
	RestartReadSideByShard(ctx context.Context, in *connect.Request[cospb.RestartReadSideByShardRequest]) (*connect.Response[cospb.RestartReadSideByShardResponse], error)
	// PauseReadSide pauses a read side. This can be useful when running some data
	// migration and this across all shards
	PauseReadSide(ctx context.Context, in *connect.Request[cospb.PauseReadSideRequest]) (*connect.Response[cospb.PauseReadSideResponse], error)
	// PauseReadSideByShard pauses a read side. This can be useful when running some data
	// migration and this for a given shard
	PauseReadSideByShard(ctx context.Context, in *connect.Request[cospb.PauseReadSideByShardRequest]) (*connect.Response[cospb.PauseReadSideByShardResponse], error)
	// ResumeReadSide resumes a paused read side and this across all shards
	ResumeReadSide(ctx context.Context, in *connect.Request[cospb.ResumeReadSideRequest]) (*connect.Response[cospb.ResumeReadSideResponse], error)
	// ResumeReadSideByShard  resumes a paused read side for a given shard
	ResumeReadSideByShard(ctx context.Context, in *connect.Request[cospb.ResumeReadSideByShardRequest]) (*connect.Response[cospb.ResumeReadSideByShardResponse], error)
	// SkipOffset skips the current offset to read across all shards and continue with next. The operation will automatically restart the read side.
	SkipOffset(ctx context.Context, in *connect.Request[cospb.SkipOffsetRequest]) (*connect.Response[cospb.SkipOffsetResponse], error)
	// SkipOffsetByShard skips the current offset to read for a given shard and continue with next. The operation will automatically restart the read side.
	SkipOffsetByShard(ctx context.Context, in *connect.Request[cospb.SkipOffsetByShardRequest]) (*connect.Response[cospb.SkipOffsetByShardResponse], error)
}

ReadSideManager will be implemented by the CoS readSideManager to send requests to CoS read side manager service

func NewReadSideManager

func NewReadSideManager(ctx context.Context, cosHost string, cosPort int) ReadSideManager

NewReadSideManager creates a new instance of ReadSideManager