asset_service.go

v0.5.0

package steward

import (
	"context"
	"fmt"
	"reflect"
	"strconv"
	"sync"
	"time"

	"github.com/soverenio/vanilla/throw"

	"go.bigb.es/auxilia/async"
	"go.bigb.es/auxilia/fn"
	"go.bigb.es/auxilia/internal/logutil"
)

type AssetStatus int

const (
	AssetStatusUnknown AssetStatus = iota
	AssetStatusInjected
	AssetStatusInitialized
	AssetStatusStarted
	AssetStatusStopping
	AssetStatusGracefulStopped
	AssetStatusStopped
	AssetStatusDestroyed
)

const callStopWarningDuration = 100 * time.Millisecond

var _ Asset = &serviceAsset{}

type serviceAsset struct {
	mu     sync.Mutex
	status AssetStatus

	dependencies   []*serviceAsset
	dependents     []*serviceAsset
	value          any
	reflectedValue reflect.Value
	roots          []assetTreeNode

	// properties
	root         bool // must be started/stopped even if it is not a dependency
	ignoreUnused bool // ignore unused assets
	singleton    any  // singleton assets are always root assets
}

func ServiceAsset(value any, opts ...ComponentOption) (Asset, error) {
	asset := &serviceAsset{
		value:          value,
		reflectedValue: reflect.ValueOf(value),
	}
	for _, opt := range opts {
		opt(asset)
	}
	return asset, nil
}

func NewServiceAsset(value any, opts ...ComponentOption) (Asset, error) {
	return ServiceAsset(value, opts...)
}

func MustServiceAsset(value any, opts ...ComponentOption) Asset {
	return fn.Must1(ServiceAsset(value, opts...))
}

func MustNewServiceAsset(value any, opts ...ComponentOption) Asset {
	return MustServiceAsset(value, opts...)
}

func SingletonAsset(key any, value any, opts ...ComponentOption) (Asset, error) {
	asset, err := ServiceAsset(value, append(opts, Singleton(key))...)
	if err != nil {
		return nil, err
	}
	return fn.FirstOf2(RegisterSingleton(key, asset)), nil
}

func MustSingletonAsset(key any, value any, opts ...ComponentOption) Asset {
	return fn.Must1(SingletonAsset(key, value, opts...))
}

func (c *serviceAsset) asset() {}

func (c *serviceAsset) String() string {
	return c.Name()
}

// TryInjectTo tries to inject the asset into the component on the given field.
func (c *serviceAsset) TryInjectTo(component *serviceAsset, field string) (injected bool) {
	objectValue := component.reflectedValue.Elem().FieldByName(field)

	c.mu.Lock()
	defer c.mu.Unlock()

	// if objectValue is slice then check assignable to slice element type
	if objectValue.Kind() == reflect.Slice {
		if !c.TypeAssignableTo(objectValue.Type().Elem()) {
			return false
		}

		component.dependencies = append(component.dependencies, c)
		// TODO: may be we need a better way to handle this
		if c.status < AssetStatusStarted {
			c.dependents = append(c.dependents, component)
		}

		// append to slice
		objectValue.Set(reflect.Append(objectValue, c.reflectedValue))

		return true
	}

	if !c.TypeAssignableTo(objectValue.Type()) {
		return false
	}

	component.dependencies = append(component.dependencies, c)
	// TODO: may be we need a better way to handle this
	if c.status < AssetStatusStarted {
		c.dependents = append(c.dependents, component)
	}

	objectValue.Set(c.reflectedValue)

	return true
}

func (c *serviceAsset) Name() string {
	return reflect.TypeOf(c.value).Elem().String()
}

