Versions in this module Expand all Collapse all v0 v0.0.1 Dec 11, 2025 Changes in this version + var CausationKey = capitan.NewStringKey("causation_id") + var CorrelationKey = capitan.NewStringKey("correlation_id") + var DeadLetterRouted = capitan.NewSignal("ago.deadletter.routed", "Message routed to dead letter") + var ErrNotFound = errors.New("ago: state not found") + var ErrTimeout = errors.New("ago: request timeout") + var ErrorKey = capitan.NewErrorKey("error") + var FlowCompleted = capitan.NewSignal("ago.flow.completed", "Flow completed") + var FlowCreated = capitan.NewSignal("ago.flow.created", "Flow created") + var FlowFailed = capitan.NewSignal("ago.flow.failed", "Flow failed") + var IdempotencyKey = capitan.NewStringKey("idempotency_key") + var RequestSent = capitan.NewSignal("ago.request.sent", "Request sent") + var RequestTimeout = capitan.NewSignal("ago.request.timeout", "Request timed out") + var ResponseReceived = capitan.NewSignal("ago.response.received", "Response received") + var SagaCompensating = capitan.NewSignal("ago.saga.compensating", "Saga compensating") + var SagaCompleted = capitan.NewSignal("ago.saga.completed", "Saga completed") + var SagaFailed = capitan.NewSignal("ago.saga.failed", "Saga failed") + var SagaStarted = capitan.NewSignal("ago.saga.started", "Saga started") + var SagaStatusKey = capitan.NewKey[SagaStatus]("saga_status", "ago.SagaStatus") + var SagaStepCompleted = capitan.NewSignal("ago.saga.step.completed", "Saga step completed") + var StepNameKey = capitan.NewStringKey("step_name") + func Backoff[T any](name string, processor pipz.Chainable[*Flow[T]], maxAttempts int, ...) *pipz.Backoff[*Flow[T]] + func CircuitBreaker[T any](name string, processor pipz.Chainable[*Flow[T]], failureThreshold int, ...) *pipz.CircuitBreaker[*Flow[T]] + func Concurrent[T any](name string, ...) *pipz.Concurrent[*Flow[T]] + func CorrelateFrom[T any](name pipz.Name, parentCorrelation string) pipz.Chainable[*Flow[T]] + func Correlate[T any](name pipz.Name) pipz.Chainable[*Flow[T]] + func Do[T any](name string, fn func(context.Context, *Flow[T]) (*Flow[T], error)) pipz.Processor[*Flow[T]] + func Effect[T any](name string, fn func(context.Context, *Flow[T]) error) pipz.Processor[*Flow[T]] + func EnrichOptional[T, V any](name pipz.Name, key capitan.GenericKey[V], ...) pipz.Chainable[*Flow[T]] + func EnrichWith[T any](name string, fn func(context.Context, *Flow[T]) (*Flow[T], error)) pipz.Processor[*Flow[T]] + func Enrich[T, V any](name pipz.Name, key capitan.GenericKey[V], ...) pipz.Chainable[*Flow[T]] + func Fallback[T any](name string, processors ...pipz.Chainable[*Flow[T]]) *pipz.Fallback[*Flow[T]] + func Filter[T any](name string, predicate func(context.Context, *Flow[T]) bool, ...) *pipz.Filter[*Flow[T]] + func From[T, V any](f *Flow[T], key capitan.GenericKey[V]) (V, bool) + func Gate[T any](name string, predicate func(context.Context, *Flow[T]) bool) pipz.Processor[*Flow[T]] + func Handle[T any](name string, processor pipz.Chainable[*Flow[T]], ...) *pipz.Handle[*Flow[T]] + func Mutate[T any](name string, fn func(context.Context, *Flow[T]) *Flow[T], ...) pipz.Processor[*Flow[T]] + func Publish[T any](name pipz.Name, provider herald.Provider) pipz.Chainable[*Flow[T]] + func Race[T any](name string, processors ...pipz.Chainable[*Flow[T]]) *pipz.Race[*Flow[T]] + func RateLimiter[T any](name string, requestsPerSecond float64, burst int) *pipz.RateLimiter[*Flow[T]] + func RecoverSagas[T any](ctx context.Context, store Store, key capitan.GenericKey[T], ...) error + func Retry[T any](name string, processor pipz.Chainable[*Flow[T]], maxAttempts int) *pipz.Retry[*Flow[T]] + func Sequence[T any](name string, processors ...pipz.Chainable[*Flow[T]]) *pipz.Sequence[*Flow[T]] + func Switch[T any, K comparable](name string, condition func(context.Context, *Flow[T]) K) *pipz.Switch[*Flow[T], K] + func TagFrom[T any](name pipz.Name, key string, valueFn func(T) string) pipz.Chainable[*Flow[T]] + func Tag[T any](name pipz.Name, key, value string) pipz.Chainable[*Flow[T]] + func Timeout[T any](name string, processor pipz.Chainable[*Flow[T]], duration time.Duration) *pipz.Timeout[*Flow[T]] + func Transform[T any](name string, fn func(context.Context, *Flow[T]) *Flow[T]) pipz.Processor[*Flow[T]] + func WorkerPool[T any](name string, workers int, processors ...pipz.Chainable[*Flow[T]]) *pipz.WorkerPool[*Flow[T]] + type Await struct + func NewAwait[T, V any](name pipz.Name, signal capitan.Signal, key capitan.GenericKey[V]) *Await[T, V] + func (*Await[T, V]) Close() error + func (a *Await[T, V]) Build() pipz.Chainable[*Flow[T]] + func (a *Await[T, V]) Name() pipz.Name + func (a *Await[T, V]) Process(ctx context.Context, f *Flow[T]) (*Flow[T], error) + func (a *Await[T, V]) Timeout(d time.Duration) *Await[T, V] + func (a *Await[T, V]) WithCapitan(c *capitan.Capitan) *Await[T, V] + type CerealStore struct + func NewCerealStore(db *sqlx.DB) *CerealStore + func (s *CerealStore) DeletePending(ctx context.Context, correlationID string) error + func (s *CerealStore) DeleteSaga(ctx context.Context, correlationID string) error + func (s *CerealStore) GetPending(ctx context.Context, correlationID string) (*PendingState, error) + func (s *CerealStore) GetSaga(ctx context.Context, correlationID string) (*SagaState, error) + func (s *CerealStore) IsCompensated(ctx context.Context, correlationID, stepName string) (bool, error) + func (s *CerealStore) ListIncompleteSagas(ctx context.Context) ([]*SagaState, error) + func (s *CerealStore) MarkCompensated(ctx context.Context, correlationID, stepName string) error + func (s *CerealStore) Migrate(ctx context.Context) error + func (s *CerealStore) SetPending(ctx context.Context, correlationID string, state *PendingState) error + func (s *CerealStore) SetSaga(ctx context.Context, correlationID string, state *SagaState) error + func (s *CerealStore) UpdateSaga(ctx context.Context, correlationID string, state *SagaState) error + func (s *CerealStore) WithSaga(ctx context.Context, correlationID string, ...) error + type Compensate struct + func NewCompensate[T any](name pipz.Name, store Store, key capitan.GenericKey[T]) *Compensate[T] + func (*Compensate[T]) Close() error + func (c *Compensate[T]) Build() pipz.Chainable[*Flow[T]] + func (c *Compensate[T]) Name() pipz.Name + func (c *Compensate[T]) Process(ctx context.Context, f *Flow[T]) (*Flow[T], error) + func (c *Compensate[T]) WithCapitan(cpt *capitan.Capitan) *Compensate[T] + type CompensationRecord struct + Data []byte + Signal capitan.Signal + StepName string + type DeadLetter struct + func NewDeadLetter[T any](name pipz.Name, key capitan.GenericKey[T]) *DeadLetter[T] + func (*DeadLetter[T]) Close() error + func (d *DeadLetter[T]) Build() pipz.Chainable[*Flow[T]] + func (d *DeadLetter[T]) Name() pipz.Name + func (d *DeadLetter[T]) Process(ctx context.Context, f *Flow[T]) (*Flow[T], error) + func (d *DeadLetter[T]) WithCapitan(c *capitan.Capitan) *DeadLetter[T] + func (d *DeadLetter[T]) WithProvider(provider herald.Provider) *DeadLetter[T] + func (d *DeadLetter[T]) WithSignal(signal capitan.Signal) *DeadLetter[T] + type Emit struct + func NewEmit[T any](name pipz.Name, signal capitan.Signal, key capitan.GenericKey[T]) *Emit[T] + func (*Emit[T]) Close() error + func (e *Emit[T]) Build() pipz.Chainable[*Flow[T]] + func (e *Emit[T]) Name() pipz.Name + func (e *Emit[T]) Process(ctx context.Context, f *Flow[T]) (*Flow[T], error) + func (e *Emit[T]) WithCapitan(c *capitan.Capitan) *Emit[T] + type Flow struct + CausationID string + CorrelationID string + Errors []error + Metadata map[string]string + Payload T + Severity capitan.Severity + Signal capitan.Signal + Timestamp time.Time + func NewFlow[T any](payload T, signal capitan.Signal) *Flow[T] + func NewFromEvent[T any](e *capitan.Event, key capitan.GenericKey[T]) *Flow[T] + func (f *Flow[T]) AddError(err error) + func (f *Flow[T]) Clone() *Flow[T] + func (f *Flow[T]) Fields() []capitan.Field + func (f *Flow[T]) Get(key capitan.Key) capitan.Field + func (f *Flow[T]) HasErrors() bool + func (f *Flow[T]) Set(field capitan.Field) + type MemoryStore struct + func NewMemoryStore() *MemoryStore + func (m *MemoryStore) DeletePending(_ context.Context, correlationID string) error + func (m *MemoryStore) DeleteSaga(_ context.Context, correlationID string) error + func (m *MemoryStore) GetPending(_ context.Context, correlationID string) (*PendingState, error) + func (m *MemoryStore) GetSaga(_ context.Context, correlationID string) (*SagaState, error) + func (m *MemoryStore) IsCompensated(_ context.Context, correlationID, stepName string) (bool, error) + func (m *MemoryStore) ListIncompleteSagas(_ context.Context) ([]*SagaState, error) + func (m *MemoryStore) MarkCompensated(_ context.Context, correlationID, stepName string) error + func (m *MemoryStore) SetPending(_ context.Context, correlationID string, state *PendingState) error + func (m *MemoryStore) SetSaga(_ context.Context, correlationID string, state *SagaState) error + func (m *MemoryStore) UpdateSaga(_ context.Context, correlationID string, state *SagaState) error + func (m *MemoryStore) WithSaga(_ context.Context, correlationID string, ...) error + type PendingState struct + CorrelationID string + CreatedAt time.Time + Signal capitan.Signal + Timeout time.Duration + type Request struct + func NewRequest[T, R any](name pipz.Name, requestSignal capitan.Signal, responseSignal capitan.Signal, ...) *Request[T, R] + func (*Request[T, R]) Close() error + func (r *Request[T, R]) Build() pipz.Chainable[*Flow[T]] + func (r *Request[T, R]) Name() pipz.Name + func (r *Request[T, R]) Process(ctx context.Context, f *Flow[T]) (*Flow[T], error) + func (r *Request[T, R]) Timeout(d time.Duration) *Request[T, R] + func (r *Request[T, R]) WithCapitan(c *capitan.Capitan) *Request[T, R] + type SagaState struct + Compensations []CompensationRecord + CorrelationID string + CreatedAt time.Time + CurrentStep int + Error string + Status SagaStatus + Timeout time.Duration + UpdatedAt time.Time + func (s *SagaState) IsExpired() bool + type SagaStatus string + const SagaStatusCompensating + const SagaStatusCompleted + const SagaStatusFailed + const SagaStatusPending + const SagaStatusRunning + type SagaStep struct + func NewSagaStep[T any](name pipz.Name, store Store, key capitan.GenericKey[T], execute capitan.Signal, ...) *SagaStep[T] + func (*SagaStep[T]) Close() error + func (s *SagaStep[T]) Build() pipz.Chainable[*Flow[T]] + func (s *SagaStep[T]) Name() pipz.Name + func (s *SagaStep[T]) Process(ctx context.Context, f *Flow[T]) (*Flow[T], error) + func (s *SagaStep[T]) WithCapitan(c *capitan.Capitan) *SagaStep[T] + func (s *SagaStep[T]) WithTimeout(d time.Duration) *SagaStep[T] + type Store interface + DeletePending func(ctx context.Context, correlationID string) error + DeleteSaga func(ctx context.Context, correlationID string) error + GetPending func(ctx context.Context, correlationID string) (*PendingState, error) + GetSaga func(ctx context.Context, correlationID string) (*SagaState, error) + IsCompensated func(ctx context.Context, correlationID, stepName string) (bool, error) + ListIncompleteSagas func(ctx context.Context) ([]*SagaState, error) + MarkCompensated func(ctx context.Context, correlationID, stepName string) error + SetPending func(ctx context.Context, correlationID string, state *PendingState) error + SetSaga func(ctx context.Context, correlationID string, state *SagaState) error + UpdateSaga func(ctx context.Context, correlationID string, state *SagaState) error + WithSaga func(ctx context.Context, correlationID string, ...) error
Go to main page