drsybren/Flamenco
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Manager: Adjust code for renaming SocketIO... types to Event...

Browse Source 
No functional changes, just adjusting to the OpenAPI renames.
This commit is contained in:
Sybren A. Stüvel 2024-02-05 09:16:16 +01:00
parent 1ed893fa84
commit e7c4285ac6
36 changed files with 118 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
internal/manager/api_impl/interfaces.go
View File

@ -106,9 +106,9 @@ var _ TaskStateMachine = (*task_state_machine.StateMachine)(nil)
type ChangeBroadcaster interface {
// BroadcastNewJob sends a 'new job' notification to all SocketIO clients.
BroadcastNewJob(jobUpdate api.SocketIOJobUpdate)
BroadcastJobUpdate(jobUpdate api.SocketIOJobUpdate)
BroadcastLastRenderedImage(update api.SocketIOLastRenderedUpdate)
BroadcastNewJob(jobUpdate api.EventJobUpdate)
BroadcastJobUpdate(jobUpdate api.EventJobUpdate)
BroadcastLastRenderedImage(update api.EventLastRenderedUpdate)
// Note that there is no BroadcastNewTask. The 'new job' broadcast is sent
// after the job's tasks have been created, and thus there is no need for a
@ -118,11 +118,11 @@ type ChangeBroadcaster interface {
// responsibility of `LogStorage.Write` to broadcast the changes to SocketIO
// clients.
BroadcastWorkerUpdate(workerUpdate api.SocketIOWorkerUpdate)
BroadcastNewWorker(workerUpdate api.SocketIOWorkerUpdate)
BroadcastWorkerUpdate(workerUpdate api.EventWorkerUpdate)
BroadcastNewWorker(workerUpdate api.EventWorkerUpdate)
BroadcastWorkerTagUpdate(workerTagUpdate api.SocketIOWorkerTagUpdate)
BroadcastNewWorkerTag(workerTagUpdate api.SocketIOWorkerTagUpdate)
BroadcastWorkerTagUpdate(workerTagUpdate api.EventWorkerTagUpdate)
BroadcastNewWorkerTag(workerTagUpdate api.EventWorkerTagUpdate)
}
// ChangeBroadcaster should be a subset of eventbus.Broker.

10
internal/manager/api_impl/jobs_test.go
View File

@ -74,7 +74,7 @@ func TestSubmitJobWithoutSettings(t *testing.T) {
mf.persistence.EXPECT().FetchJob(gomock.Any(), queuedJob.JobID).Return(&dbJob, nil)
// Expect the new job to be broadcast.
jobUpdate := api.SocketIOJobUpdate{
jobUpdate := api.EventJobUpdate{
Id: dbJob.UUID,
Name: &dbJob.Name,
Priority: dbJob.Priority,
@ -163,7 +163,7 @@ func TestSubmitJobWithSettings(t *testing.T) {
mf.persistence.EXPECT().FetchJob(gomock.Any(), queuedJob.JobID).Return(&dbJob, nil)
// Expect the new job to be broadcast.
jobUpdate := api.SocketIOJobUpdate{
jobUpdate := api.EventJobUpdate{
Id: dbJob.UUID,
Name: &dbJob.Name,
Priority: dbJob.Priority,
@ -304,7 +304,7 @@ func TestSubmitJobWithShamanCheckoutID(t *testing.T) {
mf.persistence.EXPECT().FetchJob(gomock.Any(), queuedJob.JobID).Return(&dbJob, nil)
// Expect the new job to be broadcast.
jobUpdate := api.SocketIOJobUpdate{
jobUpdate := api.EventJobUpdate{
Id: dbJob.UUID,
Name: &dbJob.Name,
Priority: dbJob.Priority,
@ -389,7 +389,7 @@ func TestSubmitJobWithWorkerTag(t *testing.T) {
mf.persistence.EXPECT().FetchJob(gomock.Any(), queuedJob.JobID).Return(&dbJob, nil)
// Expect the new job to be broadcast.
jobUpdate := api.SocketIOJobUpdate{
jobUpdate := api.EventJobUpdate{
Id: dbJob.UUID,
Name: &dbJob.Name,
Priority: dbJob.Priority,
@ -623,7 +623,7 @@ func TestSetJobPrio(t *testing.T) {
mf.persistence.EXPECT().SaveJobPriority(gomock.Not(ctx), &jobWithNewPrio)
// Expect the change to be broadcast over SocketIO.
expectUpdate := api.SocketIOJobUpdate{
expectUpdate := api.EventJobUpdate{
Id: dbJob.UUID,
Name: &dbJob.Name,
RefreshTasks: false,

14
internal/manager/api_impl/mocks/api_impl_mock.gen.go generated
View File

@ -586,7 +586,7 @@ func (m *MockChangeBroadcaster) EXPECT() *MockChangeBroadcasterMockRecorder {
}
// BroadcastJobUpdate mocks base method.
func (m *MockChangeBroadcaster) BroadcastJobUpdate(arg0 api.SocketIOJobUpdate) {
func (m *MockChangeBroadcaster) BroadcastJobUpdate(arg0 api.EventJobUpdate) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "BroadcastJobUpdate", arg0)
}
@ -598,7 +598,7 @@ func (mr *MockChangeBroadcasterMockRecorder) BroadcastJobUpdate(arg0 interface{}
}
// BroadcastLastRenderedImage mocks base method.
func (m *MockChangeBroadcaster) BroadcastLastRenderedImage(arg0 api.SocketIOLastRenderedUpdate) {
func (m *MockChangeBroadcaster) BroadcastLastRenderedImage(arg0 api.EventLastRenderedUpdate) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "BroadcastLastRenderedImage", arg0)
}
@ -610,7 +610,7 @@ func (mr *MockChangeBroadcasterMockRecorder) BroadcastLastRenderedImage(arg0 int
}
// BroadcastNewJob mocks base method.
func (m *MockChangeBroadcaster) BroadcastNewJob(arg0 api.SocketIOJobUpdate) {
func (m *MockChangeBroadcaster) BroadcastNewJob(arg0 api.EventJobUpdate) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "BroadcastNewJob", arg0)
}
@ -622,7 +622,7 @@ func (mr *MockChangeBroadcasterMockRecorder) BroadcastNewJob(arg0 interface{}) *
}
// BroadcastNewWorker mocks base method.
func (m *MockChangeBroadcaster) BroadcastNewWorker(arg0 api.SocketIOWorkerUpdate) {
func (m *MockChangeBroadcaster) BroadcastNewWorker(arg0 api.EventWorkerUpdate) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "BroadcastNewWorker", arg0)
}
@ -634,7 +634,7 @@ func (mr *MockChangeBroadcasterMockRecorder) BroadcastNewWorker(arg0 interface{}
}
// BroadcastNewWorkerTag mocks base method.
func (m *MockChangeBroadcaster) BroadcastNewWorkerTag(arg0 api.SocketIOWorkerTagUpdate) {
func (m *MockChangeBroadcaster) BroadcastNewWorkerTag(arg0 api.EventWorkerTagUpdate) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "BroadcastNewWorkerTag", arg0)
}
@ -646,7 +646,7 @@ func (mr *MockChangeBroadcasterMockRecorder) BroadcastNewWorkerTag(arg0 interfac
}
// BroadcastWorkerTagUpdate mocks base method.
func (m *MockChangeBroadcaster) BroadcastWorkerTagUpdate(arg0 api.SocketIOWorkerTagUpdate) {
func (m *MockChangeBroadcaster) BroadcastWorkerTagUpdate(arg0 api.EventWorkerTagUpdate) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "BroadcastWorkerTagUpdate", arg0)
}
@ -658,7 +658,7 @@ func (mr *MockChangeBroadcasterMockRecorder) BroadcastWorkerTagUpdate(arg0 inter
}
// BroadcastWorkerUpdate mocks base method.
func (m *MockChangeBroadcaster) BroadcastWorkerUpdate(arg0 api.SocketIOWorkerUpdate) {
func (m *MockChangeBroadcaster) BroadcastWorkerUpdate(arg0 api.EventWorkerUpdate) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "BroadcastWorkerUpdate", arg0)
}

16
internal/manager/api_impl/worker_mgt_test.go
View File

@ -165,7 +165,7 @@ func TestDeleteWorker(t *testing.T) {
mf.persistence.EXPECT().DeleteWorker(gomock.Any(), workerUUID).Return(nil)
mockedNow := mf.clock.Now()
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.SocketIOWorkerUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
DeletedAt: &mockedNow,
Id: worker.UUID,
Name: worker.Name,
@ -197,7 +197,7 @@ func TestRequestWorkerStatusChange(t *testing.T) {
mf.persistence.EXPECT().SaveWorker(gomock.Any(), &savedWorker).Return(nil)
// Expect a broadcast of the change
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.SocketIOWorkerUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
Id: worker.UUID,
Name: worker.Name,
Status: prevStatus,
@ -241,7 +241,7 @@ func TestRequestWorkerStatusChangeRevert(t *testing.T) {
mf.persistence.EXPECT().SaveWorker(gomock.Any(), &savedWorker).Return(nil)
// Expect a broadcast of the change
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.SocketIOWorkerUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
Id: worker.UUID,
Name: worker.Name,
Status: currentStatus,
@ -281,7 +281,7 @@ func TestWorkerTagCRUDHappyFlow(t *testing.T) {
Description: *apiTag.Description,
}
mf.persistence.EXPECT().CreateWorkerTag(gomock.Any(), &expectDBTag)
mf.broadcaster.EXPECT().BroadcastNewWorkerTag(api.SocketIOWorkerTagUpdate{
mf.broadcaster.EXPECT().BroadcastNewWorkerTag(api.EventWorkerTagUpdate{
Tag: apiTag,
})
echo := mf.prepareMockedJSONRequest(apiTag)
@ -305,7 +305,7 @@ func TestWorkerTagCRUDHappyFlow(t *testing.T) {
Name: newAPITag.Name,
Description: *apiTag.Description, // Not mentioning new description should keep old one.
}
mf.broadcaster.EXPECT().BroadcastWorkerTagUpdate(api.SocketIOWorkerTagUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerTagUpdate(api.EventWorkerTagUpdate{
Tag: api.WorkerTag{
Id: &UUID,
Name: newAPITag.Name,
@ -328,7 +328,7 @@ func TestWorkerTagCRUDHappyFlow(t *testing.T) {
Name: newAPITag.Name,
Description: "",
}
mf.broadcaster.EXPECT().BroadcastWorkerTagUpdate(api.SocketIOWorkerTagUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerTagUpdate(api.EventWorkerTagUpdate{
Tag: api.WorkerTag{
Id: &UUID,
Name: newAPITag.Name,
@ -351,7 +351,7 @@ func TestWorkerTagCRUDHappyFlow(t *testing.T) {
Name: newAPITag.Name,
Description: *newAPITag.Description,
}
mf.broadcaster.EXPECT().BroadcastWorkerTagUpdate(api.SocketIOWorkerTagUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerTagUpdate(api.EventWorkerTagUpdate{
Tag: api.WorkerTag{
Id: &UUID,
Name: newAPITag.Name,
@ -367,7 +367,7 @@ func TestWorkerTagCRUDHappyFlow(t *testing.T) {
// Delete.
mf.persistence.EXPECT().FetchWorkerTag(gomock.Any(), UUID).Return(&expectDBTag, nil)
mf.persistence.EXPECT().DeleteWorkerTag(gomock.Any(), UUID)
mf.broadcaster.EXPECT().BroadcastWorkerTagUpdate(api.SocketIOWorkerTagUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerTagUpdate(api.EventWorkerTagUpdate{
Tag: api.WorkerTag{Id: &UUID},
WasDeleted: ptr(true),
})

16
internal/manager/api_impl/workers_test.go
View File

@ -187,7 +187,7 @@ func TestWorkerSignOn(t *testing.T) {
mf.sleepScheduler.EXPECT().WorkerStatus(gomock.Any(), worker.UUID).
Return(api.WorkerStatusAsleep, nil)
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.SocketIOWorkerUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
Id: worker.UUID,
Name: "Lazy Boi",
PreviousStatus: &prevStatus,
@ -239,7 +239,7 @@ func TestWorkerSignoffTaskRequeue(t *testing.T) {
})
prevStatus := api.WorkerStatusAwake
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.SocketIOWorkerUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
Id: worker.UUID,
Name: worker.Name,
PreviousStatus: &prevStatus,
@ -268,7 +268,7 @@ func TestWorkerRememberPreviousStatus(t *testing.T) {
worker.Status = api.WorkerStatusAwake
worker.StatusChangeRequest(api.WorkerStatusOffline, true)
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.SocketIOWorkerUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
Id: worker.UUID,
Name: worker.Name,
PreviousStatus: ptr(api.WorkerStatusAwake),
@ -309,7 +309,7 @@ func TestWorkerDontRememberPreviousStatus(t *testing.T) {
worker.Status = api.WorkerStatusError
worker.StatusChangeRequest(api.WorkerStatusOffline, true)
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.SocketIOWorkerUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
Id: worker.UUID,
Name: worker.Name,
PreviousStatus: ptr(api.WorkerStatusError),
@ -379,7 +379,7 @@ func TestWorkerStateChanged(t *testing.T) {
prevStatus := worker.Status
// Expect a broadcast of the change
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.SocketIOWorkerUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
Id: worker.UUID,
Name: worker.Name,
PreviousStatus: &prevStatus,
@ -419,7 +419,7 @@ func TestWorkerStateChangedAfterChangeRequest(t *testing.T) {
// Expect a broadcast of the change, even though it's not the state that was requested.
// This is to allow some flexibility, for example when a worker has to go
// asleep but would do so via `offline → starting → asleep`.
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.SocketIOWorkerUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
Id: worker.UUID,
Name: worker.Name,
PreviousStatus: ptr(api.WorkerStatusOffline),
@ -452,7 +452,7 @@ func TestWorkerStateChangedAfterChangeRequest(t *testing.T) {
// Do another status change, which does meet the requested state.
{
// Expect a broadcast.
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.SocketIOWorkerUpdate{
mf.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
Id: worker.UUID,
Name: worker.Name,
PreviousStatus: ptr(api.WorkerStatusStarting),
@ -694,7 +694,7 @@ func TestTaskOutputProduced(t *testing.T) {
ctx := context.Background()
mf.persistence.EXPECT().SetLastRendered(ctx, &job)
expectBroadcast := api.SocketIOLastRenderedUpdate{
expectBroadcast := api.EventLastRenderedUpdate{
JobId: job.UUID,
Thumbnail: api.JobLastRenderedImageInfo{
Base: "/job-files/relative/path/to/job",

32
internal/manager/eventbus/events_jobs.go
View File

@ -9,12 +9,12 @@ import (
"projects.blender.org/studio/flamenco/pkg/api"
)
// NewJobUpdate returns a partial SocketIOJobUpdate struct for the given job.
// NewJobUpdate returns a partial EventJobUpdate struct for the given job.
// It only fills in the fields that represent the current state of the job. For
// example, it omits `PreviousStatus`. The ommitted fields can be filled in by
// the caller.
func NewJobUpdate(job *persistence.Job) api.SocketIOJobUpdate {
jobUpdate := api.SocketIOJobUpdate{
func NewJobUpdate(job *persistence.Job) api.EventJobUpdate {
jobUpdate := api.EventJobUpdate{
Id: job.UUID,
Name: &job.Name,
Updated: job.UpdatedAt,
@ -36,8 +36,8 @@ func NewJobUpdate(job *persistence.Job) api.SocketIOJobUpdate {
// the caller.
//
// Assumes task.Job is not nil.
func NewTaskUpdate(task *persistence.Task) api.SocketIOTaskUpdate {
taskUpdate := api.SocketIOTaskUpdate{
func NewTaskUpdate(task *persistence.Task) api.EventTaskUpdate {
taskUpdate := api.EventTaskUpdate{
Id: task.UUID,
JobId: task.Job.UUID,
Name: task.Name,
@ -48,18 +48,18 @@ func NewTaskUpdate(task *persistence.Task) api.SocketIOTaskUpdate {
return taskUpdate
}
// NewLastRenderedUpdate returns a partial SocketIOLastRenderedUpdate struct.
// NewLastRenderedUpdate returns a partial EventLastRenderedUpdate struct.
// The `Thumbnail` field still needs to be filled in, but that requires
// information from the `api_impl.Flamenco` service.
func NewLastRenderedUpdate(jobUUID string) api.SocketIOLastRenderedUpdate {
return api.SocketIOLastRenderedUpdate{
func NewLastRenderedUpdate(jobUUID string) api.EventLastRenderedUpdate {
return api.EventLastRenderedUpdate{
JobId: jobUUID,
}
}
// NewTaskLogUpdate returns a SocketIOTaskLogUpdate for the given task.
func NewTaskLogUpdate(taskUUID string, logchunk string) api.SocketIOTaskLogUpdate {
return api.SocketIOTaskLogUpdate{
// NewTaskLogUpdate returns a EventTaskLogUpdate for the given task.
func NewTaskLogUpdate(taskUUID string, logchunk string) api.EventTaskLogUpdate {
return api.EventTaskLogUpdate{
TaskId: taskUUID,
Log: logchunk,
}
@ -68,7 +68,7 @@ func NewTaskLogUpdate(taskUUID string, logchunk string) api.SocketIOTaskLogUpdat
// BroadcastNewJob sends a "new job" notification to clients.
// This function should be called when the job has been completely created, so
// including its tasks.
func (b *Broker) BroadcastNewJob(jobUpdate api.SocketIOJobUpdate) {
func (b *Broker) BroadcastNewJob(jobUpdate api.EventJobUpdate) {
if jobUpdate.PreviousStatus != nil {
log.Warn().Interface("jobUpdate", jobUpdate).Msg("eventbus: new jobs should not have a previous state")
jobUpdate.PreviousStatus = nil
@ -78,12 +78,12 @@ func (b *Broker) BroadcastNewJob(jobUpdate api.SocketIOJobUpdate) {
b.broadcast(TopicJobUpdate, jobUpdate)
}
func (b *Broker) BroadcastJobUpdate(jobUpdate api.SocketIOJobUpdate) {
func (b *Broker) BroadcastJobUpdate(jobUpdate api.EventJobUpdate) {
log.Debug().Interface("jobUpdate", jobUpdate).Msg("eventbus: broadcasting job update")
b.broadcast(TopicJobUpdate, jobUpdate)
}
func (b *Broker) BroadcastLastRenderedImage(update api.SocketIOLastRenderedUpdate) {
func (b *Broker) BroadcastLastRenderedImage(update api.EventLastRenderedUpdate) {
log.Debug().Interface("lastRenderedUpdate", update).Msg("eventbus: broadcasting last-rendered image update")
topic := topicForJobLastRendered(update.JobId)
b.broadcast(topic, update)
@ -92,13 +92,13 @@ func (b *Broker) BroadcastLastRenderedImage(update api.SocketIOLastRenderedUpdat
b.broadcast(TopicLastRenderedImage, update)
}
func (b *Broker) BroadcastTaskUpdate(taskUpdate api.SocketIOTaskUpdate) {
func (b *Broker) BroadcastTaskUpdate(taskUpdate api.EventTaskUpdate) {
log.Debug().Interface("taskUpdate", taskUpdate).Msg("eventbus: broadcasting task update")
topic := topicForJob(taskUpdate.JobId)
b.broadcast(topic, taskUpdate)
}
func (b *Broker) BroadcastTaskLogUpdate(taskLogUpdate api.SocketIOTaskLogUpdate) {
func (b *Broker) BroadcastTaskLogUpdate(taskLogUpdate api.EventTaskLogUpdate) {
// Don't log the contents here; logs can get big.
topic := topicForTaskLog(taskLogUpdate.TaskId)
log.Debug().

10
internal/manager/eventbus/events_workers.go
View File

@ -8,12 +8,12 @@ import (
"projects.blender.org/studio/flamenco/pkg/api"
)
// NewWorkerUpdate returns a partial SocketIOWorkerUpdate struct for the given worker.
// NewWorkerUpdate returns a partial EventWorkerUpdate struct for the given worker.
// It only fills in the fields that represent the current state of the worker. For
// example, it omits `PreviousStatus`. The ommitted fields can be filled in by
// the caller.
func NewWorkerUpdate(worker *persistence.Worker) api.SocketIOWorkerUpdate {
workerUpdate := api.SocketIOWorkerUpdate{
func NewWorkerUpdate(worker *persistence.Worker) api.EventWorkerUpdate {
workerUpdate := api.EventWorkerUpdate{
Id: worker.UUID,
Name: worker.Name,
Status: worker.Status,
@ -38,7 +38,7 @@ func NewWorkerUpdate(worker *persistence.Worker) api.SocketIOWorkerUpdate {
return workerUpdate
}
func (b *Broker) BroadcastNewWorker(workerUpdate api.SocketIOWorkerUpdate) {
func (b *Broker) BroadcastNewWorker(workerUpdate api.EventWorkerUpdate) {
if workerUpdate.PreviousStatus != nil {
log.Warn().Interface("workerUpdate", workerUpdate).Msg("eventbus: new workers should not have a previous state")
workerUpdate.PreviousStatus = nil
@ -48,7 +48,7 @@ func (b *Broker) BroadcastNewWorker(workerUpdate api.SocketIOWorkerUpdate) {
b.broadcast(TopicWorkerUpdate, workerUpdate)
}
func (b *Broker) BroadcastWorkerUpdate(workerUpdate api.SocketIOWorkerUpdate) {
func (b *Broker) BroadcastWorkerUpdate(workerUpdate api.EventWorkerUpdate) {
log.Debug().Interface("workerUpdate", workerUpdate).Msg("eventbus: broadcasting worker update")
b.broadcast(TopicWorkerUpdate, workerUpdate)
}

16
internal/manager/eventbus/events_workertags.go
View File

@ -8,11 +8,11 @@ import (
"projects.blender.org/studio/flamenco/pkg/api"
)
// NewWorkerTagUpdate returns a partial SocketIOWorkerTagUpdate struct for the
// NewWorkerTagUpdate returns a partial EventWorkerTagUpdate struct for the
// given worker tag. It only fills in the fields that represent the current
// state of the tag.
func NewWorkerTagUpdate(tag *persistence.WorkerTag) api.SocketIOWorkerTagUpdate {
tagUpdate := api.SocketIOWorkerTagUpdate{
func NewWorkerTagUpdate(tag *persistence.WorkerTag) api.EventWorkerTagUpdate {
tagUpdate := api.EventWorkerTagUpdate{
Tag: api.WorkerTag{
Id: &tag.UUID,
Name: tag.Name,
@ -22,11 +22,11 @@ func NewWorkerTagUpdate(tag *persistence.WorkerTag) api.SocketIOWorkerTagUpdate
return tagUpdate
}
// NewWorkerTagDeletedUpdate returns a SocketIOWorkerTagUpdate struct that indicates
// NewWorkerTagDeletedUpdate returns a EventWorkerTagUpdate struct that indicates
// the worker tag has been deleted.
func NewWorkerTagDeletedUpdate(tagUUID string) api.SocketIOWorkerTagUpdate {
func NewWorkerTagDeletedUpdate(tagUUID string) api.EventWorkerTagUpdate {
wasDeleted := true
tagUpdate := api.SocketIOWorkerTagUpdate{
tagUpdate := api.EventWorkerTagUpdate{
Tag: api.WorkerTag{
Id: &tagUUID,
},
@ -35,12 +35,12 @@ func NewWorkerTagDeletedUpdate(tagUUID string) api.SocketIOWorkerTagUpdate {
return tagUpdate
}
func (b *Broker) BroadcastWorkerTagUpdate(workerTagUpdate api.SocketIOWorkerTagUpdate) {
func (b *Broker) BroadcastWorkerTagUpdate(workerTagUpdate api.EventWorkerTagUpdate) {
log.Debug().Interface("WorkerTagUpdate", workerTagUpdate).Msg("eventbus: broadcasting worker tag update")
b.broadcast(TopicWorkerTagUpdate, workerTagUpdate)
}
func (b *Broker) BroadcastNewWorkerTag(workerTagUpdate api.SocketIOWorkerTagUpdate) {
func (b *Broker) BroadcastNewWorkerTag(workerTagUpdate api.EventWorkerTagUpdate) {
log.Debug().Interface("WorkerTagUpdate", workerTagUpdate).Msg("eventbus: broadcasting new worker tag")
b.broadcast(TopicWorkerTagUpdate, workerTagUpdate)
}

12
internal/manager/eventbus/socketio.go
View File

@ -23,12 +23,12 @@ const (
)
var socketIOEventTypes = map[string]string{
reflect.TypeOf(api.SocketIOJobUpdate{}).Name(): "/jobs",
reflect.TypeOf(api.SocketIOTaskUpdate{}).Name(): "/task",
reflect.TypeOf(api.SocketIOLastRenderedUpdate{}).Name(): "/last-rendered",
reflect.TypeOf(api.SocketIOTaskLogUpdate{}).Name(): "/tasklog",
reflect.TypeOf(api.SocketIOWorkerTagUpdate{}).Name(): "/workertags",
reflect.TypeOf(api.SocketIOWorkerUpdate{}).Name(): "/workers",
reflect.TypeOf(api.EventJobUpdate{}).Name(): "/jobs",
reflect.TypeOf(api.EventTaskUpdate{}).Name(): "/task",
reflect.TypeOf(api.EventLastRenderedUpdate{}).Name(): "/last-rendered",
reflect.TypeOf(api.EventTaskLogUpdate{}).Name(): "/tasklog",
reflect.TypeOf(api.EventWorkerTagUpdate{}).Name(): "/workertags",
reflect.TypeOf(api.EventWorkerUpdate{}).Name(): "/workers",
}
// SocketIOForwarder is an event forwarder via SocketIO.

12
internal/manager/eventbus/topics.go
View File

@ -6,12 +6,12 @@ import "fmt"
const (
// Topics on which events are published.
TopicJobUpdate EventTopic = "/jobs" // sends api.SocketIOJobUpdate
TopicLastRenderedImage EventTopic = "/last-rendered" // sends api.SocketIOLastRenderedUpdate
TopicTaskUpdate EventTopic = "/task" // sends api.SocketIOTaskUpdate
TopicWorkerUpdate EventTopic = "/workers" // sends api.SocketIOWorkerUpdate
TopicWorkerTagUpdate EventTopic = "/workertags" // sends api.SocketIOWorkerTagUpdate
TopicSubscription EventTopic = "/subscription" // clients send api.SocketIOSubscription
TopicJobUpdate EventTopic = "/jobs" // sends api.EventJobUpdate
TopicLastRenderedImage EventTopic = "/last-rendered" // sends api.EventLastRenderedUpdate
TopicTaskUpdate EventTopic = "/task" // sends api.EventTaskUpdate
TopicWorkerUpdate EventTopic = "/workers" // sends api.EventWorkerUpdate
TopicWorkerTagUpdate EventTopic = "/workertags" // sends api.EventWorkerTagUpdate
TopicSubscription EventTopic = "/subscription" // clients send api.EventSubscription
// Parameterised topics.
TopicJobSpecific EventTopic = "/jobs/%s" // %s = job UUID

2
internal/manager/job_deleter/interfaces.go
View File

@ -41,7 +41,7 @@ var _ Storage = (*local_storage.StorageInfo)(nil)
type ChangeBroadcaster interface {
// BroadcastJobUpdate sends the job update to SocketIO clients.
BroadcastJobUpdate(jobUpdate api.SocketIOJobUpdate)
BroadcastJobUpdate(jobUpdate api.EventJobUpdate)
}
// ChangeBroadcaster should be a subset of eventbus.Broker

2
internal/manager/job_deleter/job_deleter.go
View File

@ -218,7 +218,7 @@ func (s *Service) deleteJob(ctx context.Context, jobUUID string) error {
// "was deleted" flag, because there's nothing else left. And I don't want to
// do a full database query for something we'll delete anyway.
wasDeleted := true
jobUpdate := api.SocketIOJobUpdate{
jobUpdate := api.EventJobUpdate{
Id: jobUUID,
WasDeleted: &wasDeleted,
}

2
internal/manager/job_deleter/mocks/interfaces_mock.gen.go generated
View File

@ -183,7 +183,7 @@ func (m *MockChangeBroadcaster) EXPECT() *MockChangeBroadcasterMockRecorder {
}
// BroadcastJobUpdate mocks base method.
func (m *MockChangeBroadcaster) BroadcastJobUpdate(arg0 api.SocketIOJobUpdate) {
func (m *MockChangeBroadcaster) BroadcastJobUpdate(arg0 api.EventJobUpdate) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "BroadcastJobUpdate", arg0)
}

2
internal/manager/sleep_scheduler/interfaces.go
View File

@ -30,7 +30,7 @@ var _ PersistenceService = (*persistence.DB)(nil)
// TODO: Refactor the way worker status changes are handled, so that this
// service doens't need to broadcast its own worker updates.
type ChangeBroadcaster interface {
BroadcastWorkerUpdate(workerUpdate api.SocketIOWorkerUpdate)
BroadcastWorkerUpdate(workerUpdate api.EventWorkerUpdate)
}
// ChangeBroadcaster should be a subset of eventbus.Broker.

2
internal/manager/sleep_scheduler/mocks/interfaces_mock.gen.go generated
View File

@ -146,7 +146,7 @@ func (m *MockChangeBroadcaster) EXPECT() *MockChangeBroadcasterMockRecorder {
}
// BroadcastWorkerUpdate mocks base method.
func (m *MockChangeBroadcaster) BroadcastWorkerUpdate(arg0 api.SocketIOWorkerUpdate) {
func (m *MockChangeBroadcaster) BroadcastWorkerUpdate(arg0 api.EventWorkerUpdate) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "BroadcastWorkerUpdate", arg0)
}

2
internal/manager/sleep_scheduler/sleep_scheduler.go
View File

@ -170,7 +170,7 @@ func (ss *SleepScheduler) updateWorkerStatus(
}
// Broadcast worker change via SocketIO
ss.broadcaster.BroadcastWorkerUpdate(api.SocketIOWorkerUpdate{
ss.broadcaster.BroadcastWorkerUpdate(api.EventWorkerUpdate{
Id: worker.UUID,
Name: worker.Name,
Status: worker.Status,

4
internal/manager/sleep_scheduler/sleep_scheduler_test.go
View File

@ -149,9 +149,9 @@ func TestApplySleepSchedule(t *testing.T) {
mocks.persist.EXPECT().SaveWorkerStatus(ctx, &savedWorker)
// Expect SocketIO broadcast.
var sioUpdate api.SocketIOWorkerUpdate
var sioUpdate api.EventWorkerUpdate
mocks.broadcaster.EXPECT().BroadcastWorkerUpdate(gomock.Any()).DoAndReturn(
func(workerUpdate api.SocketIOWorkerUpdate) {
func(workerUpdate api.EventWorkerUpdate) {
sioUpdate = workerUpdate
})

2
internal/manager/task_logs/mocks/interfaces_mock.gen.go generated
View File

@ -72,7 +72,7 @@ func (m *MockChangeBroadcaster) EXPECT() *MockChangeBroadcasterMockRecorder {
}
// BroadcastTaskLogUpdate mocks base method.
func (m *MockChangeBroadcaster) BroadcastTaskLogUpdate(arg0 api.SocketIOTaskLogUpdate) {
func (m *MockChangeBroadcaster) BroadcastTaskLogUpdate(arg0 api.EventTaskLogUpdate) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "BroadcastTaskLogUpdate", arg0)
}

2
internal/manager/task_logs/task_logs.go
View File

@ -44,7 +44,7 @@ type LocalStorage interface {
type ChangeBroadcaster interface {
// BroadcastTaskLogUpdate sends the task log update to SocketIO clients.
BroadcastTaskLogUpdate(taskLogUpdate api.SocketIOTaskLogUpdate)
BroadcastTaskLogUpdate(taskLogUpdate api.EventTaskLogUpdate)
}
// ChangeBroadcaster should be a subset of eventbus.Broker

4
internal/manager/task_state_machine/interfaces.go
View File

@ -43,10 +43,10 @@ var _ PersistenceService = (*persistence.DB)(nil)
type ChangeBroadcaster interface {
// BroadcastJobUpdate sends the job update to SocketIO clients.
BroadcastJobUpdate(jobUpdate api.SocketIOJobUpdate)
BroadcastJobUpdate(jobUpdate api.EventJobUpdate)
// BroadcastTaskUpdate sends the task update to SocketIO clients.
BroadcastTaskUpdate(jobUpdate api.SocketIOTaskUpdate)
BroadcastTaskUpdate(jobUpdate api.EventTaskUpdate)
}
// ChangeBroadcaster should be a subset of eventbus.Broker

4
internal/manager/task_state_machine/mocks/interfaces_mock.gen.go generated
View File

@ -231,7 +231,7 @@ func (m *MockChangeBroadcaster) EXPECT() *MockChangeBroadcasterMockRecorder {
}
// BroadcastJobUpdate mocks base method.
func (m *MockChangeBroadcaster) BroadcastJobUpdate(arg0 api.SocketIOJobUpdate) {
func (m *MockChangeBroadcaster) BroadcastJobUpdate(arg0 api.EventJobUpdate) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "BroadcastJobUpdate", arg0)
}
@ -243,7 +243,7 @@ func (mr *MockChangeBroadcasterMockRecorder) BroadcastJobUpdate(arg0 interface{}
}
// BroadcastTaskUpdate mocks base method.
func (m *MockChangeBroadcaster) BroadcastTaskUpdate(arg0 api.SocketIOTaskUpdate) {
func (m *MockChangeBroadcaster) BroadcastTaskUpdate(arg0 api.EventTaskUpdate) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "BroadcastTaskUpdate", arg0)
}

6
internal/manager/task_state_machine/task_state_machine_test.go
View File

@ -419,7 +419,7 @@ func (m *StateMachineMocks) expectBroadcastJobChange(
job *persistence.Job,
fromStatus, toStatus api.JobStatus,
) *gomock.Call {
expectUpdate := api.SocketIOJobUpdate{
expectUpdate := api.EventJobUpdate{
Id: job.UUID,
Name: &job.Name,
PreviousStatus: &fromStatus,
@ -434,7 +434,7 @@ func (m *StateMachineMocks) expectBroadcastJobChangeWithTaskRefresh(
job *persistence.Job,
fromStatus, toStatus api.JobStatus,
) *gomock.Call {
expectUpdate := api.SocketIOJobUpdate{
expectUpdate := api.EventJobUpdate{
Id: job.UUID,
Name: &job.Name,
PreviousStatus: &fromStatus,
@ -449,7 +449,7 @@ func (m *StateMachineMocks) expectBroadcastTaskChange(
task *persistence.Task,
fromStatus, toStatus api.TaskStatus,
) *gomock.Call {
expectUpdate := api.SocketIOTaskUpdate{
expectUpdate := api.EventTaskUpdate{
Id: task.UUID,
JobId: task.Job.UUID,
Name: task.Name,

4
internal/manager/task_state_machine/worker_requeue_test.go
View File

@ -45,7 +45,7 @@ func TestRequeueActiveTasksOfWorker(t *testing.T) {
mocks.logStorage.EXPECT().WriteTimestamped(gomock.Any(), task1.Job.UUID, task1.UUID, logMsg2)
mocks.logStorage.EXPECT().WriteTimestamped(gomock.Any(), task2.Job.UUID, task2.UUID, logMsg2)
mocks.broadcaster.EXPECT().BroadcastTaskUpdate(api.SocketIOTaskUpdate{
mocks.broadcaster.EXPECT().BroadcastTaskUpdate(api.EventTaskUpdate{
Activity: logMsg2,
Id: task1.UUID,
JobId: task1.Job.UUID,
@ -55,7 +55,7 @@ func TestRequeueActiveTasksOfWorker(t *testing.T) {
Updated: task1.UpdatedAt,
})
mocks.broadcaster.EXPECT().BroadcastTaskUpdate(api.SocketIOTaskUpdate{
mocks.broadcaster.EXPECT().BroadcastTaskUpdate(api.EventTaskUpdate{
Activity: logMsg2,
Id: task2.UUID,
JobId: task2.Job.UUID,

2
internal/manager/timeout_checker/interfaces.go
View File

@ -40,7 +40,7 @@ type LogStorage interface {
// TODO: Refactor the way worker status changes are handled, so that this
// service doens't need to broadcast its own worker updates.
type ChangeBroadcaster interface {
BroadcastWorkerUpdate(workerUpdate api.SocketIOWorkerUpdate)
BroadcastWorkerUpdate(workerUpdate api.EventWorkerUpdate)
}
// ChangeBroadcaster should be a subset of eventbus.Broker.

2
internal/manager/timeout_checker/mocks/interfaces_mock.gen.go generated
View File

@ -194,7 +194,7 @@ func (m *MockChangeBroadcaster) EXPECT() *MockChangeBroadcasterMockRecorder {
}
// BroadcastWorkerUpdate mocks base method.
func (m *MockChangeBroadcaster) BroadcastWorkerUpdate(arg0 api.SocketIOWorkerUpdate) {
func (m *MockChangeBroadcaster) BroadcastWorkerUpdate(arg0 api.EventWorkerUpdate) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "BroadcastWorkerUpdate", arg0)
}

2
internal/manager/timeout_checker/workers.go
View File

@ -60,7 +60,7 @@ func (ttc *TimeoutChecker) timeoutWorker(ctx context.Context, worker *persistenc
}
// Broadcast worker change via SocketIO
ttc.broadcaster.BroadcastWorkerUpdate(api.SocketIOWorkerUpdate{
ttc.broadcaster.BroadcastWorkerUpdate(api.EventWorkerUpdate{
Id: worker.UUID,
Name: worker.Name,
PreviousStatus: &prevStatus,

2
internal/manager/timeout_checker/workers_test.go
View File

@ -53,7 +53,7 @@ func TestWorkerTimeout(t *testing.T) {
mocks.persist.EXPECT().SaveWorker(mocks.ctx, &persistedWorker).Return(nil)
prevStatus := worker.Status
mocks.broadcaster.EXPECT().BroadcastWorkerUpdate(api.SocketIOWorkerUpdate{
mocks.broadcaster.EXPECT().BroadcastWorkerUpdate(api.EventWorkerUpdate{
Id: worker.UUID,
Name: worker.Name,
PreviousStatus: &prevStatus,

12
web/app/src/components/UpdateListener.vue
View File

@ -143,42 +143,42 @@ export default {
this.socket.on('/jobs', (jobUpdate) => {
// Convert to API object, in order to have the same parsing of data as
// when we'd do an API call.
const apiJobUpdate = API.SocketIOJobUpdate.constructFromObject(jobUpdate);
const apiJobUpdate = API.EventJobUpdate.constructFromObject(jobUpdate);
this.$emit('jobUpdate', apiJobUpdate);
});
this.socket.on('/last-rendered', (update) => {
// Convert to API object, in order to have the same parsing of data as
// when we'd do an API call.
const apiUpdate = API.SocketIOLastRenderedUpdate.constructFromObject(update);
const apiUpdate = API.EventLastRenderedUpdate.constructFromObject(update);
this.$emit('lastRenderedUpdate', apiUpdate);
});
this.socket.on('/task', (taskUpdate) => {
// Convert to API object, in order to have the same parsing of data as
// when we'd do an API call.
const apiTaskUpdate = API.SocketIOTaskUpdate.constructFromObject(taskUpdate);
const apiTaskUpdate = API.EventTaskUpdate.constructFromObject(taskUpdate);
this.$emit('taskUpdate', apiTaskUpdate);
});
this.socket.on('/tasklog', (taskLogUpdate) => {
// Convert to API object, in order to have the same parsing of data as
// when we'd do an API call.
const apiTaskLogUpdate = API.SocketIOTaskLogUpdate.constructFromObject(taskLogUpdate);
const apiTaskLogUpdate = API.EventTaskLogUpdate.constructFromObject(taskLogUpdate);
this.$emit('taskLogUpdate', apiTaskLogUpdate);
});
this.socket.on('/workers', (workerUpdate) => {
// Convert to API object, in order to have the same parsing of data as
// when we'd do an API call.
const apiWorkerUpdate = API.SocketIOWorkerUpdate.constructFromObject(workerUpdate);
const apiWorkerUpdate = API.EventWorkerUpdate.constructFromObject(workerUpdate);
this.$emit('workerUpdate', apiWorkerUpdate);
});
this.socket.on('/workertags', (workerTagUpdate) => {
// Convert to API object, in order to have the same parsing of data as
// when we'd do an API call.
const apiWorkerTagUpdate = API.SocketIOWorkerTagUpdate.constructFromObject(workerTagUpdate);
const apiWorkerTagUpdate = API.EventWorkerTagUpdate.constructFromObject(workerTagUpdate);
this.$emit('workerTagUpdate', apiWorkerTagUpdate);
});

2
web/app/src/components/jobs/JobDetails.vue
View File

@ -182,7 +182,7 @@ export default {
},
methods: {
/**
* @param {API.SocketIOLastRenderedUpdate} lastRenderedUpdate
* @param {API.EventLastRenderedUpdate} lastRenderedUpdate
*/
refreshLastRenderedImage(lastRenderedUpdate) {
this.$refs.lastRenderedImage.refreshLastRenderedImage(lastRenderedUpdate);

2
web/app/src/components/jobs/JobsTable.vue
View File

@ -49,7 +49,7 @@ export default {
const vueComponent = this;
const options = {
// See pkg/api/flamenco-openapi.yaml, schemas Job and SocketIOJobUpdate.
// See pkg/api/flamenco-openapi.yaml, schemas Job and EventJobUpdate.
columns: [
// Useful for debugging when there are many similar jobs:
// { title: "ID", field: "id", headerSort: false, formatter: (cell) => cell.getData().id.substr(0, 8), },

4
web/app/src/components/jobs/LastRenderedImage.vue
View File

@ -7,7 +7,7 @@
<script setup>
import { reactive, ref, watch } from 'vue';
import { api } from '@/urls';
import { JobsApi, JobLastRenderedImageInfo, SocketIOLastRenderedUpdate } from '@/manager-api';
import { JobsApi, JobLastRenderedImageInfo, EventLastRenderedUpdate } from '@/manager-api';
import { getAPIClient } from '@/api-client';
const props = defineProps([
@ -79,7 +79,7 @@ function setImageURL(thumbnailInfo) {
}
/**
* @param {SocketIOLastRenderedUpdate} lastRenderedUpdate
* @param {EventLastRenderedUpdate} lastRenderedUpdate
*/
function refreshLastRenderedImage(lastRenderedUpdate) {
// Only filter out other job IDs if this component has actually a non-empty job ID.

2
web/app/src/components/workers/WorkersTable.vue
View File

@ -49,7 +49,7 @@ export default {
const vueComponent = this;
const options = {
// See pkg/api/flamenco-openapi.yaml, schemas WorkerSummary and SocketIOWorkerUpdate.
// See pkg/api/flamenco-openapi.yaml, schemas WorkerSummary and EventWorkerUpdate.
columns: [
// Useful for debugging when there are many similar workers:
// { title: "ID", field: "id", headerSort: false, formatter: (cell) => cell.getData().id.substr(0, 8), },

6
web/app/src/stores/notifications.js
View File

@ -39,7 +39,7 @@ export const useNotifs = defineStore('notifications', {
},
/**
* @param {API.SocketIOJobUpdate} jobUpdate Job update received via SocketIO.
* @param {API.EventJobUpdate} jobUpdate Job update received via SocketIO.
*/
addJobUpdate(jobUpdate) {
let msg = 'Job';
@ -56,7 +56,7 @@ export const useNotifs = defineStore('notifications', {
},
/**
* @param {API.SocketIOTaskUpdate} taskUpdate Task update received via SocketIO.
* @param {API.EventTaskUpdate} taskUpdate Task update received via SocketIO.
*/
addTaskUpdate(taskUpdate) {
let msg = `Task ${taskUpdate.name}`;
@ -70,7 +70,7 @@ export const useNotifs = defineStore('notifications', {
},
/**
* @param {API.SocketIOWorkerUpdate} workerUpdate Worker update received via SocketIO.
* @param {API.EventWorkerUpdate} workerUpdate Worker update received via SocketIO.
*/
addWorkerUpdate(workerUpdate) {
let msg = `Worker ${workerUpdate.name}`;

2
web/app/src/stores/tasklog.js
View File

@ -26,7 +26,7 @@ export const useTaskLog = defineStore('taskLog', {
},
actions: {
/**
* @param {API.SocketIOTaskLogUpdate} taskLogUpdate
* @param {API.EventTaskLogUpdate} taskLogUpdate
*/
addTaskLogUpdate(taskLogUpdate) {
this.addChunk(taskLogUpdate.log);

6
web/app/src/views/JobsView.vue
View File

@ -192,7 +192,7 @@ export default {
/**
* Event handler for SocketIO task updates.
* @param {API.SocketIOTaskUpdate} taskUpdate
* @param {API.EventTaskUpdate} taskUpdate
*/
onSioTaskUpdate(taskUpdate) {
if (this.$refs.tasksTable) this.$refs.tasksTable.processTaskUpdate(taskUpdate);
@ -202,7 +202,7 @@ export default {
/**
* Event handler for SocketIO task log updates.
* @param {API.SocketIOTaskLogUpdate} taskLogUpdate
* @param {API.EventTaskLogUpdate} taskLogUpdate
*/
onSioTaskLogUpdate(taskLogUpdate) {
this.taskLog.addTaskLogUpdate(taskLogUpdate);
@ -210,7 +210,7 @@ export default {
/**
* Event handler for SocketIO "last-rendered" updates.
* @param {API.SocketIOLastRenderedUpdate} lastRenderedUpdate
* @param {API.EventLastRenderedUpdate} lastRenderedUpdate
*/
onSioLastRenderedUpdate(lastRenderedUpdate) {
this.$refs.jobDetails.refreshLastRenderedImage(lastRenderedUpdate);

2
web/app/src/views/LastRenderedView.vue
View File

@ -32,7 +32,7 @@ export default {
methods: {
/**
* Event handler for SocketIO "last-rendered" updates.
* @param {API.SocketIOLastRenderedUpdate} lastRenderedUpdate
* @param {API.EventLastRenderedUpdate} lastRenderedUpdate
*/
onSioLastRenderedUpdate(lastRenderedUpdate) {
this.$refs.lastRenderedImage.refreshLastRenderedImage(lastRenderedUpdate);