package com.gofish.sentiment.newslinker;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.eventbus.MessageConsumer;
import io.vertx.core.json.JsonObject;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.servicediscovery.Record;
import io.vertx.servicediscovery.ServiceDiscovery;
import io.vertx.servicediscovery.types.EventBusService;
import io.vertx.serviceproxy.ProxyHelper;

import java.util.Optional;

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

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

    private MessageConsumer<JsonObject> messageConsumer;
    private ServiceDiscovery serviceDiscovery;
    private Record record;

    @Override
    public void start(Future<Void> startFuture) throws Exception {
        LOG.info("Bringing up NewsLinkerVerticle");

        JsonObject config = Optional.ofNullable(config())
                .orElseThrow(() -> new RuntimeException("Could not load linker verticle configuration"));

        NewsLinkerService newsLinkerService = NewsLinkerService.create(vertx, config);
        messageConsumer = ProxyHelper.registerService(NewsLinkerService.class, vertx, newsLinkerService, NewsLinkerService.ADDRESS);
        serviceDiscovery = ServiceDiscovery.create(vertx);
        record = EventBusService.createRecord(NewsLinkerService.NAME, NewsLinkerService.ADDRESS, NewsLinkerService.class);

        serviceDiscovery.publish(record, resultHandler -> {
            if (resultHandler.succeeded()) {
                LOG.info("Published news linker service successfully");
                startFuture.complete();
            }
            else {
                startFuture.fail(resultHandler.cause());
            }
        });
    }

    @Override
    public void stop(Future<Void> stopFuture) throws Exception {
        Future<Void> recordUnpublishFuture = Future.future();
        Future<Void> messageConsumerUnregisterFuture = Future.future();

        serviceDiscovery.unpublish(record.getRegistration(), recordUnpublishFuture.completer());
        messageConsumer.unregister(messageConsumerUnregisterFuture.completer());

        recordUnpublishFuture.compose(v -> messageConsumerUnregisterFuture).setHandler(v -> {
            serviceDiscovery.close();
            if (v.succeeded()) {
                stopFuture.complete();
            }
            else {
                stopFuture.fail(v.cause());
            }
        });
    }
}

@@ -39,47 +39,8 @@
 r;%0A%0A
-import io.vertx.core.AbstractVerticle;%0A
 impo
@@ -250,31 +250,112 @@
 rtx.
-servicediscovery.Record
+rx.java.ObservableFuture;%0Aimport io.vertx.rx.java.RxHelper;%0Aimport io.vertx.rxjava.core.AbstractVerticle
 ;%0Aim
@@ -360,32 +360,39 @@
 import io.vertx.
+rxjava.
 servicediscovery
@@ -422,24 +422,31 @@
 rt io.vertx.
+rxjava.
 servicedisco
@@ -473,16 +473,57 @@
 ervice;%0A
+import io.vertx.servicediscovery.Record;%0A
 import i
@@ -1209,46 +1209,124 @@
     
-NewsLinkerService newsLinkerService = 
+com.gofish.sentiment.newslinker.rxjava.NewsLinkerService newsLinkerService = com.gofish.sentiment.newslinker.rxjava.
 News
@@ -1446,16 +1446,30 @@
 s, vertx
+.getDelegate()
 , newsLi
@@ -1479,16 +1479,30 @@
 rService
+.getDelegate()
 , NewsLi
@@ -1523,16 +1523,17 @@
 DRESS);%0A
+%0A
         
@@ -1580,19 +1580,98 @@
 te(vertx
-);%0A
+, serviceDiscovery -%3E %7B%0A            LOG.info(%22Service Discovery intialised%22);%0A    
         
@@ -1778,24 +1778,34 @@
 ervice.class
+.getName()
 );%0A%0A        
@@ -1796,32 +1796,36 @@
 me());%0A%0A        
+    
 serviceDiscovery
@@ -1825,17 +1825,19 @@
 scovery.
-p
+rxP
 ublish(r
@@ -1845,234 +1845,69 @@
 cord
-, resultHandler -%3E %7B%0A            if (resultHandler.succeeded()) %7B%0A                LOG.info(%22Published news linker service successfully%22);%0A                startFuture.complete();%0A            %7D%0A            else %7B%0A               
+)%0A                    .subscribe(r -%3E startFuture.complete(),
  sta
@@ -1918,51 +1918,16 @@
 ture
-.
+::
 fail
-(resultHandler.cause());%0A            %7D
+);
 %0A   
@@ -2033,135 +2033,85 @@
     
-Future%3CVoid%3E recordUnpublishFuture = Future.future();%0A        Future%3CVoid%3E messageConsumerUnregisterFuture = Future.f
+ObservableFuture%3CVoid%3E messageConsumerObservable = new ObservableF
 uture
+%3C%3E
 ();%0A
-%0A
     
@@ -2131,17 +2131,19 @@
 scovery.
-u
+rxU
 npublish
@@ -2171,331 +2171,268 @@
 on()
-, recordUnpublishFuture.completer());%0A        messageConsumer.unregister(messageConsumerUnregisterFuture.completer());%0A%0A        recordUnpublishFuture.compose(v -%3E messageConsumerUnregisterFuture).setHandler(v -%3E %7B%0A            serviceDiscovery.close();%0A            if (v.succeeded()) %7B%0A                stopFuture.complete();
+)%0A                .flatMapObservable(v -%3E %7B%0A                    messageConsumer.unregister(messageConsumerObservable.toHandler());%0A                    return messageConsumerObservable;%0A                %7D)%0A                .doOnNext(v -%3E serviceDiscovery.close())
 %0A   
@@ -2444,96 +2444,56 @@
     
-%7D%0A
     
-        else %7B%0A                stopFuture.fail(v.cause());%0A            %7D%0A        %7D
+.subscribe(RxHelper.toSubscriber(stopFuture)
 );%0A 

Migrate to more rx focussed implementation