driver

package
v1.1.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 30, 2025 License: MIT Imports: 34 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	TemplateEnqueueMessageQuery      = `` /* 155-byte string literal not displayed */
	TemplateGetMessagesQuery         = `` /* 188-byte string literal not displayed */
	TemplateGetMaxMessageIDQuery     = `` /* 134-byte string literal not displayed */
	TemplateCreateQueueQuery         = `` /* 126-byte string literal not displayed */
	TemplateGetQueueQuery            = `SELECT metadata_payload, metadata_encoding, version FROM queues WHERE queue_type = ? AND queue_name = ?`
	TemplateRangeDeleteMessagesQuery = `` /* 130-byte string literal not displayed */
	TemplateUpdateQueueMetadataQuery = `` /* 129-byte string literal not displayed */

)

Variables

View Source 
var (
	// ErrEnqueueMessageConflict is returned when a message with the same ID already exists in the queue. This is
	// possible when there are concurrent writes to the queue because we enqueue a message using two queries:
	//
	// 	1. SELECT MAX(ID) to get the next message ID (for a given queue partition)
	// 	2. INSERT (ID, message) with IF NOT EXISTS
	//
	// See the following example:
	//
	//  Client A           Client B                          Cassandra DB
	//  |                  |                                            |
	//  |--1. SELECT MAX(ID) FROM queue_messages----------------------->|
	//  |                  |                                            |
	//  |<-2. Return X--------------------------------------------------|
	//  |                  |                                            |
	//  |                  |--3. SELECT MAX(ID) FROM queue_messages---->|
	//  |                  |                                            |
	//  |                  |<-4. Return X-------------------------------|
	//  |                  |                                            |
	//  |--5. INSERT INTO queue_messages (ID = X)---------------------->|
	//  |                  |                                            |
	//  |<-6. Acknowledge-----------------------------------------------|
	//  |                  |                                            |
	//  |                  |--7. INSERT INTO queue_messages (ID = X)--->|
	//  |                  |                                            |
	//  |                  |<-8. Conflict/Error-------------------------|
	//  |                  |                                            |
	ErrEnqueueMessageConflict = &persistence.ConditionFailedError{
		Msg: "conflict inserting queue message, likely due to concurrent writes",
	}
	// ErrUpdateQueueConflict is returned when a queue is updated with the wrong version. This happens when there are
	// concurrent writes to the queue because we update a queue using two queries, similar to the enqueue message query.
	//
	// 	1. SELECT (queue, version) FROM queues
	// 	2. UPDATE queue, version IF version = version from step 1
	//
	// See the following example:
	//
	//  Client A           Client B                           Cassandra DB
	//  |                  |                                            |
	//  |--1. SELECT (queue, version) FROM queues---------------------->|
	//  |                  |                                            |
	//  |<-2. Return (queue, v1)----------------------------------------|
	//  |                  |                                            |
	//  |                  |--3. SELECT (queue, version) FROM queues--->|
	//  |                  |                                            |
	//  |                  |<-4. Return (queue, v1)---------------------|
	//  |                  |                                            |
	//  |--5. UPDATE queue, version IF version = v1-------------------->|
	//  |                  |                                            |
	//  |<-6. Acknowledge-----------------------------------------------|
	//  |                  |                                            |
	//  |                  |--7. UPDATE queue, version IF version = v1->|
	//  |                  |                                            |
	//  |                  |<-8. Conflict/Error-------------------------|
	//  |                  |                                            |
	ErrUpdateQueueConflict = &persistence.ConditionFailedError{
		Msg: "conflict updating queue, likely due to concurrent writes",
	}
)

Functions

func CheckCompatibleVersion 

func CheckCompatibleVersion(
	cfg config.CustomDatastoreConfig,
	r resolver.ServiceResolver,
	expectedVersion string,
) error

CheckCompatibleVersion check the version compatibility

func GetTaskTTL 

func GetTaskTTL(expireTime *timestamppb.Timestamp) int64

