datafx

Jun 10, 2025

Changes in this version

+ var ErrQueueStreamNotSupported = errors.New("connection does not support queue stream operations")

+ var ErrQueueStreamOperation = errors.New("queue stream operation failed")

type Queue

+ func (q *Queue) AckMessage(ctx context.Context, queueName, consumerGroup, receiptHandle string) error

+ func (q *Queue) ClaimPendingMessages(ctx context.Context, queueName string, consumerGroup string, ...) ([]connfx.Message, error)

+ func (q *Queue) ConsumeWithGroup(ctx context.Context, queueName string, consumerGroup string, ...) (<-chan connfx.Message, <-chan error)

+ func (q *Queue) DeclareQueueWithConfig(ctx context.Context, name string, config connfx.QueueConfig) (string, error)

+ func (q *Queue) DeleteMessage(ctx context.Context, queueName, receiptHandle string) error

+ func (q *Queue) GetStreamRepository() (connfx.QueueStreamRepository, error)

+ func (q *Queue) IsStreamSupported() bool

+ func (q *Queue) ProcessMessagesWithGroup(ctx context.Context, queueName string, consumerGroup string, ...) error

+ func (q *Queue) PublishRawWithHeaders(ctx context.Context, queueName string, data []byte, headers map[string]any) error

+ func (q *Queue) PublishWithHeaders(ctx context.Context, queueName string, message any, headers map[string]any) error

+ type QueueStream struct

+ func NewQueueStream(conn connfx.Connection) (*QueueStream, error)

+ func (s *QueueStream) AckMessage(ctx context.Context, streamName string, consumerGroup string, messageID string) error

+ func (s *QueueStream) ClaimPendingMessages(ctx context.Context, streamName string, consumerGroup string, ...) ([]connfx.Message, error)

+ func (s *QueueStream) ConsumeFromGroup(ctx context.Context, streamName string, consumerGroup string, ...) (<-chan connfx.Message, <-chan error)

+ func (s *QueueStream) CreateConsumerGroup(ctx context.Context, streamName string, consumerGroup string, startFrom string) error

+ func (s *QueueStream) CreateStream(ctx context.Context, streamName string) error

+ func (s *QueueStream) GetConnection() connfx.Connection

+ func (s *QueueStream) GetConsumerGroupInfo(ctx context.Context, streamName string) ([]connfx.ConsumerGroupInfo, error)

+ func (s *QueueStream) GetRepository() connfx.QueueStreamRepository

+ func (s *QueueStream) GetStreamInfo(ctx context.Context, streamName string) (connfx.StreamInfo, error)

+ func (s *QueueStream) ProcessMessagesFromGroup(ctx context.Context, streamName string, consumerGroup string, ...) error

+ func (s *QueueStream) SendMessage(ctx context.Context, streamName string, message any) (string, error)

+ func (s *QueueStream) SendMessageWithHeaders(ctx context.Context, streamName string, message any, headers map[string]any) (string, error)

+ func (s *QueueStream) TrimStream(ctx context.Context, streamName string, maxLength int64) error

Jun 10, 2025

Changes in this version

+ var ErrCacheNotSupported = errors.New("connection does not support cache operations")

+ var ErrCacheOperation = errors.New("cache operation failed")

+ var ErrConnectionNotSupported = errors.New("connection does not support required operations")

+ var ErrContextCanceled = errors.New("context canceled")

+ var ErrFailedToMarshal = errors.New("failed to marshal data")

+ var ErrFailedToUnmarshal = errors.New("failed to unmarshal data")

+ var ErrInvalidData = errors.New("invalid data")

+ var ErrKeyExpired = errors.New("key has expired")

+ var ErrKeyNotFound = errors.New("key not found")

+ var ErrMessageProcessing = errors.New("message processing failed")

+ var ErrQueueNotSupported = errors.New("connection does not support queue operations")

+ var ErrQueueOperation = errors.New("queue operation failed")

+ var ErrRepositoryOperation = errors.New("repository operation failed")

+ var ErrTransactionFailed = errors.New("transaction failed")

+ var ErrTransactionNotSupported = errors.New("connection does not support transactions")

+ var ErrTransactionOperation = errors.New("transaction operation failed")

+ type Cache struct

+ func NewCache(conn connfx.Connection) (*Cache, error)

+ func (c *Cache) Delete(ctx context.Context, key string) error

+ func (c *Cache) Exists(ctx context.Context, key string) (bool, error)

+ func (c *Cache) Expire(ctx context.Context, key string, expiration time.Duration) error

+ func (c *Cache) Get(ctx context.Context, key string, dest any) error

+ func (c *Cache) GetConnection() connfx.Connection

