cassandra

package
v1.32.1 Latest Latest
Warning

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

 Go to latest Published: Oct 15, 2026 License: MIT Imports: 39 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.Cassandra,
	r resolver.ServiceResolver,
	expectedVersion string,
	logger log.Logger,
) error

CheckCompatibleVersion check the version compatibility

func CreateCassandraKeyspace 

func CreateCassandraKeyspace(s gocql.Session, keyspace string, replicas int, overwrite bool, logger log.Logger) (err error)

CreateCassandraKeyspace creates the keyspace using this session for given replica count

func DropCassandraKeyspace 

func DropCassandraKeyspace(s gocql.Session, keyspace string, logger log.Logger) (err error)

DropCassandraKeyspace drops the given keyspace, if it exists

func NewClusterMetadataStore 

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

NewClusterMetadataStore is used to create an instance of ClusterMetadataStore implementation

func NewMatchingTaskStore 

func NewMatchingTaskStore(
	session gocql.Session,
	logger log.Logger,
	enableFairness bool,
) p.TaskStore

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,
	logger log.Logger,
) 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(
	_ 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, serializer serialization.Serializer, logger log.Logger) *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 Factory 

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

Factory vends datastore implementations backed by cassandra

func NewFactory 

func NewFactory(
	cfg config.Cassandra,
	r resolver.ServiceResolver,
	clusterName string,
	logger log.Logger,
	metricsHandler metrics.Handler,
	serializer serialization.Serializer,
) *Factory

NewFactory returns an instance of a factory object which can be used to create data stores that are backed by cassandra

func NewFactoryFromSession 

func NewFactoryFromSession(
	cfg config.Cassandra,
	clusterName string,
	logger log.Logger,
	session commongocql.Session,
	serializer serialization.Serializer,
) *Factory

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

func (*Factory) Close 

func (f *Factory) Close()

Close closes the factory

func (*Factory) NewClusterMetadataStore 

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

NewClusterMetadataStore returns a metadata store

func (*Factory) NewExecutionStore 

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

NewExecutionStore returns a new ExecutionStore.

func (*Factory) NewFairTaskStore 

func (f *Factory) NewFairTaskStore() (p.TaskStore, error)

NewTaskStore returns a new task store

func (*Factory) NewMetadataStore 

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

NewMetadataStore returns a metadata store

func (*Factory) NewNexusEndpointStore 

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

NewNexusEndpointStore returns a new NexusEndpointStore

func (*Factory) NewQueue 

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

NewQueue returns a new queue backed by cassandra

func (*Factory) NewQueueV2 

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

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

func (*Factory) NewShardStore 

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

NewShardStore returns a new shard store

func (*Factory) NewTaskStore 

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

NewTaskStore returns a new task store

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.HistoryBranchUtil
}

func NewHistoryStore 

func NewHistoryStore(
	session gocql.Session,
	serializer serialization.Serializer,
) *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) GetHistoryBranchUtil 

func (h *HistoryStore) GetHistoryBranchUtil() p.HistoryBranchUtil

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 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)

CreateNamespace create a namespace Cassandra does not support conditional updates across multiple tables. For this reason we have to first insert into 'Namespaces' table and then do a conditional insert into namespaces_by_name table. If the conditional write fails we delete the orphaned entry from namespaces table. There is a chance delete entry could fail and we never delete the orphaned entry from namespaces table. We might need a background job to delete those orphaned record.

func (*MetadataStore) CreateNamespaceInV2Table 

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

CreateNamespaceInV2Table is the temporary function used by namespace v1 -> v2 migration

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

RenameNamespace should be used with caution. Not every namespace can be renamed because namespace name are stored in the database. It may leave database in inconsistent state and must be retried until success. Step 1. Update row in `namespaces_by_id` table with the new name. Step 2. Batch of: 

Insert row into `namespaces` table with new name and new `notification_version`.
Delete row from `namespaces` table with old name.
Update `notification_version` in metadata row.

NOTE: `namespaces_by_id` is currently used only for `DescribeNamespace` API and namespace Id collision check.

func (*MetadataStore) UpdateNamespace 

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

type MutableStateStore 

type MutableStateStore struct {
	Session gocql.Session
	// contains filtered or unexported fields
}

func NewMutableStateStore 

func NewMutableStateStore(session gocql.Session, serializer serialization.Serializer, logger log.Logger) *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
	// contains filtered or unexported fields
}

func NewMutableStateTaskStore 

func NewMutableStateTaskStore(session gocql.Session, serializer serialization.Serializer) *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 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

type TestCluster 

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

TestCluster allows executing cassandra operations in testing.

func NewTestCluster 

func NewTestCluster(keyspace, username, password, host string, port int, schemaDir string, faultInjection *config.FaultInjection, logger log.Logger) *TestCluster

NewTestCluster returns a new cassandra test cluster

func (*TestCluster) Config 

func (s *TestCluster) Config() config.Persistence

Config returns the persistence config for connecting to this test cluster

func (*TestCluster) CreateDatabase 

func (s *TestCluster) CreateDatabase()

CreateDatabase from PersistenceTestCluster interface

func (*TestCluster) CreateSession 

func (s *TestCluster) CreateSession(
	keyspace string,
)

CreateSession from PersistenceTestCluster interface

func (*TestCluster) DatabaseName 

func (s *TestCluster) DatabaseName() string

DatabaseName from PersistenceTestCluster interface

func (*TestCluster) DropDatabase 

func (s *TestCluster) DropDatabase()

DropDatabase from PersistenceTestCluster interface

func (*TestCluster) GetSession 

func (s *TestCluster) GetSession() commongocql.Session

func (*TestCluster) LoadSchema 

func (s *TestCluster) LoadSchema(schemaFile string)

LoadSchema from PersistenceTestCluster interface

func (*TestCluster) SetupTestDatabase 

func (s *TestCluster) SetupTestDatabase()

SetupTestDatabase from PersistenceTestCluster interface

func (*TestCluster) TearDownTestDatabase 

func (s *TestCluster) TearDownTestDatabase()

TearDownTestDatabase from PersistenceTestCluster interface

Source Files

View all Source files

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.