func NewClusterMetadataStore 

func NewClusterMetadataStore(
	session gocql.Session,
	logger log.Logger,
) (p.ClusterMetadataStore, error)

NewClusterMetadataStore is used to create an instance of ClusterMetadataStore implementation

func NewFactoryFromSession 

func NewFactoryFromSession(
	cfg ybconfig.Yugabyte,
	clusterName string,
	logger log.Logger,
	session localgocql.Session,
) p.DataStoreFactory

NewFactoryFromSession returns an instance of a factory object from the given session.

func NewMetadataStore 

func NewMetadataStore(
	currentClusterName string,
	session gocql.Session,
	logger log.Logger,
) (p.MetadataStore, error)

NewMetadataStore is used to create an instance of the Namespace MetadataStore implementation

func NewNexusEndpointStore 

func NewNexusEndpointStore(
	session gocql.Session,
	logger log.Logger,
) p.NexusEndpointStore

func NewQueueStore 

func NewQueueStore(
	queueType persistence.QueueType,
	session gocql.Session,
	logger log.Logger,
) (persistence.Queue, error)

func NewQueueV2Store 

func NewQueueV2Store(session gocql.Session, logger log.Logger) persistence.QueueV2

func VerifyCompatibleVersion 

func VerifyCompatibleVersion(
	cfg config.Persistence,
	r resolver.ServiceResolver,
) error

VerifyCompatibleVersion ensures that the installed version of temporal and visibility keyspaces is greater than or equal to the expected version. In most cases, the versions should match. However if after a schema upgrade there is a code rollback, the code version (expected version) would fall lower than the actual version in cassandra.

Types

type ClusterMetadataStore 

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

func (*ClusterMetadataStore) Close 

func (m *ClusterMetadataStore) Close()

func (*ClusterMetadataStore) DeleteClusterMetadata 

func (m *ClusterMetadataStore) DeleteClusterMetadata(
	ctx context.Context,
	request *p.InternalDeleteClusterMetadataRequest,
) error

func (*ClusterMetadataStore) GetClusterMembers 

func (m *ClusterMetadataStore) GetClusterMembers(
	ctx context.Context,
	request *p.GetClusterMembersRequest,
) (*p.GetClusterMembersResponse, error)

func (*ClusterMetadataStore) GetClusterMetadata 

func (m *ClusterMetadataStore) GetClusterMetadata(
	ctx context.Context,
	request *p.InternalGetClusterMetadataRequest,
) (*p.InternalGetClusterMetadataResponse, error)

func (*ClusterMetadataStore) GetName 

func (m *ClusterMetadataStore) GetName() string

func (*ClusterMetadataStore) ListClusterMetadata 

func (m *ClusterMetadataStore) ListClusterMetadata(
	ctx context.Context,
	request *p.InternalListClusterMetadataRequest,
) (*p.InternalListClusterMetadataResponse, error)

func (*ClusterMetadataStore) PruneClusterMembership 

func (m *ClusterMetadataStore) PruneClusterMembership(
	ctx context.Context,
	request *p.PruneClusterMembershipRequest,
) error

func (*ClusterMetadataStore) SaveClusterMetadata 

func (m *ClusterMetadataStore) SaveClusterMetadata(
	ctx context.Context,
	request *p.InternalSaveClusterMetadataRequest,
) (bool, error)

func (*ClusterMetadataStore) UpsertClusterMembership 

func (m *ClusterMetadataStore) UpsertClusterMembership(
	ctx context.Context,
	request *p.UpsertClusterMembershipRequest,
) error

type ExecutionStore 

type ExecutionStore struct {
	*HistoryStore
	*MutableStateStore
	*MutableStateTaskStore
}

func NewExecutionStore 

func NewExecutionStore(session gocql.Session) *ExecutionStore

func (*ExecutionStore) Close 

func (d *ExecutionStore) Close()

func (*ExecutionStore) ConflictResolveWorkflowExecution 

