mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-03 03:56:04 +00:00
zeekay
5a601d081b
Full module path rebrand from Temporal to Hanzo Tasks: - go.mod module: github.com/hanzoai/tasks - All internal Go imports updated (1906 files) - cmd/server renamed to cmd/tasksd - Binary name: tasksd (was temporal-server) - Docker images: hanzoai/tasks (was temporalio/server) - README rewritten with Hanzo Tasks branding - LLM.md created with integration notes - Makefile, goreleaser, CI workflows updated - go.temporal.io/sdk and go.temporal.io/api imports preserved (separate repos) Fork of Temporal.io (MIT license).
113 lines
3.9 KiB
Go
113 lines
3.9 KiB
Go
package queues
|
|
|
|
import (
|
|
"go.opentelemetry.io/otel/trace"
|
|
"github.com/hanzoai/tasks/chasm"
|
|
"github.com/hanzoai/tasks/common/clock"
|
|
"github.com/hanzoai/tasks/common/cluster"
|
|
"github.com/hanzoai/tasks/common/dynamicconfig"
|
|
"github.com/hanzoai/tasks/common/log"
|
|
"github.com/hanzoai/tasks/common/metrics"
|
|
"github.com/hanzoai/tasks/common/namespace"
|
|
"github.com/hanzoai/tasks/service/history/tasks"
|
|
)
|
|
|
|
type (
|
|
// ExecutableFactory and the related interfaces here are needed to make it possible to extend the functionality of
|
|
// task processing without changing its core logic. This is similar to ExecutorWrapper.
|
|
ExecutableFactory interface {
|
|
NewExecutable(task tasks.Task, readerID int64) Executable
|
|
}
|
|
// ExecutableFactoryFn is a convenience type to avoid having to create a struct that implements ExecutableFactory.
|
|
ExecutableFactoryFn func(readerID int64, t tasks.Task) Executable
|
|
executableFactoryImpl struct {
|
|
executor Executor
|
|
scheduler Scheduler
|
|
rescheduler Rescheduler
|
|
priorityAssigner PriorityAssigner
|
|
timeSource clock.TimeSource
|
|
namespaceRegistry namespace.Registry
|
|
clusterMetadata cluster.Metadata
|
|
chasmRegistry *chasm.Registry
|
|
taskTypeTagProvider TaskTypeTagProvider
|
|
logger log.Logger
|
|
metricsHandler metrics.Handler
|
|
tracer trace.Tracer
|
|
dlqWriter *DLQWriter
|
|
dlqEnabled dynamicconfig.BoolPropertyFn
|
|
attemptsBeforeSendingToDlq dynamicconfig.IntPropertyFn
|
|
dlqInternalErrors dynamicconfig.BoolPropertyFn
|
|
dlqErrorPattern dynamicconfig.StringPropertyFn
|
|
}
|
|
)
|
|
|
|
func (f ExecutableFactoryFn) NewExecutable(task tasks.Task, readerID int64) Executable {
|
|
return f(readerID, task)
|
|
}
|
|
|
|
func NewExecutableFactory(
|
|
executor Executor,
|
|
scheduler Scheduler,
|
|
rescheduler Rescheduler,
|
|
priorityAssigner PriorityAssigner,
|
|
timeSource clock.TimeSource,
|
|
namespaceRegistry namespace.Registry,
|
|
clusterMetadata cluster.Metadata,
|
|
chasmRegistry *chasm.Registry,
|
|
taskTypeTagProvider TaskTypeTagProvider,
|
|
logger log.Logger,
|
|
metricsHandler metrics.Handler,
|
|
tracer trace.Tracer,
|
|
dlqWriter *DLQWriter,
|
|
dlqEnabled dynamicconfig.BoolPropertyFn,
|
|
attemptsBeforeSendingToDlq dynamicconfig.IntPropertyFn,
|
|
dlqInternalErrors dynamicconfig.BoolPropertyFn,
|
|
dlqErrorPattern dynamicconfig.StringPropertyFn,
|
|
) *executableFactoryImpl {
|
|
return &executableFactoryImpl{
|
|
executor: executor,
|
|
scheduler: scheduler,
|
|
rescheduler: rescheduler,
|
|
priorityAssigner: priorityAssigner,
|
|
timeSource: timeSource,
|
|
namespaceRegistry: namespaceRegistry,
|
|
clusterMetadata: clusterMetadata,
|
|
chasmRegistry: chasmRegistry,
|
|
taskTypeTagProvider: taskTypeTagProvider,
|
|
logger: logger,
|
|
metricsHandler: metricsHandler.WithTags(defaultExecutableMetricsTags...),
|
|
tracer: tracer,
|
|
dlqWriter: dlqWriter,
|
|
dlqEnabled: dlqEnabled,
|
|
attemptsBeforeSendingToDlq: attemptsBeforeSendingToDlq,
|
|
dlqInternalErrors: dlqInternalErrors,
|
|
dlqErrorPattern: dlqErrorPattern,
|
|
}
|
|
}
|
|
|
|
func (f *executableFactoryImpl) NewExecutable(task tasks.Task, readerID int64) Executable {
|
|
return NewExecutable(
|
|
readerID,
|
|
task,
|
|
f.executor,
|
|
f.scheduler,
|
|
f.rescheduler,
|
|
f.priorityAssigner,
|
|
f.timeSource,
|
|
f.namespaceRegistry,
|
|
f.clusterMetadata,
|
|
f.chasmRegistry,
|
|
f.taskTypeTagProvider,
|
|
f.logger,
|
|
f.metricsHandler,
|
|
f.tracer,
|
|
func(params *ExecutableParams) {
|
|
params.DLQEnabled = f.dlqEnabled
|
|
params.DLQWriter = f.dlqWriter
|
|
params.MaxUnexpectedErrorAttempts = f.attemptsBeforeSendingToDlq
|
|
params.DLQInternalErrors = f.dlqInternalErrors
|
|
params.DLQErrorPattern = f.dlqErrorPattern
|
|
},
|
|
)
|
|
}
|