func (c *serviceAsset) populateConfiguration(ctx context.Context, configurations []*configurationAsset) {
	log := logutil.FromContext(ctx).With(
		"component", "manager",
		"InjectorType", "configuration",
	)

	structFieldIteratorFromType(c.reflectedValue).Filter(func(d structField) bool {
		switch {
		case !hasTag(d.First(), "config"):
			return false
		case !isPublic(d.First()):
			// better to catch that early on
			panic(fmt.Sprintf("we've got config tag on private field: target=%s targetField=%s",
				c.Name(), d.First().Name))
		case d.First().Type.Kind() != reflect.Struct:
			// better to catch that early on
			panic(fmt.Sprintf("we've got config tag on invalid type: target=%s targetField=%s expected=%s got=%s",
				c.Name(), d.First().Name, reflect.Struct.String(), d.First().Type.Kind().String()))
		default:
			return true
		}
	}).Apply(func(d structField) {
		for _, dep := range configurations {
			if !dep.TryInjectTo(c, d.First().Name) {
				continue
			}

			log.Debug("injecting configuration",
				"target", c.Name()+"."+d.First().Name,
				"dependency", d.First().Type.Name(),
			)

			return
		}

		panic(fmt.Sprintf("failed to find dependency: target=%s targetField=%s dependencyType=%s",
			c.Name(), d.First().Name, d.First().Type.String()))
	})
}

func (c *serviceAsset) populateDependencies(ctx context.Context, assets []*serviceAsset) {
	log := logutil.FromContext(ctx).With(
		"component", "manager",
		"InjectorType", "injections",
	)

	structFieldIteratorFromType(c.reflectedValue).Filter(func(d structField) bool {
		switch {
		case !hasTag(d.First(), "inject"):
			return false
		case !isPublic(d.First()):
			// better to catch that early on
			panic(fmt.Sprintf("we've got inject tag on private field: target=%s targetField=%s",
				c.Name(), d.First().Name))
		case !isAllowedInjectType(d.First().Type):
			// better to catch that early on
			panic(fmt.Sprintf("we've got inject tag on invalid type: target=%s targetField=%s expected=%s got=%s",
				c.Name(), d.First().Name, "interface/pointer to struct/list of interface or pointer to struct", d.First().Type.Kind().String()))
		default:
			return true
		}
	}).Apply(func(d structField) {
		var (
			found          bool
			stopAfterFirst bool

			fieldName = d.First().Name
			fieldType = d.First().Type
		)

		switch {
		case isSingleInjectType(fieldType):
			stopAfterFirst = true
		case isMultiInjectType(fieldType):
			stopAfterFirst = false
		default:
			panic(throw.IllegalState())
		}

		// fieldPath is used for logging and consists of component name and field name
		fieldPath := c.Name() + "." + fieldName

		for _, dep := range assets {
			// full fieldPathWithIdx looks like <component>.<field>[<idx>]
			fieldPathWithIdx := fieldPath
			if !stopAfterFirst { // if we're inserting into slice then we need to add index
				fieldPathWithIdx += "[" + strconv.Itoa(d.Second().Len()) + "]"
			}

			switch {
			case dep == c:
			case dep.TryInjectTo(c, fieldName):
				log.Debug("injecting component",
					"target", fieldPathWithIdx,
					"dependency", dep.reflectedValue.Type().String(),
				)
				found = true

				if stopAfterFirst {
					return
				}
			}
		}

		if !found && !hasTagBool(d.First(), "optional") {
			panic(fmt.Sprintf("failed to find dependency: target=%s targetField=%s dependencyType=%s",
				c.Name(), fieldName, fieldType.String()))
		}
	})

}

// TypeAssignableTo returns true if the asset type is assignable to the given type.
func (c *serviceAsset) TypeAssignableTo(r reflect.Type) bool {
	return c.reflectedValue.Type().AssignableTo(r)
}

type ComponentOption func(asset *serviceAsset)

// Root marks the asset as a root asset. Root assets are always started/stopped.
func Root() ComponentOption {
	return func(asset *serviceAsset) {
		asset.root = true
	}
}

func IgnoreUnused() ComponentOption {
	return func(asset *serviceAsset) {
		asset.ignoreUnused = true
	}
}

