84
joinTime time.Time,
85
replicaPoints int,
87
>
lifecycleCtx, lifecycleCancel := context.WithCancel(context.Background())
88
>
lifecycleCtx = headers.SetCallerInfo(
89
>
lifecycleCtx,
90
>
headers.SystemBackgroundHighCallerInfo,
91
>
)
92
>
hostID, _ := uuid.New().MarshalBinary()
93
>
// MarshalBinary should never error.
94
>
95
>
rpo := &monitor{
96
>
status: common.DaemonStatusInitialized,
97
>
98
>
lifecycleCtx: lifecycleCtx,
99
>
lifecycleCancel: lifecycleCancel,
100
>
101
>
serviceName: serviceName,
102
>
services: services,
103
>
rp: rp,
104
>
rings: make(map[primitives.ServiceName]*serviceResolver),
105
>
replicaPoints: replicaPoints,
106
>
logger: logger,
107
>
metadataManager: metadataManager,
108
>
broadcastHostPortResolver: broadcastHostPortResolver,
109
>
hostID: hostID,
110
>
initialized: future.NewFuture[struct{}](),
111
>
maxJoinDuration: maxJoinDuration,
112
>
propagationTime: propagationTime,
113
>
joinTime: joinTime,
114
>
}
115
>
for service, port := range services {
116
>
rpo.rings[service] = newServiceResolver(service, port, rp, replicaPoints, logger)
117
>
}
118
>
return rpo
119
}
120