mirror of
https://github.com/hanzoai/team.git
synced 2026-11-01 07:55:31 +00:00
zeekay
8feb24ea1f
The server module named github.com/hanzoai/team/server/v8 — a path that resolves to the retired workspace's repository, not this one — while server/public still published github.com/mattermost/mattermost/server/public. Both are github.com/hanzoteam/server/... now, which is where this code actually lives. That second half is what made the binary unbootable. Upstream's public path stayed claimed by upstream, mattermost-plugin-ai imported it, and the build linked public/model twice; public/plugin's gob.Register then ran twice under one name and panicked at init, before main. Reverting the rename traded the brand for a working binary. Rebranding the plugin alongside it needs no such trade: the agents fork is github.com/hanzoteam/agents/v2 against our public module, so no module named mattermost/mattermost is in the graph at all. Measured: `go list -deps ./cmd/mattermost` names exactly one public/model, and it is ours.
87 lines
2.1 KiB
Go
87 lines
2.1 KiB
Go
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
|
// See LICENSE.txt for license information.
|
|
|
|
package jobs
|
|
|
|
import (
|
|
"math/rand"
|
|
"time"
|
|
|
|
"github.com/hanzoteam/server/server/public/model"
|
|
"github.com/hanzoteam/server/server/public/shared/mlog"
|
|
"github.com/hanzoteam/server/server/public/shared/request"
|
|
)
|
|
|
|
// Default polling interval for jobs termination.
|
|
// (Defining as `var` rather than `const` allows tests to lower the interval.)
|
|
var DefaultWatcherPollingInterval = 15000
|
|
|
|
type Watcher struct {
|
|
srv *JobServer
|
|
workers *Workers
|
|
|
|
stop chan struct{}
|
|
stopped chan struct{}
|
|
pollingInterval int
|
|
}
|
|
|
|
func (srv *JobServer) MakeWatcher(workers *Workers, pollingInterval int) *Watcher {
|
|
return &Watcher{
|
|
stop: make(chan struct{}),
|
|
stopped: make(chan struct{}),
|
|
pollingInterval: pollingInterval,
|
|
workers: workers,
|
|
srv: srv,
|
|
}
|
|
}
|
|
|
|
func (watcher *Watcher) Start() {
|
|
mlog.Debug("Watcher Started")
|
|
// Delay for some random number of milliseconds before starting to ensure that multiple
|
|
// instances of the jobserver don't poll at a time too close to each other.
|
|
rand.Seed(time.Now().UTC().UnixNano())
|
|
<-time.After(time.Duration(rand.Intn(watcher.pollingInterval)) * time.Millisecond)
|
|
|
|
defer func() {
|
|
mlog.Debug("Watcher Finished")
|
|
close(watcher.stopped)
|
|
}()
|
|
|
|
for {
|
|
select {
|
|
case <-watcher.stop:
|
|
mlog.Debug("Watcher: Received stop signal")
|
|
return
|
|
case <-time.After(time.Duration(watcher.pollingInterval) * time.Millisecond):
|
|
watcher.PollAndNotify()
|
|
}
|
|
}
|
|
}
|
|
|
|
func (watcher *Watcher) Stop() {
|
|
mlog.Debug("Watcher Stopping")
|
|
close(watcher.stop)
|
|
<-watcher.stopped
|
|
|
|
watcher.stop = make(chan struct{})
|
|
watcher.stopped = make(chan struct{})
|
|
}
|
|
|
|
func (watcher *Watcher) PollAndNotify() {
|
|
jobs, err := watcher.srv.Store.Job().GetAllByStatus(request.EmptyContext(watcher.srv.logger), model.JobStatusPending)
|
|
if err != nil {
|
|
mlog.Error("Error occurred getting all pending statuses.", mlog.Err(err))
|
|
return
|
|
}
|
|
|
|
for _, job := range jobs {
|
|
worker := watcher.workers.Get(job.Type)
|
|
if worker != nil {
|
|
select {
|
|
case worker.JobChannel() <- *job:
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
}
|