package com.gofish.sentiment.api;

import com.gofish.sentiment.sentimentservice.SentimentService;
import io.vertx.core.Future;
import io.vertx.core.json.JsonObject;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.rx.java.SingleOnSubscribeAdapter;
import io.vertx.rxjava.core.AbstractVerticle;
import io.vertx.rxjava.core.http.HttpServerResponse;
import io.vertx.rxjava.ext.web.Router;
import io.vertx.rxjava.ext.web.handler.LoggerHandler;
import io.vertx.rxjava.servicediscovery.ServiceDiscovery;
import io.vertx.rxjava.servicediscovery.ServiceReference;
import rx.Single;

/**
 * @author Luke Herron
 */
public class APIGatewayVerticle extends AbstractVerticle {

    private static final Logger LOG = LoggerFactory.getLogger(APIGatewayVerticle.class);

    private ServiceDiscovery serviceDiscovery;

    @Override
    public void start(Future<Void> startFuture) throws Exception {
        serviceDiscovery = ServiceDiscovery.create(vertx);

        Router router = Router.router(vertx);
        router.route().handler(LoggerHandler.create());

        router.route("/search").blockingHandler(requestHandler -> {
            String q = requestHandler.request().params().get("q");
            if (q != null && !q.isEmpty()) {
                final String query = q.toLowerCase();
                HttpServerResponse response = requestHandler.response();

                rxGetSentiment(query).subscribe(
                        result -> {
                            response.end(String.valueOf(result));
                            LOG.info("Finished retrieving sentiment");
                        },
                        failure -> {
                            LOG.error(failure.getMessage(), failure);
                            requestHandler.fail(failure);
                        });
            }
            else {
                requestHandler.fail(400); // Bad request
            }
        });

        // Setup a failure handler
        router.route("/*").failureHandler(failureHandler -> {
            HttpServerResponse response = failureHandler.response();

            LOG.error(failureHandler.failure());

            int statusCode = failureHandler.statusCode();
            if (statusCode < 0) {
                response.end("Unknown failure occurred");
            }
            else {
                response.setStatusCode(statusCode).end("Sorry! Not today");
            }
        });

        // Launch server and start listening
        vertx.createHttpServer().requestHandler(router::accept).rxListen().subscribe(
                result -> startFuture.complete(),
                failure -> startFuture.fail(failure));
    }

    private Single<JsonObject> rxGetSentiment(String query) {
        return serviceDiscovery.rxGetRecord(record -> record.getName().equals(SentimentService.NAME))
                .map(serviceDiscovery::getReference)
                .map(ServiceReference::<SentimentService>get)
                .flatMap(service -> Single.create(new SingleOnSubscribeAdapter<JsonObject>(handler ->
                        service.getSentiment(query, handler)))
                        .doOnEach(notification -> ServiceDiscovery.releaseServiceObject(serviceDiscovery, service)));
    }

    @Override
    public void stop() throws Exception {
        serviceDiscovery.close();
    }
}

@@ -73,16 +73,23 @@
 service.
+rxjava.
 Sentimen
@@ -127,16 +127,53 @@
 Future;%0A
+import io.vertx.core.json.JsonArray;%0A
 import i
@@ -627,42 +627,185 @@
 ery.
-ServiceReference;%0Aimport rx.Single
+types.EventBusService;%0Aimport io.vertx.servicediscovery.Record;%0Aimport rx.Observable;%0Aimport rx.Single;%0A%0Aimport java.util.concurrent.TimeUnit;%0Aimport java.util.stream.Collectors
 ;%0A%0A/
@@ -1117,68 +1117,8 @@
 n %7B%0A
-        serviceDiscovery = ServiceDiscovery.create(vertx);%0A%0A
     
@@ -1551,135 +1551,184 @@
 etSe
-ntiment(query).subscribe(%0A                        result -%3E %7B%0A                            response.end(String.valueOf
+rvice(SentimentService.name(), SentimentService.class)%0A                        .flatMap(service -%3E service.rxGetSentiment(query))%0A                        .subscribe
 (result
-));
+ -%3E %7B
 %0A   
@@ -1823,10 +1823,42 @@
     
-%7D,
+    response.end(result.encode());
 %0A   
@@ -1869,32 +1869,35 @@
                 
+ %7D,
  failure -%3E %7B%0A  
@@ -2682,521 +2682,925 @@
     
-// Launch server and start listening%0A        vertx.createHttpServer().requestHandler(router::accept).rxListen().subscribe(%0A                result -%3E startFuture.complete(),%0A                failure -%3E startFuture.fail(failure
+JsonArray serviceDependencies = new JsonArray().add(SentimentService.name());%0A        serviceDiscovery = ServiceDiscovery.create(vertx, serviceDiscovery -%3E %7B%0A            LOG.info(%22Service Discovery initialised%22);%0A%0A            serviceDiscovery.rxGetRecords((JsonObject) null)%0A                    .doOnSuccess(r -%3E LOG.info(%22Searching for service dependencies%22))%0A                    .map(records -%3E records.stream().map(Record::getName).collect(Collectors.toList()).containsAll(serviceDependencies.getList()
 ))
-;
 %0A    
-%7D%0A%0A
+   
     
-private Single%3CJsonObject%3E rxGetSentiment(String query) %7B%0A        return serviceDiscovery.rxGetRecord(record
+         .flatMap(dependenciesMet -%3E dependenciesMet ? Single.%3CVoid%3Ejust(null) : Single.error(new Throwable()))%0A                    .retryWhen(errors
  -%3E 
-record.getName().equals(SentimentService.NAME))%0A                .map(serviceDiscov
+errors.flatMap(error -%3E Observable.timer(5, TimeUnit.SECONDS)))%0A                    .flatMap(v -%3E vertx.createHttpServer().requestHandler(rout
 er
-y
 ::
-getReference)%0A                .map(ServiceReference::%3CSentimentService%3Eget)
+accept).rxListen())%0A                    .subscribe(r -%3E startFuture.complete(), startFuture::fail);
 %0A   
@@ -3608,35 +3608,110 @@
     
+%7D);%0A
     
+%7D%0A%0A
     
-.flatMap(service -%3E
+private %3CT%3E Single%3CT%3E rxGetService(String recordName, Class%3CT%3E clazz) %7B%0A%0A        return
  Sin
@@ -3754,39 +3754,18 @@
 ter%3C
-JsonObject%3E(handler -%3E%0A        
+T%3E(fut -%3E%0A
     
@@ -3772,25 +3772,33 @@
             
-s
+EventBusS
 ervice.getSe
@@ -3801,148 +3801,104 @@
 etSe
-ntiment(query, handler)))%0A                        .doOnEach(notification -%3E ServiceDiscovery.releaseServiceObject(serviceDiscovery, service)
+rviceProxy(serviceDiscovery, record -%3E record.getName().equals(recordName), clazz, fut)%0A        
 ));%0A

Migrate to more rx focussed implementation