storage

Documentation

Index

• Constants
• func BSIKey(tid uint64, instanceID uint64, wid uint32, step []byte) []byte
• func MappingIDKey(tenantID uint64, userID uint32) []byte
• func MappingKey(tenantID uint64, from metapb.IDValue, to string) []byte
• func OutputNotifyKey(tid uint64, ts int64, value []byte) []byte
• func PartitionKey(id uint64, partition uint32) []byte
• func ProfileKey(tenantID uint64, uid uint32) []byte
• func QueueItemKey(src []byte, offset uint64) []byte
• func QueueKVKey(id uint64, src []byte) []byte
• func QueueMetaKey(id uint64, partition uint32) []byte
• func QueueStateKey(id uint64, group []byte) []byte
• func ShardBitmapKey(value []byte, index uint32) []byte
• func TempKey(value []byte) []byte
• func TenantMetadataKey(id uint64) []byte
• func TenantRunnerKey(tid uint64, runner uint64) []byte
• func TenantRunnerLockKey(tid uint64, runner uint64, lockKey []byte) []byte
• func TenantRunnerMetadataKey(tid uint64, runner uint64) []byte
• func TenantRunnerOffsetKey(tid uint64, runner uint64) []byte
• func TenantRunnerWorkerInstanceMaxKey(tid uint64, runner uint64, wid uint64) []byte
• func TenantRunnerWorkerInstanceMinKey(tid uint64, runner uint64, wid uint64) []byte
• func TenantRunnerWorkerKey(tid uint64, runner uint64, wid uint64, worker uint32) []byte
• func TenantRunnerWorkerMaxKey(tid uint64, runner uint64) []byte
• func TenantRunnerWorkerMinKey(tid uint64, runner uint64) []byte
• func WorkflowCrowdShardKey(wid, instanceID uint64, version uint32) []byte
• func WorkflowCurrentInstanceKey(id uint64) []byte
• func WorkflowHistoryInstanceKey(wid, instanceID uint64) []byte
• type Event
• type Storage
  • func NewStorage(dataPath string, metadataStorage bhstorage.MetadataStorage, ...) (Storage, error)
  • func NewStorageWithOptions(dataPath string, metadataStorage bhstorage.MetadataStorage, ...) (Storage, error)
  • func NewTestStorage(t *testing.T, start bool) (Storage, func())

Constants

const (
	// StartingInstanceEvent  load a new instance event
	StartingInstanceEvent = iota
	// RunningInstanceEvent instance started event
	RunningInstanceEvent
	// StoppingInstanceEvent instance stopping event
	StoppingInstanceEvent
	// StoppedInstanceEvent instance stopped event
	StoppedInstanceEvent
	// InstanceWorkerCreatedEvent create a new instance state event
	InstanceWorkerCreatedEvent
	// InstanceWorkerDestoriedEvent destoried instance state event
	InstanceWorkerDestoriedEvent
	// StartRunnerEvent start runner
	StartRunnerEvent
	// StopRunnerEvent stop runner
	StopRunnerEvent
)

Functions

func BSIKey

func BSIKey(tid uint64, instanceID uint64, wid uint32, step []byte) []byte

BSIKey returns the bsi key

func MappingIDKey

func MappingIDKey(tenantID uint64, userID uint32) []byte

MappingIDKey returns a user id key,

func MappingKey

func MappingKey(tenantID uint64, from metapb.IDValue, to string) []byte

MappingKey returns a mapping key

func OutputNotifyKey

func OutputNotifyKey(tid uint64, ts int64, value []byte) []byte

OutputNotifyKey returns output notify key

func PartitionKey

func PartitionKey(id uint64, partition uint32) []byte

PartitionKey returns partition key

func ProfileKey

func ProfileKey(tenantID uint64, uid uint32) []byte

ProfileKey returns a profile key

func QueueItemKey

func QueueItemKey(src []byte, offset uint64) []byte

QueueItemKey store the item at the offset in the queue

func QueueKVKey

func QueueKVKey(id uint64, src []byte) []byte

QueueKVKey returns partition kv key

func QueueMetaKey

func QueueMetaKey(id uint64, partition uint32) []byte

QueueMetaKey returns concurrency queue meta key

func QueueStateKey

func QueueStateKey(id uint64, group []byte) []byte

QueueStateKey returns concurrency queue key

func ShardBitmapKey

func ShardBitmapKey(value []byte, index uint32) []byte

ShardBitmapKey returns bitmap shard key

func TempKey

func TempKey(value []byte) []byte

TempKey returns the temp key

func TenantMetadataKey

func TenantMetadataKey(id uint64) []byte

TenantMetadataKey returns queue metadata key

func TenantRunnerKey

func TenantRunnerKey(tid uint64, runner uint64) []byte

TenantRunnerKey tenant runner key

func TenantRunnerLockKey

func TenantRunnerLockKey(tid uint64, runner uint64, lockKey []byte) []byte

TenantRunnerLockKey tenant runner lock key

func TenantRunnerMetadataKey

func TenantRunnerMetadataKey(tid uint64, runner uint64) []byte

TenantRunnerMetadataKey tenant runner metadata key

func TenantRunnerOffsetKey

func TenantRunnerOffsetKey(tid uint64, runner uint64) []byte

