1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
// Copyright (C) 2023 The Syncthing Authors.
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this file,
// You can obtain one at https://mozilla.org/MPL/2.0/.
package model
import (
"context"
"fmt"
"time"
"github.com/syncthing/syncthing/lib/events"
"github.com/syncthing/syncthing/lib/svcutil"
"github.com/thejerf/suture/v4"
)
var errSvcNotFound = fmt.Errorf("service not found")
// A serviceMap is a utility map of arbitrary keys to a suture.Service of
// some kind, where adding and removing services ensures they are properly
// started and stopped on the given Supervisor. The serviceMap is itself a
// suture.Service and should be added to a Supervisor.
// Not safe for concurrent use.
type serviceMap[K comparable, S suture.Service] struct {
services map[K]S
tokens map[K]suture.ServiceToken
supervisor *suture.Supervisor
eventLogger events.Logger
}
func newServiceMap[K comparable, S suture.Service](eventLogger events.Logger) *serviceMap[K, S] {
m := &serviceMap[K, S]{
services: make(map[K]S),
tokens: make(map[K]suture.ServiceToken),
eventLogger: eventLogger,
}
m.supervisor = suture.New(m.String(), svcutil.SpecWithDebugLogger(l))
return m
}
// Add adds a service to the map, starting it on the supervisor. If there is
// already a service at the given key, it is removed first.
func (s *serviceMap[K, S]) Add(k K, v S) {
if tok, ok := s.tokens[k]; ok {
// There is already a service at this key, remove it first.
s.supervisor.Remove(tok)
s.eventLogger.Log(events.Failure, fmt.Sprintf("%s replaced service at key %v", s, k))
}
s.services[k] = v
s.tokens[k] = s.supervisor.Add(v)
}
// Get returns the service at the given key, or the empty value and false if
// there is no service at that key.
func (s *serviceMap[K, S]) Get(k K) (v S, ok bool) {
v, ok = s.services[k]
return
}
// Stop removes the service at the given key from the supervisor, stopping it.
// The service itself is still retained, i.e. a call to Get with the same key
// will still return a result.
func (s *serviceMap[K, S]) Stop(k K) {
if tok, ok := s.tokens[k]; ok {
s.supervisor.Remove(tok)
}
return
}
// StopAndWaitChan removes the service at the given key from the supervisor,
// stopping it. The service itself is still retained, i.e. a call to Get with
// the same key will still return a result.
// The returned channel will produce precisely one error value: either the
// return value from RemoveAndWait (possibly nil), or errSvcNotFound if the
// service was not found.
func (s *serviceMap[K, S]) StopAndWaitChan(k K, timeout time.Duration) <-chan error {
ret := make(chan error, 1)
if tok, ok := s.tokens[k]; ok {
go func() {
ret <- s.supervisor.RemoveAndWait(tok, timeout)
}()
} else {
ret <- errSvcNotFound
}
return ret
}
// Remove removes the service at the given key, stopping it on the supervisor.
// If there is no service at the given key, nothing happens. The return value
// indicates whether a service was removed.
func (s *serviceMap[K, S]) Remove(k K) (found bool) {
if tok, ok := s.tokens[k]; ok {
found = true
s.supervisor.Remove(tok)
} else {
_, found = s.services[k]
}
delete(s.services, k)
delete(s.tokens, k)
return
}
// RemoveAndWait removes the service at the given key, stopping it on the
// supervisor. Returns errSvcNotFound if there is no service at the given
// key, otherwise the return value from the supervisor's RemoveAndWait.
func (s *serviceMap[K, S]) RemoveAndWait(k K, timeout time.Duration) error {
return <-s.RemoveAndWaitChan(k, timeout)
}
// RemoveAndWaitChan removes the service at the given key, stopping it on
// the supervisor. The returned channel will produce precisely one error
// value: either the return value from RemoveAndWait (possibly nil), or
// errSvcNotFound if the service was not found.
func (s *serviceMap[K, S]) RemoveAndWaitChan(k K, timeout time.Duration) <-chan error {
ret := s.StopAndWaitChan(k, timeout)
delete(s.services, k)
return ret
}
// Each calls the given function for each service in the map. An error from
// fn will stop the iteration and be returned as-is.
func (s *serviceMap[K, S]) Each(fn func(K, S) error) error {
for key, svc := range s.services {
if err := fn(key, svc); err != nil {
return err
}
}
return nil
}
// Suture implementation
func (s *serviceMap[K, S]) Serve(ctx context.Context) error {
return s.supervisor.Serve(ctx)
}
func (s *serviceMap[K, S]) String() string {
var kv K
var sv S
return fmt.Sprintf("serviceMap[%T, %T]@%p", kv, sv, s)
}
|