// Singleton marks the asset as a singleton asset.
// Singleton assets are always root assets and created only once.
// NB: Singleton assets cant depend on other assets, for now.
func Singleton(singletonKey any) ComponentOption {
	if singletonKey == nil || !isComparable(singletonKey) {
		panic(throw.IllegalValue())
	}

	return func(asset *serviceAsset) {
		asset.singleton = singletonKey
		asset.root = true
	}
}

func (c *serviceAsset) IsSingleton() bool {
	return c.singleton != nil
}

func (c *serviceAsset) NeedStart() bool {
	return c.root || c.dependents != nil
}

func (c *serviceAsset) CallInit(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	switch {
	case c.status == AssetStatusInjected:
		c.status = AssetStatusInitialized
	case c.singleton == nil:
		panic(throw.IllegalState()) // double call of init is not allowed on non-singleton assets
	default:
		return nil
	}

	_, log := logutil.WithField(ctx, "component", "manager")

	converted, ok := c.value.(Initer)
	if ok {
		log.Debug("initializing component", "component", c.Name())

		return converted.Init(ctx)
	}

	return nil
}

func (c *serviceAsset) CallStart(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	switch {
	case c.status == AssetStatusInitialized:
		c.status = AssetStatusStarted
	case c.singleton == nil:
		panic(throw.IllegalState()) // double call of start is not allowed on non-singleton assets
	default:
		return nil
	}

	if !c.NeedStart() {
		return nil
	}

	_, log := logutil.WithField(ctx, "component", "manager")

	converted, ok := c.value.(Starter)
	if ok {
		log.Debug("starting component", "component", c.Name())

		return converted.Start(ctx)
	}

	return nil
}

func (c *serviceAsset) CallStop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	startTime := time.Now()

	switch c.status {
	case AssetStatusStarted, AssetStatusStopping:
		c.status = AssetStatusStopped
	default:
		return nil
	}

	if !c.NeedStart() {
		return nil
	}

	_, log := logutil.WithField(ctx, "component", "manager")

	converted, ok := c.value.(Stopper)
	if !ok {
		return nil
	}

	log.Info("stopping component (CallStop)", "component", c.Name())

	err := converted.Stop(ctx)
	if err != nil {
		return err
	}

	log.Info("stopped component (CallStop)", "component", c.Name())

	dur := time.Since(startTime)
	if dur > callStopWarningDuration {
		logutil.FromContext(ctx).Warn("CallStop took too long",
			"component", c.Name(),
			"duration", dur,
		)
	}

	return nil
}

// CallGracefulStop calls GracefulStop on the asset if it implements GracefulStopper.
// It also honors the ctx timeout.
// If the asset does not implement GracefulStopper then it calls Stop on the asset if it implements Stopper.
func (c *serviceAsset) CallGracefulStop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	switch {
	case c.status == AssetStatusStarted:
		c.status = AssetStatusStopping
	case c.singleton == nil:
		panic(throw.IllegalState()) // double call of stop is not allowed on non-singleton assets
	default:
		return nil
	}

	if !c.NeedStart() {
		return nil
	}

	_, log := logutil.WithField(ctx, "component", "manager")

	converted, ok := c.value.(GracefulStopper)
	if ok {
		log.Info("gracefully stopping component", "component", c.Name())

		task := async.WrapErrFunc[struct{}](converted.GracefulStop)
		err := fn.ThirdOf3(async.Start(ctx, task).Wait(ctx))
		if err != nil {
			return err
		}

		c.status = AssetStatusGracefulStopped
		log.Info("gracefully stopped component", "component", c.Name())

		return nil
	}

	convertedStopper, ok := c.value.(Stopper)
	if ok {
		log.Info("stopping component (CallGracefulStop)", "component", c.Name())

		err := convertedStopper.Stop(ctx)
		if err != nil {
			return err
		}

		c.status = AssetStatusStopped
		log.Info("stopped component (CallGracefulStop)", "component", c.Name())

		return nil
	}

	return nil
}

