package nl.jpoint.vertx.mod.deploy.aws;


import com.amazonaws.AmazonClientException;
import com.amazonaws.regions.Region;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.elasticloadbalancing.AmazonElasticLoadBalancingAsyncClient;
import com.amazonaws.services.elasticloadbalancing.model.*;
import com.amazonaws.util.EC2MetadataUtils;
import nl.jpoint.vertx.mod.deploy.DeployConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.Observable;

import java.util.Optional;

import static rx.Observable.just;

public class AwsElbUtil {
    private static final Logger LOG = LoggerFactory.getLogger(AwsElbUtil.class);
    private final AmazonElasticLoadBalancingAsyncClient elbAsyncClient;
    private final String instanceId;

    public AwsElbUtil(DeployConfig config) {
        this.elbAsyncClient = new AmazonElasticLoadBalancingAsyncClient();
        this.elbAsyncClient.setRegion(Region.getRegion(Regions.fromName(config.getAwsRegion())));
        this.instanceId = EC2MetadataUtils.getInstanceId();
    }

    public Observable<String> registerInstanceWithLoadBalancer(String loadBalancer) throws AwsException {
        if (instanceId == null || loadBalancer == null) {
            LOG.error("Unable to register instance {}, on load balancer {}.", instanceId, loadBalancer);
            throw new IllegalStateException();
        }
        try {
            return Observable.from(elbAsyncClient.registerInstancesWithLoadBalancerAsync(new RegisterInstancesWithLoadBalancerRequest().withLoadBalancerName(loadBalancer).withInstances(new Instance().withInstanceId(instanceId))))
                    .flatMap(x -> Observable.just(loadBalancer));
        } catch (AmazonClientException e) {
            LOG.error("Error executing request {}.", e);
            throw new AwsException(e);
        }

    }

    public Observable<String> deRegisterInstanceFromLoadbalancer(String loadBalancer) throws AwsException {

        if (instanceId == null || loadBalancer == null) {
            LOG.error("Unable to register instance {}, on load balancer {}.", instanceId, loadBalancer);
            throw new IllegalStateException();
        }

        try {
            return Observable.from(elbAsyncClient.deregisterInstancesFromLoadBalancerAsync(new DeregisterInstancesFromLoadBalancerRequest().withLoadBalancerName(loadBalancer).withInstances(new Instance().withInstanceId(instanceId))))
                    .flatMap(x -> Observable.just(loadBalancer));
        } catch (AmazonClientException e) {
            LOG.error("Error executing request {}.", e);
            throw new AwsException(e);
        }
    }


    public Observable<AwsState> pollForInstanceState(final String loadBalancer) throws AwsException {
        try {
            return Observable.from(elbAsyncClient.describeInstanceHealthAsync(new DescribeInstanceHealthRequest().withLoadBalancerName(loadBalancer).withInstances(new Instance().withInstanceId(instanceId))))
                    .flatMap(result -> {
                        Optional<InstanceState> state = result.getInstanceStates().stream().filter(i -> i.getInstanceId().equals(instanceId)).findFirst();
                        return just(state.isPresent() ? AwsState.map(state.get().getState()) : AwsState.UNKNOWN);
                    });
        } catch (AmazonClientException e) {
            LOG.error("Error executing request {}.", e);
            throw new AwsException(e);
        }
    }

    public String getInstanceId() {
        return instanceId;
    }
}

@@ -1667,32 +1667,113 @@
 t(loadBalancer))
+%0A                    .doOnError(t -%3E LOG.error(%22Error executing request %7B%7D.%22, t))
 ;%0A        %7D catc

Add doOnError