+ func (c *Cache) GetRaw(ctx context.Context, key string) ([]byte, error)

+ func (c *Cache) GetRepository() connfx.CacheRepository

+ func (c *Cache) GetTTL(ctx context.Context, key string) (time.Duration, error)

+ func (c *Cache) Set(ctx context.Context, key string, value any, expiration time.Duration) error

+ func (c *Cache) SetRaw(ctx context.Context, key string, value []byte, expiration time.Duration) error

+ type Queue struct

+ func NewQueue(conn connfx.Connection) (*Queue, error)

+ func (q *Queue) Consume(ctx context.Context, queueName string, config connfx.ConsumerConfig) (<-chan connfx.Message, <-chan error)

+ func (q *Queue) ConsumeWithDefaults(ctx context.Context, queueName string) (<-chan connfx.Message, <-chan error)

+ func (q *Queue) DeclareQueue(ctx context.Context, name string) (string, error)

+ func (q *Queue) GetConnection() connfx.Connection

+ func (q *Queue) GetRepository() connfx.QueueRepository

+ func (q *Queue) ProcessMessages(ctx context.Context, queueName string, config connfx.ConsumerConfig, ...) error

+ func (q *Queue) ProcessMessagesWithDefaults(ctx context.Context, queueName string, ...) error

+ func (q *Queue) Publish(ctx context.Context, queueName string, message any) error

+ func (q *Queue) PublishRaw(ctx context.Context, queueName string, data []byte) error

+ type Store struct

+ func NewStore(conn connfx.Connection) (*Store, error)

+ func (s *Store) Exists(ctx context.Context, key string) (bool, error)

+ func (s *Store) Get(ctx context.Context, key string, dest any) error

+ func (s *Store) GetConnection() connfx.Connection

+ func (s *Store) GetRaw(ctx context.Context, key string) ([]byte, error)

+ func (s *Store) GetRepository() connfx.Repository

+ func (s *Store) Remove(ctx context.Context, key string) error

+ func (s *Store) Set(ctx context.Context, key string, value any) error

+ func (s *Store) SetRaw(ctx context.Context, key string, value []byte) error

+ func (s *Store) Update(ctx context.Context, key string, value any) error

+ func (s *Store) UpdateRaw(ctx context.Context, key string, value []byte) error

+ type TransactionStore struct

+ Repository connfx.Repository

+ func (ts *TransactionStore) Exists(ctx context.Context, key string) (bool, error)

+ func (ts *TransactionStore) Get(ctx context.Context, key string, dest any) error

+ func (ts *TransactionStore) GetRaw(ctx context.Context, key string) ([]byte, error)

+ func (ts *TransactionStore) Remove(ctx context.Context, key string) error

+ func (ts *TransactionStore) Set(ctx context.Context, key string, value any) error

+ func (ts *TransactionStore) SetRaw(ctx context.Context, key string, value []byte) error

+ func (ts *TransactionStore) Update(ctx context.Context, key string, value any) error

+ func (ts *TransactionStore) UpdateRaw(ctx context.Context, key string, value []byte) error

+ type TransactionalStore struct

+ func NewTransactionalStore(conn connfx.Connection) (*TransactionalStore, error)

+ func (ts *TransactionalStore) ExecuteTransaction(ctx context.Context, fn func(*TransactionStore) error) error

Apr 15, 2025

Changes in this version

+ var ErrNoTransaction = errors.New("no transaction in progress")

+ var ErrTransactionAlreadyStarted = errors.New("transaction already started")

+ var ErrTransactionBeginFailed = errors.New("transaction begin failed")

+ var ErrTransactionCommitFailed = errors.New("transaction commit failed")

+ var ErrTransactionRollbackFailed = errors.New("transaction rollback failed")

type Datasource

+ ExecuteUnitOfWork func(ctx context.Context, fn func(uow *UnitOfWork) error) error

type SqlDatasource

+ func (datasource *SqlDatasource) ExecuteUnitOfWork(ctx context.Context, fn func(uow *UnitOfWork) error) error

+ type TransactionController interface

+ Commit func() error

+ Rollback func() error

type UnitOfWork

+ TransactionController TransactionController

+ TransactionStarter TransactionStarter

+ func CurrentUnitOfWork(ctx context.Context) *UnitOfWork

+ func (uow *UnitOfWork) Begin(ctx context.Context) error

+ func (uow *UnitOfWork) Execute(ctx context.Context, fn func(uow *UnitOfWork) error) error

+ func (uow *UnitOfWork) Rollback() error

Feb 27, 2025

Changes in this version

type Repository

+ Queries *T

+ func (r *Repository[T]) GetQueries() *T

+ func (r *Repository[T]) WithQueries(queries *T) *Repository[T]