func (d *ExecutionStore) ConflictResolveWorkflowExecution(
	ctx context.Context,
	request *p.InternalConflictResolveWorkflowExecutionRequest,
) error

func (*ExecutionStore) CreateWorkflowExecution 

func (d *ExecutionStore) CreateWorkflowExecution(
	ctx context.Context,
	request *p.InternalCreateWorkflowExecutionRequest,
) (*p.InternalCreateWorkflowExecutionResponse, error)

func (*ExecutionStore) GetName 

func (d *ExecutionStore) GetName() string

func (*ExecutionStore) UpdateWorkflowExecution 

func (d *ExecutionStore) UpdateWorkflowExecution(
	ctx context.Context,
	request *p.InternalUpdateWorkflowExecutionRequest,
) error

type FieldNotFoundError 

type FieldNotFoundError struct {
	Msg string
}

FieldNotFoundError is an error type returned when an untyped query return does not contain the expected fields.

func (FieldNotFoundError) Error 

func (f FieldNotFoundError) Error() string

type HistoryStore 

type HistoryStore struct {
	Session gocql.Session
	p.HistoryBranchUtilImpl
}

func NewHistoryStore 

func NewHistoryStore(session gocql.Session) *HistoryStore

func (*HistoryStore) AppendHistoryNodes 

func (h *HistoryStore) AppendHistoryNodes(
	ctx context.Context,
	request *p.InternalAppendHistoryNodesRequest,
) error

AppendHistoryNodes upsert a batch of events as a single node to a history branch Note that it's not allowed to append above the branch's ancestors' nodes, which means nodeID >= ForkNodeID

func (*HistoryStore) DeleteHistoryBranch 

func (h *HistoryStore) DeleteHistoryBranch(
	ctx context.Context,
	request *p.InternalDeleteHistoryBranchRequest,
) error

DeleteHistoryBranch removes a branch

func (*HistoryStore) DeleteHistoryNodes 

func (h *HistoryStore) DeleteHistoryNodes(
	ctx context.Context,
	request *p.InternalDeleteHistoryNodesRequest,
) error

DeleteHistoryNodes delete a history node

func (*HistoryStore) ForkHistoryBranch 

func (h *HistoryStore) ForkHistoryBranch(
	ctx context.Context,
	request *p.InternalForkHistoryBranchRequest,
) error

