toClusterKey ClusterShardKey,
dc *configs.Config,
defer streamStopper()
streamRetryPolicy := backoff.NewExponentialRetryPolicy(500 * time.Millisecond).
WithMaximumAttempts(dc.ReplicationStreamEventLoopRetryMaxAttempts()).
WithMaximumInterval(time.Second * 2)
ops := func() error {
if ctx.Err() != nil {
return ctx.Err()
}
if err != nil {
var streamError *StreamError
if errors.As(err, &streamError) {