120 lines
3.4 KiB
Go
120 lines
3.4 KiB
Go
package service
|
|
|
|
import (
|
|
"strings"
|
|
|
|
"browser.local/platform/domain"
|
|
)
|
|
|
|
const logEventSubscriberBuffer = 512
|
|
|
|
type LogEventSubscriptionEventKind string
|
|
|
|
const (
|
|
LogEventSubscriptionEventLog LogEventSubscriptionEventKind = "log"
|
|
LogEventSubscriptionEventProcessState LogEventSubscriptionEventKind = "process-state"
|
|
)
|
|
|
|
type LogEventSubscriptionEvent struct {
|
|
Kind LogEventSubscriptionEventKind
|
|
LogEvent domain.LogStreamEvent
|
|
ServerInstanceID string
|
|
ProcessState domain.ServerInstanceState
|
|
}
|
|
|
|
type LogEventSubscription struct {
|
|
Events <-chan LogEventSubscriptionEvent
|
|
Close func()
|
|
}
|
|
|
|
type logEventSubscriber struct {
|
|
serverInstanceID string
|
|
events chan LogEventSubscriptionEvent
|
|
}
|
|
|
|
func (svc *CoreService) SubscribeLogEvents(serverInstanceID string) (LogEventSubscription, error) {
|
|
serverInstanceID = strings.TrimSpace(serverInstanceID)
|
|
if serverInstanceID == "" {
|
|
return LogEventSubscription{}, validationError("serverInstanceId is required")
|
|
}
|
|
if _, err := svc.store.ServerInstances().Get(serverInstanceID); err != nil {
|
|
return LogEventSubscription{}, err
|
|
}
|
|
events := make(chan LogEventSubscriptionEvent, logEventSubscriberBuffer)
|
|
svc.logEventMu.Lock()
|
|
svc.logEventSubscriberSeq++
|
|
id := svc.logEventSubscriberSeq
|
|
svc.logEventSubscribers[id] = logEventSubscriber{serverInstanceID: serverInstanceID, events: events}
|
|
svc.logEventMu.Unlock()
|
|
closeOnce := func() {
|
|
svc.logEventMu.Lock()
|
|
if subscriber, ok := svc.logEventSubscribers[id]; ok {
|
|
delete(svc.logEventSubscribers, id)
|
|
close(subscriber.events)
|
|
}
|
|
svc.logEventMu.Unlock()
|
|
}
|
|
return LogEventSubscription{Events: events, Close: closeOnce}, nil
|
|
}
|
|
|
|
func (svc *CoreService) SubscribeLogEventsForSession(sessionID string, serverInstanceID string) (LogEventSubscription, error) {
|
|
instance, err := svc.GetServerInstanceForSession(sessionID, serverInstanceID)
|
|
if err != nil {
|
|
return LogEventSubscription{}, err
|
|
}
|
|
return svc.SubscribeLogEvents(instance.ID)
|
|
}
|
|
|
|
func (svc *CoreService) publishLogEvents(stream domain.LogStream, entries []domain.LogEntry) {
|
|
if len(entries) == 0 {
|
|
return
|
|
}
|
|
events := make([]LogEventSubscriptionEvent, len(entries))
|
|
for index, entry := range entries {
|
|
events[index] = LogEventSubscriptionEvent{
|
|
Kind: LogEventSubscriptionEventLog,
|
|
LogEvent: domain.CopyLogStreamEvent(domain.LogStreamEvent{
|
|
ServerInstanceID: stream.ServerInstanceID,
|
|
Stream: stream,
|
|
Entry: entry,
|
|
LatestSeq: stream.LatestSeq,
|
|
}),
|
|
}
|
|
}
|
|
svc.publishLogSubscriptionEvents(stream.ServerInstanceID, events)
|
|
}
|
|
|
|
func (svc *CoreService) publishLogProcessState(instance domain.ServerInstance) {
|
|
svc.publishLogSubscriptionEvents(instance.ID, []LogEventSubscriptionEvent{{
|
|
Kind: LogEventSubscriptionEventProcessState,
|
|
ServerInstanceID: instance.ID,
|
|
ProcessState: instance.State,
|
|
}})
|
|
}
|
|
|
|
func (svc *CoreService) publishLogSubscriptionEvents(serverInstanceID string, events []LogEventSubscriptionEvent) {
|
|
if len(events) == 0 {
|
|
return
|
|
}
|
|
svc.logEventMu.Lock()
|
|
for id, subscriber := range svc.logEventSubscribers {
|
|
if subscriber.serverInstanceID != serverInstanceID {
|
|
continue
|
|
}
|
|
dropped := false
|
|
for _, event := range events {
|
|
select {
|
|
case subscriber.events <- event:
|
|
default:
|
|
delete(svc.logEventSubscribers, id)
|
|
close(subscriber.events)
|
|
dropped = true
|
|
}
|
|
if dropped {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
svc.logEventMu.Unlock()
|
|
}
|