ForkHistoryBranch forks a new branch from an existing branch Note that application must provide a void forking nodeID, it must be a valid nodeID in that branch. A valid forking nodeID can be an ancestor from the existing branch. For example, we have branch B1 with three nodes(1[1,2], 3[3,4,5] and 6[6,7,8]. 1, 3 and 6 are nodeIDs (first eventID of the batch). So B1 looks like this: 

     1[1,2]
     /
   3[3,4,5]
  /
6[6,7,8]

Assuming we have branch B2 which contains one ancestor B1 stopping at 6 (exclusive). So B2 inherit nodeID 1 and 3 from B1, and have its own nodeID 6 and 8. Branch B2 looks like this: 

  1[1,2]
  /
3[3,4,5]
 \
  6[6,7]
  \
   8[8]

Now we want to fork a new branch B3 from B2. The only valid forking nodeIDs are 3,6 or 8. 1 is not valid because we can't fork from first node. 2/4/5 is NOT valid either because they are inside a batch.

Case #1: If we fork from nodeID 6, then B3 will have an ancestor B1 which stops at 6(exclusive). As we append a batch of events[6,7,8,9] to B3, it will look like : 

  1[1,2]
  /
3[3,4,5]
 \
6[6,7,8,9]

Case #2: If we fork from node 8, then B3 will have two ancestors: B1 stops at 6(exclusive) and ancestor B2 stops at 8(exclusive) As we append a batch of events[8,9] to B3, it will look like: 

     1[1,2]
     /
   3[3,4,5]
  /
6[6,7]
 \
 8[8,9]

func (*HistoryStore) GetAllHistoryTreeBranches 

func (h *HistoryStore) GetAllHistoryTreeBranches(
	ctx context.Context,
	request *p.GetAllHistoryTreeBranchesRequest,
) (*p.InternalGetAllHistoryTreeBranchesResponse, error)

func (*HistoryStore) GetHistoryTreeContainingBranch 

func (h *HistoryStore) GetHistoryTreeContainingBranch(
	ctx context.Context,
	request *p.InternalGetHistoryTreeContainingBranchRequest,
) (*p.InternalGetHistoryTreeContainingBranchResponse, error)

GetHistoryTreeContainingBranch returns all branch information of a tree

func (*HistoryStore) ReadHistoryBranch 

func (h *HistoryStore) ReadHistoryBranch(
	ctx context.Context,
	request *p.InternalReadHistoryBranchRequest,
) (*p.InternalReadHistoryBranchResponse, error)

ReadHistoryBranch returns history node data for a branch NOTE: For branch that has ancestors, we need to query Cassandra multiple times, because it doesn't support OR/UNION operator

type InstanceFactory 

type InstanceFactory struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

InstanceFactory vends datastore implementations backed by driver

func (*InstanceFactory) Close 

func (f *InstanceFactory) Close()

Close closes the factory

func (*InstanceFactory) NewClusterMetadataStore 

func (f *InstanceFactory) NewClusterMetadataStore() (p.ClusterMetadataStore, error)

NewClusterMetadataStore returns a metadata store

func (*InstanceFactory) NewExecutionStore 

func (f *InstanceFactory) NewExecutionStore() (p.ExecutionStore, error)

NewExecutionStore returns a new ExecutionStore.

func (*InstanceFactory) NewMetadataStore 

func (f *InstanceFactory) NewMetadataStore() (p.MetadataStore, error)

NewMetadataStore returns a metadata store

func (*InstanceFactory) NewNexusEndpointStore 

func (f *InstanceFactory) NewNexusEndpointStore() (p.NexusEndpointStore, error)

NewNexusEndpointStore returns a new NexusEndpointStore

func (*InstanceFactory) NewQueue 

func (f *InstanceFactory) NewQueue(queueType p.QueueType) (p.Queue, error)

NewQueue returns a new queue backed by driver

func (*InstanceFactory) NewQueueV2 

func (f *InstanceFactory) NewQueueV2() (p.QueueV2, error)

NewQueueV2 returns a new data-access object for queues and messages stored in driver . It will never return an error.

func (*InstanceFactory) NewShardStore 

func (f *InstanceFactory) NewShardStore() (p.ShardStore, error)

NewShardStore returns a new shard store

func (*InstanceFactory) NewTaskStore 

func (f *InstanceFactory) NewTaskStore() (p.TaskStore, error)

NewTaskStore returns a new task store

type MatchingTaskStore 

type MatchingTaskStore struct {
	Session gocql.Session
	Logger  log.Logger
}

func NewMatchingTaskStore 

func NewMatchingTaskStore(
	session gocql.Session,
	logger log.Logger,
) *MatchingTaskStore

func (*MatchingTaskStore) Close 

func (d *MatchingTaskStore) Close()

func (*MatchingTaskStore) CompleteTasksLessThan 

func (d *MatchingTaskStore) CompleteTasksLessThan(
	ctx context.Context,
	request *p.CompleteTasksLessThanRequest,
) (int, error)

CompleteTasksLessThan deletes all tasks less than the given task id. This API ignores the Limit request parameter i.e. either all tasks leq the task_id will be deleted or an error will be returned to the caller

func (*MatchingTaskStore) CountTaskQueuesByBuildId 

func (d *MatchingTaskStore) CountTaskQueuesByBuildId(ctx context.Context, request *p.CountTaskQueuesByBuildIdRequest) (int, error)

func (*MatchingTaskStore) CreateTaskQueue 

func (d *MatchingTaskStore) CreateTaskQueue(
	ctx context.Context,
	request *p.InternalCreateTaskQueueRequest,
) error

func (*MatchingTaskStore) CreateTasks 

func (d *MatchingTaskStore) CreateTasks(
	ctx context.Context,
	request *p.InternalCreateTasksRequest,
) (*p.CreateTasksResponse, error)

CreateTasks add tasks

func (*MatchingTaskStore) DeleteTaskQueue 

func (d *MatchingTaskStore) DeleteTaskQueue(
	ctx context.Context,
	request *p.DeleteTaskQueueRequest,
) error

func (*MatchingTaskStore) GetName 

func (d *MatchingTaskStore) GetName() string

func (*MatchingTaskStore) GetTaskQueue 

func (d *MatchingTaskStore) GetTaskQueue(
	ctx context.Context,
	request *p.InternalGetTaskQueueRequest,
) (*p.InternalGetTaskQueueResponse, error)

func (*MatchingTaskStore) GetTaskQueueUserData 

func (d *MatchingTaskStore) GetTaskQueueUserData(
	ctx context.Context,
	request *p.GetTaskQueueUserDataRequest,
) (*p.InternalGetTaskQueueUserDataResponse, error)

func (*MatchingTaskStore) GetTaskQueuesByBuildId 

func (d *MatchingTaskStore) GetTaskQueuesByBuildId(ctx context.Context, request *p.GetTaskQueuesByBuildIdRequest) ([]string, error)

func (*MatchingTaskStore) GetTasks 

func (d *MatchingTaskStore) GetTasks(
	ctx context.Context,
	request *p.GetTasksRequest,
) (*p.InternalGetTasksResponse, error)

GetTasks get a task

func (*MatchingTaskStore) ListTaskQueue 

func (d *MatchingTaskStore) ListTaskQueue(
	_ context.Context,
	_ *p.ListTaskQueueRequest,
) (*p.InternalListTaskQueueResponse, error)

func (*MatchingTaskStore) ListTaskQueueUserDataEntries 

func (d *MatchingTaskStore) ListTaskQueueUserDataEntries(ctx context.Context, request *p.ListTaskQueueUserDataEntriesRequest) (*p.InternalListTaskQueueUserDataEntriesResponse, error)

func (*MatchingTaskStore) UpdateTaskQueue 

func (d *MatchingTaskStore) UpdateTaskQueue(
	ctx context.Context,
	request *p.InternalUpdateTaskQueueRequest,
) (*p.UpdateTaskQueueResponse, error)

UpdateTaskQueue update task queue

func (*MatchingTaskStore) UpdateTaskQueueUserData 

func (d *MatchingTaskStore) UpdateTaskQueueUserData(
	ctx context.Context,
	request *p.InternalUpdateTaskQueueUserDataRequest,
) error

type MetaFactory 

type MetaFactory struct {
}

func (*MetaFactory) NewFactory 

func (f *MetaFactory) NewFactory(
	cfg config.CustomDatastoreConfig,
	r resolver.ServiceResolver,
	clusterName string,
	logger log.Logger,
	metricsHandler metrics.Handler,
) p.DataStoreFactory

NewFactory returns an InstanceFactory object which can be used to create data stores that are backed by yugabyte

type MetadataStore 

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

func (*MetadataStore) Close 

func (m *MetadataStore) Close()

func (*MetadataStore) CreateNamespace 

func (m *MetadataStore) CreateNamespace(
	ctx context.Context,
	request *p.InternalCreateNamespaceRequest,
) (*p.CreateNamespaceResponse, error)

func (*MetadataStore) DeleteNamespace 

func (m *MetadataStore) DeleteNamespace(
	ctx context.Context,
	request *p.DeleteNamespaceRequest,
) error

func (*MetadataStore) DeleteNamespaceByName 

func (m *MetadataStore) DeleteNamespaceByName(
	ctx context.Context,
	request *p.DeleteNamespaceByNameRequest,
) error

func (*MetadataStore) GetMetadata 

func (m *MetadataStore) GetMetadata(
	ctx context.Context,
) (*p.GetMetadataResponse, error)

func (*MetadataStore) GetName 

func (m *MetadataStore) GetName() string

func (*MetadataStore) GetNamespace 

func (m *MetadataStore) GetNamespace(
	ctx context.Context,
	request *p.GetNamespaceRequest,
) (*p.InternalGetNamespaceResponse, error)

func (*MetadataStore) ListNamespaces 

func (m *MetadataStore) ListNamespaces(
	ctx context.Context,
	request *p.InternalListNamespacesRequest,
) (*p.InternalListNamespacesResponse, error)

func (*MetadataStore) RenameNamespace 

func (m *MetadataStore) RenameNamespace(
	ctx context.Context,
	request *p.InternalRenameNamespaceRequest,
) error

func (*MetadataStore) UpdateNamespace 

func (m *MetadataStore) UpdateNamespace(
	ctx context.Context,
	request *p.InternalUpdateNamespaceRequest,
) error

type MutableStateStore 

type MutableStateStore struct {
	Session gocql.Session
}

func NewMutableStateStore 

func NewMutableStateStore(session gocql.Session) *MutableStateStore

func (*MutableStateStore) ConflictResolveWorkflowExecution 

func (d *MutableStateStore) ConflictResolveWorkflowExecution(
	ctx context.Context,
	request *p.InternalConflictResolveWorkflowExecutionRequest,
) error

func (*MutableStateStore) CreateWorkflowExecution 

func (d *MutableStateStore) CreateWorkflowExecution(
	ctx context.Context,
	request *p.InternalCreateWorkflowExecutionRequest,
) (*p.InternalCreateWorkflowExecutionResponse, error)

func (*MutableStateStore) DeleteCurrentWorkflowExecution 

func (d *MutableStateStore) DeleteCurrentWorkflowExecution(
	ctx context.Context,
	request *p.DeleteCurrentWorkflowExecutionRequest,
) error

func (*MutableStateStore) DeleteWorkflowExecution 

func (d *MutableStateStore) DeleteWorkflowExecution(
	ctx context.Context,
	request *p.DeleteWorkflowExecutionRequest,
) error

func (*MutableStateStore) GetCurrentExecution 

func (d *MutableStateStore) GetCurrentExecution(
	ctx context.Context,
	request *p.GetCurrentExecutionRequest,
) (*p.InternalGetCurrentExecutionResponse, error)

func (*MutableStateStore) GetWorkflowExecution 

func (d *MutableStateStore) GetWorkflowExecution(
	ctx context.Context,
	request *p.GetWorkflowExecutionRequest,
) (*p.InternalGetWorkflowExecutionResponse, error)

func (*MutableStateStore) ListConcreteExecutions 

func (d *MutableStateStore) ListConcreteExecutions(
	ctx context.Context,
	request *p.ListConcreteExecutionsRequest,
) (*p.InternalListConcreteExecutionsResponse, error)

func (*MutableStateStore) SetWorkflowExecution 

func (d *MutableStateStore) SetWorkflowExecution(
	ctx context.Context,
	request *p.InternalSetWorkflowExecutionRequest,
) error

func (*MutableStateStore) UpdateWorkflowExecution 

func (d *MutableStateStore) UpdateWorkflowExecution(
	ctx context.Context,
	request *p.InternalUpdateWorkflowExecutionRequest,
) error

type MutableStateTaskStore 

type MutableStateTaskStore struct {
	Session gocql.Session
}

func NewMutableStateTaskStore 

func NewMutableStateTaskStore(session gocql.Session) *MutableStateTaskStore

func (*MutableStateTaskStore) AddHistoryTasks 

func (d *MutableStateTaskStore) AddHistoryTasks(
	ctx context.Context,
	request *p.InternalAddHistoryTasksRequest,
) error

func (*MutableStateTaskStore) CompleteHistoryTask 

func (d *MutableStateTaskStore) CompleteHistoryTask(
	ctx context.Context,
	request *p.CompleteHistoryTaskRequest,
) error

func (*MutableStateTaskStore) DeleteReplicationTaskFromDLQ 

func (d *MutableStateTaskStore) DeleteReplicationTaskFromDLQ(
	ctx context.Context,
	request *p.DeleteReplicationTaskFromDLQRequest,
) error

func (*MutableStateTaskStore) GetHistoryTasks 

func (d *MutableStateTaskStore) GetHistoryTasks(
	ctx context.Context,
	request *p.GetHistoryTasksRequest,
) (*p.InternalGetHistoryTasksResponse, error)

func (*MutableStateTaskStore) GetReplicationTasksFromDLQ 

func (d *MutableStateTaskStore) GetReplicationTasksFromDLQ(
	ctx context.Context,
	request *p.GetReplicationTasksFromDLQRequest,
) (*p.InternalGetHistoryTasksResponse, error)

func (*MutableStateTaskStore) IsReplicationDLQEmpty 

func (d *MutableStateTaskStore) IsReplicationDLQEmpty(
	ctx context.Context,
	request *p.GetReplicationTasksFromDLQRequest,
) (bool, error)

func (*MutableStateTaskStore) PutReplicationTaskToDLQ 

func (d *MutableStateTaskStore) PutReplicationTaskToDLQ(
	ctx context.Context,
	request *p.PutReplicationTaskToDLQRequest,
) error

func (*MutableStateTaskStore) RangeCompleteHistoryTasks 

func (d *MutableStateTaskStore) RangeCompleteHistoryTasks(
	ctx context.Context,
	request *p.RangeCompleteHistoryTasksRequest,
) error

func (*MutableStateTaskStore) RangeDeleteReplicationTaskFromDLQ 

func (d *MutableStateTaskStore) RangeDeleteReplicationTaskFromDLQ(
	ctx context.Context,
	request *p.RangeDeleteReplicationTaskFromDLQRequest,
) error

type NexusEndpointStore 

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

func (*NexusEndpointStore) Close 

func (s *NexusEndpointStore) Close()

func (*NexusEndpointStore) CreateOrUpdateNexusEndpoint 

func (s *NexusEndpointStore) CreateOrUpdateNexusEndpoint(
	ctx context.Context,
	request *p.InternalCreateOrUpdateNexusEndpointRequest,
) error

func (*NexusEndpointStore) DeleteNexusEndpoint 

func (s *NexusEndpointStore) DeleteNexusEndpoint(
	ctx context.Context,
	request *p.DeleteNexusEndpointRequest,
) error

func (*NexusEndpointStore) GetName 

func (s *NexusEndpointStore) GetName() string

func (*NexusEndpointStore) GetNexusEndpoint 

func (s *NexusEndpointStore) GetNexusEndpoint(
	ctx context.Context,
	request *p.GetNexusEndpointRequest,
) (*p.InternalNexusEndpoint, error)

func (*NexusEndpointStore) ListNexusEndpoints 

func (s *NexusEndpointStore) ListNexusEndpoints(
	ctx context.Context,
	request *p.ListNexusEndpointsRequest,
) (*p.InternalListNexusEndpointsResponse, error)

type PersistedTypeMismatchError 

type PersistedTypeMismatchError struct {
	Msg string
}

PersistedTypeMismatchError is an error type returned when a persisted cassandra value does not match the expected type.

func (PersistedTypeMismatchError) Error 

func (f PersistedTypeMismatchError) Error() string

type PostQueryValidation 

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

func NewPostQueryValidation 

func NewPostQueryValidation() *PostQueryValidation

func (*PostQueryValidation) Add 

func (p *PostQueryValidation) Add(validator func() error)

func (*PostQueryValidation) Validate 

func (d *PostQueryValidation) Validate() error

type Queue 

type Queue struct {
	Metadata *persistencespb.Queue
	Version  int64
}

func GetQueue 

func GetQueue(
	ctx context.Context,
	session gocql.Session,
	queueName string,
	queueType persistence.QueueV2Type,
) (*Queue, error)

type QueueStore 

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

func (*QueueStore) Close 

func (q *QueueStore) Close()

func (*QueueStore) DeleteMessageFromDLQ 

func (q *QueueStore) DeleteMessageFromDLQ(
	ctx context.Context,
	messageID int64,
) error

func (*QueueStore) DeleteMessagesBefore 

func (q *QueueStore) DeleteMessagesBefore(
	ctx context.Context,
	messageID int64,
) error

func (*QueueStore) EnqueueMessage 

func (q *QueueStore) EnqueueMessage(
	ctx context.Context,
	blob *commonpb.DataBlob,
) error

func (*QueueStore) EnqueueMessageToDLQ 

func (q *QueueStore) EnqueueMessageToDLQ(
	ctx context.Context,
	blob *commonpb.DataBlob,
) (int64, error)

func (*QueueStore) GetAckLevels 

func (q *QueueStore) GetAckLevels(
	ctx context.Context,
) (*persistence.InternalQueueMetadata, error)

func (*QueueStore) GetDLQAckLevels 

func (q *QueueStore) GetDLQAckLevels(
	ctx context.Context,
) (*persistence.InternalQueueMetadata, error)

func (*QueueStore) Init 

func (q *QueueStore) Init(
	ctx context.Context,
	blob *commonpb.DataBlob,
) error

func (*QueueStore) RangeDeleteMessagesFromDLQ 

func (q *QueueStore) RangeDeleteMessagesFromDLQ(
	ctx context.Context,
	firstMessageID int64,
	lastMessageID int64,
) error

func (*QueueStore) ReadMessages 

func (q *QueueStore) ReadMessages(
	ctx context.Context,
	lastMessageID int64,
	maxCount int,
) ([]*persistence.QueueMessage, error)

func (*QueueStore) ReadMessagesFromDLQ 

func (q *QueueStore) ReadMessagesFromDLQ(
	ctx context.Context,
	firstMessageID int64,
	lastMessageID int64,
	pageSize int,
	pageToken []byte,
) ([]*persistence.QueueMessage, []byte, error)

func (*QueueStore) UpdateAckLevel 

func (q *QueueStore) UpdateAckLevel(
	ctx context.Context,
	metadata *persistence.InternalQueueMetadata,
) error

func (*QueueStore) UpdateDLQAckLevel 

func (q *QueueStore) UpdateDLQAckLevel(
	ctx context.Context,
	metadata *persistence.InternalQueueMetadata,
) error

type SchemaVersionReader 

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

func NewSchemaVersionReader 

func NewSchemaVersionReader(session gocql.Session) *SchemaVersionReader

func (*SchemaVersionReader) ReadSchemaVersion 

func (svr *SchemaVersionReader) ReadSchemaVersion(keyspace string) (string, error)

ReadSchemaVersion returns the current schema version for the Keyspace

type ShardStore 

type ShardStore struct {
	ClusterName string
	Session     gocql.Session
	Logger      log.Logger
}

func NewShardStore 

func NewShardStore(
	clusterName string,
	session gocql.Session,
	logger log.Logger,
) *ShardStore

func (*ShardStore) AssertShardOwnership 

func (d *ShardStore) AssertShardOwnership(
	ctx context.Context,
	request *p.AssertShardOwnershipRequest,
) error

func (*ShardStore) Close 

func (d *ShardStore) Close()

func (*ShardStore) GetClusterName 

func (d *ShardStore) GetClusterName() string

func (*ShardStore) GetName 

func (d *ShardStore) GetName() string

func (*ShardStore) GetOrCreateShard 

func (d *ShardStore) GetOrCreateShard(
	ctx context.Context,
	request *p.InternalGetOrCreateShardRequest,
) (*p.InternalGetOrCreateShardResponse, error)

func (*ShardStore) UpdateShard 

func (d *ShardStore) UpdateShard(
	ctx context.Context,
	request *p.InternalUpdateShardRequest,
) error

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.