Fixes issues in HealthCheckServiceInstanceListSupplier

[robotmrv]

• WebClient response leaks
• potential Scheduled task leak
• rework polling to plain reactor operators
• remove non atomic operations on ServiceInstance lists

original issue gh-683
related to gh-629

[robotmrv]

I do not see contract about ServiceInstanceListSupplier#get() Flux
Is it finite or could potentially be infinite?

[spencergibb]

yes

[robotmrv]

If it is finite (and assume that it has only one emit) what is the reason to return Flux instead of Mono from ServiceInstanceListSupplier#get?
If there is such contract maybe it is a good idea to document it, because it is not clear from current returning type

[robotmrv]

should some logs be added here?
if so debug or warn?

[spencergibb]

yes, debug is my first instinct.

[robotmrv]

should non standard codes be taken into account?
if so - it is better to use ClientResponse.rawStatusCode()

[spencergibb]

sure

[robotmrv]

as far as List is mutated should it be emitted on any update or only on first?
this could be rewritten to

List<ServiceInstance> result = new CopyOnWriteArrayList<>();
AtomicBoolean first = new AtomicBoolean(false);
return Flux.merge(checks).<List<ServiceInstance>>handle((alive, sink) -> {
	result.add(alive);
	if (first.compareAndSet(false, true)) {
		sink.next(result);
	}
})
.defaultIfEmpty(result);

[bsideup]

Why not just Flux.merge(checks).collectList().defaultIfEmpty(emptyList())?

[robotmrv]

this question was addressed to my intermediate implementation which mutated healthyInstances field.
I was trying to preserve initial behavior.
.collectList() will wait until all health checks are completed, in initial implementation List was emitted on the first healthy instance presence. As far as I understand it was made due to the fact that all health checks could take a long time (in current implementation it is this.healthCheck.getInterval() but in initial it could be infinity) but it should return healthy instances as soon as possible.

[robotmrv]

I've tried to make minimal changes to the current logic but
I've found testing of ping functionality a little bit inconvenient
HealthCheckServiceInstanceListSupplier starts ping implicitly and eventually I test just another stream which uses common state...
IMO It signals that this functionality must be either moved to a new class or just spitted into different methods.
So I would like to make HealthCheckServiceInstanceListSupplier an InitializingBean , and move start of pings out of constructor
and
make this

Flux<List<ServiceInstance>> instancesFlux = this.delegate.get().doOnNext(delegateInstances -> {
	this.instances = Collections.unmodifiableList(new ArrayList<>(delegateInstances));
});

and

Flux<List<ServiceInstance>> verifiedInstancesFlux = healthCheckFlux()
		.doOnNext((verifiedInstances -> {
			this.healthyInstances = verifiedInstances;
		}));

two distinct methods which are invoked from afterPropertiesSet()
does this make sense @spencergibb @OlgaMaciaszek

[robotmrv]

could remove state with such changes

private Flux<List<ServiceInstance>> aliveInstancesReplay;
...
//in the constructor 
this.aliveInstancesReplay = this.delegate.get()
				.switchMap(serviceInstances -> {
					return healthCheckFlux(serviceInstances).map(alive -> alive.isEmpty() ? serviceInstances : alive);
				})
				.replay(1)
				.autoConnect()
				.onBackpressureLatest();
...
protected Flux<List<ServiceInstance>> healthCheckFlux(List<ServiceInstance> instances) {
...//the same logic but uses `instances` method arg instead of class field
}
...
public void afterPropertiesSet() {
    this.healthCheckDisposable = aliveInstancesReplay.subscribe();
}

//and `get()` become just 
@Override
public Flux<List<ServiceInstance>> get() {
	return aliveInstancesReplay;
}

less state and it is easier to test

[robotmrv]

downside - requires protected method signature change, but it was introduced recently so it is not a problem

[robotmrv]

@spencergibb @OlgaMaciaszek
Changes are ready for review.
Initial issue #629 is closed
Should it be reopened or should I create new issue?

[spencergibb]

@robotmrv this PR is fine, no need to open anything else.

[robotmrv]

@spencergibb @OlgaMaciaszek
kindly reminder that this PR is ready for review

[bsideup]

can be replaced with .filter(isHealthy -> isHealthy)

[robotmrv]

@bsideup , not really.
it can be replaced by

.filter(isHealthy -> isHealthy)
.map(it -> instance)

but is filter + map more efficient then handle? I thought handle is just for such cases

[OlgaMaciaszek]

@robotmrv Reviewing after merge as requested. Looks definitely good to me :) . Have learnt some better ways of doing certain things with Reactor in this PR. Also, like the way you've used StepVerifier in test. We should probably use it more frequently in Cloud projects.

Not sure why you've added the setup method in the test, though - AFAIK, works same with field instantiation. Also, why are you adding a local variable assignment in afterPropertiesSet() and in destroy() methods? I assume it's for performance improvement?

The logic looks good to me. Will, however, change some cosmetics things, such as unnecessary use of this. and abbreviatory variable names to adjust to Spring Cloud team conventions.

[robotmrv]

@OlgaMaciaszek

Not sure why you've added the setup method in the test, though - AFAIK, works same with field instantiation.

it creates new instance for each test so previous test will not influence next one, also it is just to avoid instantiation in each test method.

Also, why are you adding a local variable assignment in afterPropertiesSet() and in destroy() methods? I assume it's for performance improvement?

It is not for performance reasons. I just dispose that was checked

Will, however, change some cosmetics things, such as unnecessary use of this.

No problem it is redundant but just looks more clear to me, especially when I look at code on github. Recently spring-security team asked me to add this. in another PR, I thought you have similar conventions.