mirror of
https://github.com/moby/moby.git
synced 2022-11-09 12:21:53 -05:00
Remove publisher if no one is listening
Signed-off-by: Michael Crosby <crosbymichael@gmail.com>
This commit is contained in:
parent
76141a0077
commit
217a2bd1b6
3 changed files with 13 additions and 2 deletions
|
@ -68,6 +68,9 @@ func (s *statsCollector) unsubscribe(c *Container, ch chan interface{}) {
|
||||||
publisher := s.publishers[c]
|
publisher := s.publishers[c]
|
||||||
if publisher != nil {
|
if publisher != nil {
|
||||||
publisher.Evict(ch)
|
publisher.Evict(ch)
|
||||||
|
if publisher.Len() == 0 {
|
||||||
|
delete(s.publishers, c)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
s.m.Unlock()
|
s.m.Unlock()
|
||||||
}
|
}
|
||||||
|
|
|
@ -274,10 +274,10 @@ func TestGetContainerStats(t *testing.T) {
|
||||||
t.Fatalf("GET containers/stats sockRequest failed: %v", err)
|
t.Fatalf("GET containers/stats sockRequest failed: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
dec := json.NewDecoder(bytes.NewBuffer(body))
|
||||||
var s *stats.Stats
|
var s *stats.Stats
|
||||||
if err := json.Unmarshal(body, &s); err != nil {
|
if err := dec.Decode(&s); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
logDone("container REST API - check GET containers/stats")
|
logDone("container REST API - check GET containers/stats")
|
||||||
}
|
}
|
||||||
|
|
|
@ -26,6 +26,14 @@ type Publisher struct {
|
||||||
subscribers map[subscriber]struct{}
|
subscribers map[subscriber]struct{}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Len returns the number of subscribers for the publisher
|
||||||
|
func (p *Publisher) Len() int {
|
||||||
|
p.m.RLock()
|
||||||
|
i := len(p.subscribers)
|
||||||
|
p.m.RUnlock()
|
||||||
|
return i
|
||||||
|
}
|
||||||
|
|
||||||
// Subscribe adds a new subscriber to the publisher returning the channel.
|
// Subscribe adds a new subscriber to the publisher returning the channel.
|
||||||
func (p *Publisher) Subscribe() chan interface{} {
|
func (p *Publisher) Subscribe() chan interface{} {
|
||||||
ch := make(chan interface{}, p.buffer)
|
ch := make(chan interface{}, p.buffer)
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue