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).
79 lines
1.9 KiB
Go
79 lines
1.9 KiB
Go
package matching
|
|
|
|
import (
|
|
"time"
|
|
|
|
taskqueuepb "go.temporal.io/api/taskqueue/v1"
|
|
"github.com/hanzoai/tasks/common/cache"
|
|
"google.golang.org/protobuf/types/known/timestamppb"
|
|
"google.golang.org/protobuf/types/known/wrapperspb"
|
|
)
|
|
|
|
const (
|
|
pollerHistoryInitMaxSize = 1000
|
|
)
|
|
|
|
type (
|
|
pollerIdentity string
|
|
|
|
pollerInfo struct {
|
|
pollMetadata
|
|
}
|
|
)
|
|
|
|
type pollerHistory struct {
|
|
// poller ID -> pollerInfo
|
|
// pollers map[pollerID]pollerInfo
|
|
history cache.Cache
|
|
}
|
|
|
|
func newPollerHistory(pollerHistoryTTL time.Duration) *pollerHistory {
|
|
opts := &cache.Options{
|
|
TTL: pollerHistoryTTL,
|
|
Pin: false,
|
|
}
|
|
|
|
return &pollerHistory{
|
|
history: cache.New(pollerHistoryInitMaxSize, opts),
|
|
}
|
|
}
|
|
|
|
func (pollers *pollerHistory) updatePollerInfo(id pollerIdentity, pollMetadata *pollMetadata) {
|
|
pollers.history.Put(id, &pollerInfo{pollMetadata: *pollMetadata})
|
|
}
|
|
|
|
func (pollers *pollerHistory) removePoller(id pollerIdentity) {
|
|
pollers.history.Delete(id)
|
|
}
|
|
|
|
func (pollers *pollerHistory) getPollerInfo(earliestAccessTime time.Time) []*taskqueuepb.PollerInfo {
|
|
var result []*taskqueuepb.PollerInfo
|
|
|
|
ite := pollers.history.Iterator()
|
|
defer ite.Close()
|
|
for ite.HasNext() {
|
|
entry := ite.Next()
|
|
key := entry.Key().(pollerIdentity)
|
|
value := entry.Value().(*pollerInfo)
|
|
lastAccessTime := entry.CreateTime()
|
|
if earliestAccessTime.Before(lastAccessTime) {
|
|
result = append(result, &taskqueuepb.PollerInfo{
|
|
Identity: string(key),
|
|
LastAccessTime: timestamppb.New(lastAccessTime),
|
|
RatePerSecond: defaultRPS(value.taskQueueMetadata.GetMaxTasksPerSecond()),
|
|
WorkerVersionCapabilities: value.workerVersionCapabilities,
|
|
DeploymentOptions: value.deploymentOptions,
|
|
})
|
|
}
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
func defaultRPS(wrapper *wrapperspb.DoubleValue) float64 {
|
|
if wrapper != nil {
|
|
return wrapper.Value
|
|
}
|
|
return defaultTaskDispatchRPS
|
|
}
|