Feb 27, 2025

Changes in this version

+ type Repository struct

+ Tx *T

+ func NewRepository[T any](tx *T) *Repository[T]

+ func (r *Repository[T]) GetTx() *T

+ func (r *Repository[T]) WithTx(tx *T) *Repository[T]

type UnitOfWork

+ func NewUnitOfWork(ctx context.Context, transactionStarter TransactionStarter) (*UnitOfWork, error)

Feb 9, 2025

Changes in this version

+ type Datasource interface

+ GetConnection func() SqlExecutor

+ GetDialect func() Dialect

+ UseUnitOfWork func(ctx context.Context) (*UnitOfWork, error)

type Registry

+ func (registry *Registry) GetDefault() Datasource

+ func (registry *Registry) GetNamed(name string) Datasource

type SqlDatasource

+ func NewSqlDatasource(ctx context.Context, dialect Dialect, dsn string) (*SqlDatasource, error)

Feb 6, 2025

Changes in this version

+ const DefaultDatasource

+ type ConfigDatasource struct

+ DSN string

+ Provider string

+ type SqlDatasource interface

+ GetConnection func() SqlExecutor

+ GetDialect func() Dialect

+ UseUnitOfWork func(ctx context.Context) (*UnitOfWork, error)

+ type SqlDatasourceStd struct

+ func NewSqlDatasourceStd(ctx context.Context, dialect Dialect, dsn string) (*SqlDatasourceStd, error)

+ func (dataSource *SqlDatasourceStd) GetConnection() SqlExecutor

+ func (dataSource *SqlDatasourceStd) GetDialect() Dialect

+ func (dataSource *SqlDatasourceStd) UseUnitOfWork(ctx context.Context) (*UnitOfWork, error)

Feb 5, 2025

Changes in this version

type SqlDataSource

+ GetDialect func() Dialect

type SqlDataSourceStd

+ func (dataSource *SqlDataSourceStd) GetDialect() Dialect

Feb 4, 2025

Changes in this version

+ var ErrUnableToDetermineDialect = errors.New("unable to determine dialect")

+ var ErrUnknownProvider = errors.New("unknown provider")

type ConfigDataSource

+ Provider string

Jan 22, 2025

Changes in this version

+ const DefaultDB

+ type Config struct

+ Sources map[string]ConfigDataSource

+ type ConfigDataSource struct

+ DSN string

+ type ContextKey string

+ const ContextKeyUnitOfWork

+ type Dialect string

+ const DialectMySQL

+ const DialectPostgres

+ const DialectSQLite

+ func DetermineDialect(dsn string) Dialect

+ type Registry struct

+ func NewRegistry(logger *slog.Logger) *Registry

+ func (registry *Registry) AddConnection(ctx context.Context, name string, dsn string) error

+ func (registry *Registry) GetDefaultSql() SqlDataSource

+ func (registry *Registry) GetNamedSql(name string) SqlDataSource

+ func (registry *Registry) LoadFromConfig(ctx context.Context, config *Config) error

+ type SqlDataSource interface

+ GetConnection func() SqlExecutor

+ UseUnitOfWork func(ctx context.Context) (*UnitOfWork, error)

+ type SqlDataSourceStd struct

+ func NewSqlDataSourceStd(ctx context.Context, dialect Dialect, dsn string) (*SqlDataSourceStd, error)

+ func (dataSource *SqlDataSourceStd) GetConnection() SqlExecutor

+ func (dataSource *SqlDataSourceStd) UseUnitOfWork(ctx context.Context) (*UnitOfWork, error)

+ type SqlExecutor interface

+ ExecContext func(ctx context.Context, query string, args ...any) (sql.Result, error)

+ PrepareContext func(ctx context.Context, query string) (*sql.Stmt, error)

+ QueryContext func(ctx context.Context, query string, args ...any) (*sql.Rows, error)

+ QueryRowContext func(ctx context.Context, query string, args ...any) *sql.Row

+ type TransactionFinalizer interface

+ Commit func() error

+ Rollback func() error

+ type TransactionStarter interface

+ BeginTx func(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error)

+ type UnitOfWork struct

+ func UseUnitOfWork(ctx context.Context, transactionStarter TransactionStarter) (*UnitOfWork, error)

+ func (uow *UnitOfWork) Bind(context context.Context, txScope TransactionFinalizer)

+ func (uow *UnitOfWork) Close() error

+ func (uow *UnitOfWork) Commit() error

+ func (uow *UnitOfWork) Context() context.Context

+ func (uow *UnitOfWork) TxScope() TransactionFinalizer

+ func (uow *UnitOfWork) Use(fn func(TransactionFinalizer) any)