TenantRunnerOffsetKey tenant runner offset key

func TenantRunnerWorkerInstanceMaxKey

func TenantRunnerWorkerInstanceMaxKey(tid uint64, runner uint64, wid uint64) []byte

TenantRunnerWorkerInstanceMaxKey tenant runner worker min key

func TenantRunnerWorkerInstanceMinKey

func TenantRunnerWorkerInstanceMinKey(tid uint64, runner uint64, wid uint64) []byte

TenantRunnerWorkerInstanceMinKey tenant runner worker min key

func TenantRunnerWorkerKey

func TenantRunnerWorkerKey(tid uint64, runner uint64, wid uint64, worker uint32) []byte

TenantRunnerWorkerKey tenant runner worker key

func TenantRunnerWorkerMaxKey

func TenantRunnerWorkerMaxKey(tid uint64, runner uint64) []byte

TenantRunnerWorkerMaxKey tenant runner worker max key

func TenantRunnerWorkerMinKey

func TenantRunnerWorkerMinKey(tid uint64, runner uint64) []byte

TenantRunnerWorkerMinKey tenant runner worker min key

func WorkflowCrowdShardKey

func WorkflowCrowdShardKey(wid, instanceID uint64, version uint32) []byte

WorkflowCrowdShardKey returns workflow instance crow shard key

func WorkflowCurrentInstanceKey

func WorkflowCurrentInstanceKey(id uint64) []byte

WorkflowCurrentInstanceKey workflow current instance key

func WorkflowHistoryInstanceKey

func WorkflowHistoryInstanceKey(wid, instanceID uint64) []byte

WorkflowHistoryInstanceKey returns workflow instance history key

Types

type Event

type Event struct {
	EventType int
	Data      interface{}
}

Event the event

type Storage

type Storage interface {
	// Start the storage
	Start() error
	// Close close the storage
	Close()
	// WatchInstance watch instance event
	WatchEvent() chan Event
	// Lock distruibuted lock
	Lock([]byte, []byte, int, time.Duration, bool, func(), metapb.Group) (bool, error)
	// Unlock unlock
	Unlock([]byte, []byte) error
	// Set set key value
	Set([]byte, []byte) error
	// Set set key value with a TTL in seconds
	SetWithTTL([]byte, []byte, int64) error
	// Get returns the value of key
	Get([]byte) ([]byte, error)
	// GetWithGroup returns the value of key
	GetWithGroup([]byte, metapb.Group) ([]byte, error)
	// Delete remove the key from the store
	Delete([]byte) error
	// Scan scan [start,end) data
	Scan([]byte, []byte, uint64) ([][]byte, [][]byte, error)
	// Scan scan [start,end) data
	ScanWithGroup([]byte, []byte, uint64, metapb.Group) ([][]byte, [][]byte, error)
	// PutToQueue put data to queue
	PutToQueue(id uint64, partition uint32, group metapb.Group, data ...[]byte) error
	// PutToQueueAndKV put data to queue and put a kv
	PutToQueueWithKV(id uint64, partition uint32, group metapb.Group, items [][]byte, kvs ...[]byte) error
	// PutToQueueWithKVAndCondition put data to queue and put a kv and a condition
	PutToQueueWithKVAndCondition(id uint64, partition uint32, group metapb.Group, items [][]byte, cond *rpcpb.Condition, kvs ...[]byte) error
	// PutToQueueWithAlloc put data to queue
	PutToQueueWithAlloc(id uint64, group metapb.Group, data ...[]byte) error
	// PutToQueueWithAllocAndKV put data to queue and put a kv
	PutToQueueWithAllocAndKV(id uint64, group metapb.Group, items [][]byte, kvs ...[]byte) error
	// PutToQueueWithAllocAndKVAndCondition put data to queue and put a kv and a condition
	PutToQueueWithAllocAndKVAndCondition(id uint64, group metapb.Group, items [][]byte, cond *rpcpb.Condition, kvs ...[]byte) error
	// ExecCommand exec command
	ExecCommand(cmd interface{}) ([]byte, error)
	// AsyncExecCommand async exec command
	AsyncExecCommand(interface{}, func(interface{}, []byte, error), interface{})
	// ExecCommandWithGroup exec command with group
	ExecCommandWithGroup(interface{}, metapb.Group) ([]byte, error)
	// AsyncExecCommandWithGroup async exec command with group
	AsyncExecCommandWithGroup(interface{}, metapb.Group, func(interface{}, []byte, error), interface{})
	// RaftStore returns the raft store
	RaftStore() raftstore.Store
}

Storage storage

func NewStorage

func NewStorage(dataPath string,
	metadataStorage bhstorage.MetadataStorage,
	dataStorages []bhstorage.DataStorage) (Storage, error)

NewStorage returns a beehive request handler

func NewStorageWithOptions

func NewStorageWithOptions(dataPath string,
	metadataStorage bhstorage.MetadataStorage,
	dataStorages []bhstorage.DataStorage, opts ...raftstore.Option) (Storage, error)

NewStorageWithOptions returns a beehive request handler

func NewTestStorage

func NewTestStorage(t *testing.T, start bool) (Storage, func())

NewTestStorage returns test storage