func (c *serviceAsset) CallHealthCheck(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.status != AssetStatusStarted {
		return throw.New("asset is not running", struct{ Asset string }{Asset: c.Name()})
	}

	converted, ok := c.value.(HealthChecker)
	if !ok {
		return nil
	}
	return converted.HealthCheck(ctx)
}

func (c *serviceAsset) CallDestroy(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	switch c.status {
	case AssetStatusGracefulStopped, AssetStatusStopped:
		c.status = AssetStatusDestroyed
	default:
		// No exit here
	}

	if !c.NeedStart() {
		return nil
	}

	_, log := logutil.WithField(ctx, "component", "manager")

	converted, ok := c.value.(Destroyer)
	if !ok {
		c.status = AssetStatusDestroyed // For CheckStopped correctness

		return nil
	}

	log.Info("destroying component", "component", c.Name())

	err := converted.Destroy(ctx)
	if err != nil {
		log.Info("destroyed component", "component", c.Name())
	}

	return err
}


1	package steward
2
3	import (
4	"context"
5	"fmt"
6	"reflect"
7	"strconv"
8	"sync"
9	"time"
10
11	"github.com/soverenio/vanilla/throw"
12
13	"go.bigb.es/auxilia/async"
14	"go.bigb.es/auxilia/fn"
15	"go.bigb.es/auxilia/internal/logutil"
16	)
17
18	type AssetStatus int
19
20	const (
21	AssetStatusUnknown AssetStatus = iota
22	AssetStatusInjected
23	AssetStatusInitialized
24	AssetStatusStarted
25	AssetStatusStopping
26	AssetStatusGracefulStopped
27	AssetStatusStopped
28	AssetStatusDestroyed
29	)
30
31	const callStopWarningDuration = 100 * time.Millisecond
32
33	var _ Asset = &serviceAsset{}
34
35	type serviceAsset struct {
36	mu sync.Mutex
37	status AssetStatus
38
39	dependencies []*serviceAsset
40	dependents []*serviceAsset
41	value any
42	reflectedValue reflect.Value
43	roots []assetTreeNode
44
45	// properties
46	root bool // must be started/stopped even if it is not a dependency
47	ignoreUnused bool // ignore unused assets
48	singleton any // singleton assets are always root assets
49	}
50
51	func ServiceAsset(value any, opts ...ComponentOption) (Asset, error) {
52	asset := &serviceAsset{
53	value: value,
54	reflectedValue: reflect.ValueOf(value),
55	}
56	for _, opt := range opts {
57	opt(asset)
58	}
59	return asset, nil
60	}
61
62	func NewServiceAsset(value any, opts ...ComponentOption) (Asset, error) {
63	return ServiceAsset(value, opts...)
64	}
65
66	func MustServiceAsset(value any, opts ...ComponentOption) Asset {
67	return fn.Must1(ServiceAsset(value, opts...))
68	}
69
70	func MustNewServiceAsset(value any, opts ...ComponentOption) Asset {
71	return MustServiceAsset(value, opts...)
72	}
73
74	func SingletonAsset(key any, value any, opts ...ComponentOption) (Asset, error) {
75	asset, err := ServiceAsset(value, append(opts, Singleton(key))...)
76	if err != nil {
77	return nil, err
78	}
79	return fn.FirstOf2(RegisterSingleton(key, asset)), nil
80	}
81
82	func MustSingletonAsset(key any, value any, opts ...ComponentOption) Asset {
83	return fn.Must1(SingletonAsset(key, value, opts...))
84	}
85
86	func (c *serviceAsset) asset() {}
87
88	func (c *serviceAsset) String() string {
89	return c.Name()
90	}
91
92	// TryInjectTo tries to inject the asset into the component on the given field.
93	func (c serviceAsset) TryInjectTo(component serviceAsset, field string) (injected bool) {
94	objectValue := component.reflectedValue.Elem().FieldByName(field)
95
96	c.mu.Lock()
97	defer c.mu.Unlock()
98
99	// if objectValue is slice then check assignable to slice element type
100	if objectValue.Kind() == reflect.Slice {
101	if !c.TypeAssignableTo(objectValue.Type().Elem()) {
102	return false
103	}
104
105	component.dependencies = append(component.dependencies, c)
106	// TODO: may be we need a better way to handle this
107	if c.status < AssetStatusStarted {
108	c.dependents = append(c.dependents, component)
109	}
110
111	// append to slice
112	objectValue.Set(reflect.Append(objectValue, c.reflectedValue))
113
114	return true
115	}
116
117	if !c.TypeAssignableTo(objectValue.Type()) {
118	return false
119	}
120
121	component.dependencies = append(component.dependencies, c)
122	// TODO: may be we need a better way to handle this
123	if c.status < AssetStatusStarted {
124	c.dependents = append(c.dependents, component)
125	}
126
127	objectValue.Set(c.reflectedValue)
128
129	return true
130	}
131
132	func (c *serviceAsset) Name() string {
133	return reflect.TypeOf(c.value).Elem().String()
134	}
135
136	func (c serviceAsset) populateConfiguration(ctx context.Context, configurations []configurationAsset) {
137	log := logutil.FromContext(ctx).With(
138	"component", "manager",
139	"InjectorType", "configuration",
140	)
141
142	structFieldIteratorFromType(c.reflectedValue).Filter(func(d structField) bool {
143	switch {
144	case !hasTag(d.First(), "config"):
145	return false
146	case !isPublic(d.First()):
147	// better to catch that early on
148	panic(fmt.Sprintf("we've got config tag on private field: target=%s targetField=%s",
149	c.Name(), d.First().Name))
150	case d.First().Type.Kind() != reflect.Struct:
151	// better to catch that early on
152	panic(fmt.Sprintf("we've got config tag on invalid type: target=%s targetField=%s expected=%s got=%s",
153	c.Name(), d.First().Name, reflect.Struct.String(), d.First().Type.Kind().String()))
154	default:
155	return true
156	}
157	}).Apply(func(d structField) {
158	for _, dep := range configurations {
159	if !dep.TryInjectTo(c, d.First().Name) {
160	continue
161	}
162
163	log.Debug("injecting configuration",
164	"target", c.Name()+"."+d.First().Name,
165	"dependency", d.First().Type.Name(),
166	)
167
168	return
169	}
170
171	panic(fmt.Sprintf("failed to find dependency: target=%s targetField=%s dependencyType=%s",
172	c.Name(), d.First().Name, d.First().Type.String()))
173	})
174	}
175
176	func (c serviceAsset) populateDependencies(ctx context.Context, assets []serviceAsset) {
177	log := logutil.FromContext(ctx).With(
178	"component", "manager",
179	"InjectorType", "injections",
180	)
181
182	structFieldIteratorFromType(c.reflectedValue).Filter(func(d structField) bool {
183	switch {
184	case !hasTag(d.First(), "inject"):
185	return false
186	case !isPublic(d.First()):
187	// better to catch that early on
188	panic(fmt.Sprintf("we've got inject tag on private field: target=%s targetField=%s",
189	c.Name(), d.First().Name))
190	case !isAllowedInjectType(d.First().Type):
191	// better to catch that early on
192	panic(fmt.Sprintf("we've got inject tag on invalid type: target=%s targetField=%s expected=%s got=%s",
193	c.Name(), d.First().Name, "interface/pointer to struct/list of interface or pointer to struct", d.First().Type.Kind().String()))
194	default:
195	return true
196	}
197	}).Apply(func(d structField) {
198	var (
199	found bool
200	stopAfterFirst bool
201
202	fieldName = d.First().Name
203	fieldType = d.First().Type
204	)
205
206	switch {
207	case isSingleInjectType(fieldType):
208	stopAfterFirst = true
209	case isMultiInjectType(fieldType):
210	stopAfterFirst = false
211	default:
212	panic(throw.IllegalState())
213	}
214
215	// fieldPath is used for logging and consists of component name and field name
216	fieldPath := c.Name() + "." + fieldName
217
218	for _, dep := range assets {
219	// full fieldPathWithIdx looks like <component>.<field>[<idx>]
220	fieldPathWithIdx := fieldPath
221	if !stopAfterFirst { // if we're inserting into slice then we need to add index
222	fieldPathWithIdx += "[" + strconv.Itoa(d.Second().Len()) + "]"
223	}
224
225	switch {
226	case dep == c:
227	case dep.TryInjectTo(c, fieldName):
228	log.Debug("injecting component",
229	"target", fieldPathWithIdx,
230	"dependency", dep.reflectedValue.Type().String(),
231	)
232	found = true
233
234	if stopAfterFirst {
235	return
236	}
237	}
238	}
239
240	if !found && !hasTagBool(d.First(), "optional") {
241	panic(fmt.Sprintf("failed to find dependency: target=%s targetField=%s dependencyType=%s",
242	c.Name(), fieldName, fieldType.String()))
243	}
244	})
245
246	}
247
248	// TypeAssignableTo returns true if the asset type is assignable to the given type.
249	func (c *serviceAsset) TypeAssignableTo(r reflect.Type) bool {
250	return c.reflectedValue.Type().AssignableTo(r)
251	}
252
253	type ComponentOption func(asset *serviceAsset)
254
255	// Root marks the asset as a root asset. Root assets are always started/stopped.
256	func Root() ComponentOption {
257	return func(asset *serviceAsset) {
258	asset.root = true
259	}
260	}
261
262	func IgnoreUnused() ComponentOption {
263	return func(asset *serviceAsset) {
264	asset.ignoreUnused = true
265	}
266	}
267
268	// Singleton marks the asset as a singleton asset.
269	// Singleton assets are always root assets and created only once.
270	// NB: Singleton assets cant depend on other assets, for now.
271	func Singleton(singletonKey any) ComponentOption {
272	if singletonKey == nil \|\| !isComparable(singletonKey) {
273	panic(throw.IllegalValue())
274	}
275
276	return func(asset *serviceAsset) {
277	asset.singleton = singletonKey
278	asset.root = true
279	}
280	}
281
282	func (c *serviceAsset) IsSingleton() bool {
283	return c.singleton != nil
284	}
285
286	func (c *serviceAsset) NeedStart() bool {
287	return c.root \|\| c.dependents != nil
288	}
289
290	func (c *serviceAsset) CallInit(ctx context.Context) error {
291	c.mu.Lock()
292	defer c.mu.Unlock()
293
294	switch {
295	case c.status == AssetStatusInjected:
296	c.status = AssetStatusInitialized
297	case c.singleton == nil:
298	panic(throw.IllegalState()) // double call of init is not allowed on non-singleton assets
299	default:
300	return nil
301	}
302
303	_, log := logutil.WithField(ctx, "component", "manager")
304
305	converted, ok := c.value.(Initer)
306	if ok {
307	log.Debug("initializing component", "component", c.Name())
308
309	return converted.Init(ctx)
310	}
311
312	return nil
313	}
314
315	func (c *serviceAsset) CallStart(ctx context.Context) error {
316	c.mu.Lock()
317	defer c.mu.Unlock()
318
319	switch {
320	case c.status == AssetStatusInitialized:
321	c.status = AssetStatusStarted
322	case c.singleton == nil:
323	panic(throw.IllegalState()) // double call of start is not allowed on non-singleton assets
324	default:
325	return nil
326	}
327
328	if !c.NeedStart() {
329	return nil
330	}
331
332	_, log := logutil.WithField(ctx, "component", "manager")
333
334	converted, ok := c.value.(Starter)
335	if ok {
336	log.Debug("starting component", "component", c.Name())
337
338	return converted.Start(ctx)
339	}
340
341	return nil
342	}
343
344	func (c *serviceAsset) CallStop(ctx context.Context) error {
345	c.mu.Lock()
346	defer c.mu.Unlock()
347
348	startTime := time.Now()
349
350	switch c.status {
351	case AssetStatusStarted, AssetStatusStopping:
352	c.status = AssetStatusStopped
353	default:
354	return nil
355	}
356
357	if !c.NeedStart() {
358	return nil
359	}
360
361	_, log := logutil.WithField(ctx, "component", "manager")
362
363	converted, ok := c.value.(Stopper)
364	if !ok {
365	return nil
366	}
367
368	log.Info("stopping component (CallStop)", "component", c.Name())
369
370	err := converted.Stop(ctx)
371	if err != nil {
372	return err
373	}
374
375	log.Info("stopped component (CallStop)", "component", c.Name())
376
377	dur := time.Since(startTime)
378	if dur > callStopWarningDuration {
379	logutil.FromContext(ctx).Warn("CallStop took too long",
380	"component", c.Name(),
381	"duration", dur,
382	)
383	}
384
385	return nil
386	}
387
388	// CallGracefulStop calls GracefulStop on the asset if it implements GracefulStopper.
389	// It also honors the ctx timeout.
390	// If the asset does not implement GracefulStopper then it calls Stop on the asset if it implements Stopper.
391	func (c *serviceAsset) CallGracefulStop(ctx context.Context) error {
392	c.mu.Lock()
393	defer c.mu.Unlock()
394
395	switch {
396	case c.status == AssetStatusStarted:
397	c.status = AssetStatusStopping
398	case c.singleton == nil:
399	panic(throw.IllegalState()) // double call of stop is not allowed on non-singleton assets
400	default:
401	return nil
402	}
403
404	if !c.NeedStart() {
405	return nil
406	}
407
408	_, log := logutil.WithField(ctx, "component", "manager")
409
410	converted, ok := c.value.(GracefulStopper)
411	if ok {
412	log.Info("gracefully stopping component", "component", c.Name())
413
414	task := async.WrapErrFunc[struct{}](converted.GracefulStop)
415	err := fn.ThirdOf3(async.Start(ctx, task).Wait(ctx))
416	if err != nil {
417	return err
418	}
419
420	c.status = AssetStatusGracefulStopped
421	log.Info("gracefully stopped component", "component", c.Name())
422
423	return nil
424	}
425
426	convertedStopper, ok := c.value.(Stopper)
427	if ok {
428	log.Info("stopping component (CallGracefulStop)", "component", c.Name())
429
430	err := convertedStopper.Stop(ctx)
431	if err != nil {
432	return err
433	}
434
435	c.status = AssetStatusStopped
436	log.Info("stopped component (CallGracefulStop)", "component", c.Name())
437
438	return nil
439	}
440
441	return nil
442	}
443
444	func (c *serviceAsset) CallHealthCheck(ctx context.Context) error {
445	c.mu.Lock()
446	defer c.mu.Unlock()
447
448	if c.status != AssetStatusStarted {
449	return throw.New("asset is not running", struct{ Asset string }{Asset: c.Name()})
450	}
451
452	converted, ok := c.value.(HealthChecker)
453	if !ok {
454	return nil
455	}
456	return converted.HealthCheck(ctx)
457	}
458
459	func (c *serviceAsset) CallDestroy(ctx context.Context) error {
460	c.mu.Lock()
461	defer c.mu.Unlock()
462
463	switch c.status {
464	case AssetStatusGracefulStopped, AssetStatusStopped:
465	c.status = AssetStatusDestroyed
466	default:
467	// No exit here
468	}
469
470	if !c.NeedStart() {
471	return nil
472	}
473
474	_, log := logutil.WithField(ctx, "component", "manager")
475
476	converted, ok := c.value.(Destroyer)
477	if !ok {
478	c.status = AssetStatusDestroyed // For CheckStopped correctness
479
480	return nil
481	}
482
483	log.Info("destroying component", "component", c.Name())
484
485	err := converted.Destroy(ctx)
486	if err != nil {
487	log.Info("destroyed component", "component", c.Name())
488	}
489
490	return err
491	}
492

asset_service.go

Source Files