// Copyright 2019 Drone IO, Inc. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package livelog import ( "context" "sync" "github.com/drone/drone/core" ) // this is the amount of items that are stored in memory // in the buffer. This should result in approximately 10kb // of memory allocated per-stream and per-subscriber, not // including any logdata stored in these structures. const bufferSize = 5000 type stream struct { sync.Mutex hist []*core.Line list map[*subscriber]struct{} } func newStream() *stream { return &stream{ list: map[*subscriber]struct{}{}, } } func (s *stream) write(line *core.Line) error { s.Lock() s.hist = append(s.hist, line) for l := range s.list { l.publish(line) } // the history should not be unbounded. The history // slice is capped and items are removed in a FIFO // ordering when capacity is reached. if size := len(s.hist); size >= bufferSize { s.hist = s.hist[size-bufferSize:] } s.Unlock() return nil } func (s *stream) subscribe(ctx context.Context) (<-chan *core.Line, <-chan error) { sub := &subscriber{ handler: make(chan *core.Line, bufferSize), closec: make(chan struct{}), } err := make(chan error) s.Lock() for _, line := range s.hist { sub.publish(line) } s.list[sub] = struct{}{} s.Unlock() go func() { defer close(err) select { case <-sub.closec: case <-ctx.Done(): sub.close() } }() return sub.handler, err } func (s *stream) close() error { s.Lock() defer s.Unlock() for sub := range s.list { delete(s.list, sub) sub.close() } return nil }