data

Documentation

Index

• type CacheReadOption
• type Manager
  • func NewDataManager(ctx context.Context, ns *core.Namespace, di database.Plugin, ...) (Manager, error)
• type NewMessage
• type Validator

Types

type CacheReadOption

type CacheReadOption int

Messages have fields that are mutable, in two categories

1) Can change multiple times like state - you cannot rely on the cache for these 2) Can go from being un-set, to being set, and once set are immutable. For (2) the cache provides a set of CacheReadOption modifiers that makes it safe to query the cache, even if the cache we slow to update asynchronously (active/active cluster being the ultimate example here, but from code inspection this is possible in the current cache).

const (
	// If you use CRORequirePublicBlobRefs then the cache will return a miss, if all data blobs do not have a `public` reference string
	CRORequirePublicBlobRefs CacheReadOption = iota
	// If you use CRORequirePins then the cache will return a miss, if the number of pins does not match the number of topics in the message.
	CRORequirePins
	// If you use CRORequestBatchID then the cache will return a miss, if there is no BatchID set.
	CRORequireBatchID
)

type Manager

type Manager interface {
	CheckDatatype(ctx context.Context, datatype *core.Datatype) error
	ValidateAll(ctx context.Context, data core.DataArray) (valid bool, err error)
	GetMessageWithDataCached(ctx context.Context, msgID *fftypes.UUID, options ...CacheReadOption) (msg *core.Message, data core.DataArray, foundAllData bool, err error)
	GetMessageDataCached(ctx context.Context, msg *core.Message, options ...CacheReadOption) (data core.DataArray, foundAll bool, err error)
	PeekMessageCache(ctx context.Context, id *fftypes.UUID, options ...CacheReadOption) (msg *core.Message, data core.DataArray)
	UpdateMessageCache(msg *core.Message, data core.DataArray)
	UpdateMessageIfCached(ctx context.Context, msg *core.Message)
	UpdateMessageStateIfCached(ctx context.Context, id *fftypes.UUID, state core.MessageState, confirmed *fftypes.FFTime, rejectReason string)
	ResolveInlineData(ctx context.Context, msg *NewMessage) error
	WriteNewMessage(ctx context.Context, newMsg *NewMessage) error
	BlobsEnabled() bool

	UploadJSON(ctx context.Context, inData *core.DataRefOrValue) (*core.Data, error)
	UploadBlob(ctx context.Context, inData *core.DataRefOrValue, blob *ffapi.Multipart, autoMeta bool) (*core.Data, error)
	DownloadBlob(ctx context.Context, dataID string) (*core.Blob, io.ReadCloser, error)
	DeleteData(ctx context.Context, dataID string) error
	HydrateBatch(ctx context.Context, persistedBatch *core.BatchPersisted) (*core.Batch, error)
	Start()
	WaitStop()
}

func NewDataManager

func NewDataManager(ctx context.Context, ns *core.Namespace, di database.Plugin, dx dataexchange.Plugin, cacheManager cache.Manager) (Manager, error)

type NewMessage

type NewMessage struct {
	Message *core.MessageInOut
	AllData core.DataArray
	NewData core.DataArray
}

type Validator

type Validator interface {
	Validate(ctx context.Context, data *core.Data) error
	ValidateValue(ctx context.Context, value *fftypes.JSONAny, expectedHash *fftypes.Bytes32) error
	Size() int64 // for cache management
}