58
}
59
60
>
func (o *ownership) start(controller *ControllerImpl) {
ownership.go
61
>
o.goros.Go(func(ctx context.Context) error {
62
>
o.eventLoop(ctx)
63
>
return nil
64
>
})
65
66
>
o.goros.Go(func(ctx context.Context) error {
ownership.go
67
>
o.acquireLoop(ctx, controller)
68
>
return nil
69
>
})
70
71
>
if err := o.historyServiceResolver.AddListener(
ownership.go
72
>
shardControllerMembershipUpdateListenerName,
73
>
o.membershipUpdateCh,
74
>
); err != nil {
75
o.logger.Fatal("Error adding listener", tag.Error(err))
76
}
77
}
78
79
>
func (o *ownership) eventLoop(ctx context.Context) {
ownership.go
80
>
acquireTicker := time.NewTicker(o.config.AcquireShardInterval())
81
>
defer acquireTicker.Stop()
82
>
83
>
for {
84
>
select {
85
case <-ctx.Done():
86
return