80 lines
2.2 KiB
Go
80 lines
2.2 KiB
Go
package serverimp
|
|
|
|
import (
|
|
"time"
|
|
|
|
"github.com/tech/sendico/discovery/internal/appversion"
|
|
"github.com/tech/sendico/pkg/discovery"
|
|
"github.com/tech/sendico/pkg/merrors"
|
|
msg "github.com/tech/sendico/pkg/messaging"
|
|
msgproducer "github.com/tech/sendico/pkg/messaging/producer"
|
|
"github.com/tech/sendico/pkg/mservice"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
func (i *Imp) startDiscovery(cfg *config) error {
|
|
if cfg == nil || cfg.Messaging == nil || cfg.Messaging.Driver == "" {
|
|
//nolint:wrapcheck
|
|
return merrors.InvalidArgument("discovery service: messaging configuration is required", "messaging")
|
|
}
|
|
|
|
broker, err := msg.CreateMessagingBroker(i.logger.Named("discovery_bus"), cfg.Messaging)
|
|
if err != nil {
|
|
return err //nolint:wrapcheck
|
|
}
|
|
|
|
i.logger.Info("Discovery messaging broker ready", zap.String("messaging_driver", string(cfg.Messaging.Driver)))
|
|
producer := msgproducer.NewProducer(i.logger.Named("discovery_producer"), broker)
|
|
|
|
registry := discovery.NewRegistry()
|
|
|
|
var registryOpts []discovery.RegistryOption
|
|
|
|
if cfg.Registry != nil && cfg.Registry.KVTTLSeconds != nil {
|
|
ttlSeconds := *cfg.Registry.KVTTLSeconds
|
|
if ttlSeconds < 0 {
|
|
i.logger.Warn("Discovery registry TTL is negative, disabling TTL", zap.Int("ttl_seconds", ttlSeconds))
|
|
ttlSeconds = 0
|
|
}
|
|
|
|
registryOpts = append(registryOpts, discovery.WithRegistryKVTTL(time.Duration(ttlSeconds)*time.Second))
|
|
}
|
|
|
|
svc, err := discovery.NewRegistryService(i.logger, broker, producer, registry, mservice.Discovery, registryOpts...)
|
|
if err != nil {
|
|
return err //nolint:wrapcheck
|
|
}
|
|
|
|
svc.Start()
|
|
i.registrySvc = svc
|
|
|
|
announce := discovery.Announcement{
|
|
Service: mservice.Discovery,
|
|
InstanceID: discovery.InstanceID(),
|
|
Operations: []string{discovery.OperationDiscoveryLookup},
|
|
Version: appversion.Create().Short(),
|
|
}
|
|
i.announcer = discovery.NewAnnouncer(i.logger, producer, mservice.Discovery, announce)
|
|
i.announcer.Start()
|
|
|
|
i.logger.Info("Discovery registry service started", zap.String("messaging_driver", string(cfg.Messaging.Driver)))
|
|
|
|
return nil
|
|
}
|
|
|
|
func (i *Imp) stopDiscovery() {
|
|
if i == nil {
|
|
return
|
|
}
|
|
|
|
if i.announcer != nil {
|
|
i.announcer.Stop()
|
|
i.announcer = nil
|
|
}
|
|
|
|
if i.registrySvc != nil {
|
|
i.registrySvc.Stop()
|
|
i.registrySvc = nil
|
|
}
|
|
}
|