78 lines
2.5 KiB
Go
78 lines
2.5 KiB
Go
package service
|
|
|
|
import (
|
|
"sync"
|
|
|
|
"browser.local/platform/domain"
|
|
"browser.local/platform/validator"
|
|
)
|
|
|
|
const defaultRunControlEventRetrySeconds = 5
|
|
|
|
type RunControlEventSubscription struct {
|
|
Initial *domain.RunControlEvent
|
|
Events <-chan domain.RunControlEvent
|
|
Cancel func()
|
|
}
|
|
|
|
func (svc *CoreService) SubscribeRunControlEvents(request domain.RunControlStreamRequest) (RunControlEventSubscription, error) {
|
|
request = domain.CopyRunControlStreamRequest(request)
|
|
if err := validator.ValidateRunControlStreamRequest(request); err != nil {
|
|
return RunControlEventSubscription{}, err
|
|
}
|
|
if err := svc.validateRunSession(request.RunEndpointID, request.SessionToken); err != nil {
|
|
return RunControlEventSubscription{}, err
|
|
}
|
|
|
|
updates := make(chan domain.RunControlEvent, 8)
|
|
var initial *domain.RunControlEvent
|
|
svc.controlStreamMu.Lock()
|
|
if latest, exists := svc.controlStreamEvents[request.RunEndpointID]; exists && latest.Sequence > request.LastEventSeq {
|
|
copy := domain.CopyRunControlEvent(latest)
|
|
initial = ©
|
|
}
|
|
svc.controlStreamWaiters[request.RunEndpointID] = append(svc.controlStreamWaiters[request.RunEndpointID], updates)
|
|
svc.controlStreamMu.Unlock()
|
|
|
|
var once sync.Once
|
|
cancel := func() {
|
|
once.Do(func() {
|
|
svc.controlStreamMu.Lock()
|
|
waiters := svc.controlStreamWaiters[request.RunEndpointID]
|
|
for index, candidate := range waiters {
|
|
if candidate == updates {
|
|
waiters = append(waiters[:index], waiters[index+1:]...)
|
|
break
|
|
}
|
|
}
|
|
if len(waiters) == 0 {
|
|
delete(svc.controlStreamWaiters, request.RunEndpointID)
|
|
} else {
|
|
svc.controlStreamWaiters[request.RunEndpointID] = waiters
|
|
}
|
|
svc.controlStreamMu.Unlock()
|
|
close(updates)
|
|
})
|
|
}
|
|
return RunControlEventSubscription{Initial: initial, Events: updates, Cancel: cancel}, nil
|
|
}
|
|
|
|
func (svc *CoreService) publishRunControlEvent(runEndpointID string, eventType string) {
|
|
if runEndpointID == "" || eventType == "" {
|
|
return
|
|
}
|
|
svc.controlStreamMu.Lock()
|
|
sequence := svc.controlStreamSeq[runEndpointID] + 1
|
|
svc.controlStreamSeq[runEndpointID] = sequence
|
|
event := domain.RunControlEvent{RunEndpointID: runEndpointID, Sequence: sequence, Type: eventType, ServerTime: svc.now(), RetrySeconds: defaultRunControlEventRetrySeconds}
|
|
svc.controlStreamEvents[runEndpointID] = event
|
|
waiters := append([]chan domain.RunControlEvent(nil), svc.controlStreamWaiters[runEndpointID]...)
|
|
svc.controlStreamMu.Unlock()
|
|
for _, waiter := range waiters {
|
|
select {
|
|
case waiter <- event:
|
|
default:
|
|
}
|
|
}
|
|
}
|