mirror of
https://github.com/go-kratos/kratos.git
synced 2025-04-11 11:42:10 +02:00
feat(registry/consul): add WithHeartbeat option (#1738)
This commit is contained in:
parent
1e749de5db
commit
d2f8b45603
@ -23,6 +23,8 @@ type Client struct {
|
||||
resolver ServiceResolver
|
||||
// healthcheck time interval in seconds
|
||||
healthcheckInterval int
|
||||
// heartbeat enable heartbeat
|
||||
heartbeat bool
|
||||
}
|
||||
|
||||
// NewClient creates consul client
|
||||
@ -31,6 +33,7 @@ func NewClient(cli *api.Client) *Client {
|
||||
cli: cli,
|
||||
resolver: defaultResolver,
|
||||
healthcheckInterval: 10,
|
||||
heartbeat: true,
|
||||
}
|
||||
c.ctx, c.cancel = context.WithCancel(context.Background())
|
||||
return c
|
||||
@ -123,18 +126,20 @@ func (c *Client) Register(_ context.Context, svc *registry.ServiceInstance, enab
|
||||
return err
|
||||
}
|
||||
_ = c.cli.Agent().UpdateTTL("service:"+svc.ID, "pass", "pass")
|
||||
go func() {
|
||||
ticker := time.NewTicker(time.Second * time.Duration(c.healthcheckInterval))
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
_ = c.cli.Agent().UpdateTTL("service:"+svc.ID, "pass", "pass")
|
||||
case <-c.ctx.Done():
|
||||
return
|
||||
if c.heartbeat {
|
||||
go func() {
|
||||
ticker := time.NewTicker(time.Second * time.Duration(c.healthcheckInterval))
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
_ = c.cli.Agent().UpdateTTL("service:"+svc.ID, "pass", "pass")
|
||||
case <-c.ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
@ -26,6 +26,15 @@ func WithHealthCheck(enable bool) Option {
|
||||
}
|
||||
}
|
||||
|
||||
// WithHeartbeat enable or disable heartbeat
|
||||
func WithHeartbeat(enable bool) Option {
|
||||
return func(o *Registry) {
|
||||
if o.cli != nil {
|
||||
o.cli.heartbeat = enable
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WithServiceResolver with endpoint function option.
|
||||
func WithServiceResolver(fn ServiceResolver) Option {
|
||||
return func(o *Registry) {
|
||||
|
@ -25,7 +25,7 @@ func tcpServer(t *testing.T, lis net.Listener) {
|
||||
}
|
||||
|
||||
func TestRegister(t *testing.T) {
|
||||
addr := fmt.Sprintf("%s:8081", getIntranetIP())
|
||||
addr := fmt.Sprintf("%s:9091", getIntranetIP())
|
||||
lis, err := net.Listen("tcp", addr)
|
||||
if err != nil {
|
||||
t.Errorf("listen tcp %s failed!", addr)
|
||||
@ -38,7 +38,12 @@ func TestRegister(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("create consul client failed: %v", err)
|
||||
}
|
||||
r := New(cli)
|
||||
opts := []Option{
|
||||
WithHeartbeat(true),
|
||||
WithHealthCheck(true),
|
||||
WithHealthCheckInterval(5),
|
||||
}
|
||||
r := New(cli, opts...)
|
||||
assert.Nil(t, err)
|
||||
version := strconv.FormatInt(time.Now().Unix(), 10)
|
||||
svc := ®istry.ServiceInstance{
|
||||
|
Loading…
x
Reference in New Issue
Block a user