package com.gofish.sentiment.storage;

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.ext.mongo.MongoClient;
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 StorageVerticle extends AbstractVerticle {

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

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

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

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

        // We bring up the mongo client here, even though it is utilised in the worker verticles, and not here.
        // This is to enable us to keep a connection open, and simply close it when this verticle closes, rather than
        // closing it everytime the worker verticle stops (which is after every DB call).
        mongo = MongoClient.createShared(vertx, config);

        // Initialise a service proxy and publish it for service discovery
        StorageService storageService = StorageService.create(vertx, config());
        messageConsumer = ProxyHelper.registerService(StorageService.class, vertx, storageService, StorageService.ADDRESS);
        serviceDiscovery = ServiceDiscovery.create(vertx);
        record = EventBusService.createRecord(StorageService.NAME, StorageService.ADDRESS, StorageService.class);

        serviceDiscovery.publish(record, resultHandler -> {
            if (resultHandler.succeeded()) {
                startFuture.complete();
            }
            else {
                startFuture.fail(resultHandler.cause());
            }
        });
    }

    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();
            mongo.close();

            if (v.succeeded()) {
                stopFuture.complete();
            }
            else {
                stopFuture.fail(v.cause());
            }
        });
    }
}

@@ -36,47 +36,8 @@
 e;%0A%0A
-import io.vertx.core.AbstractVerticle;%0A
 impo
@@ -247,70 +247,158 @@
 rtx.
-ext.mongo.MongoClient;%0Aimport io.vertx.servicediscovery.Record
+rx.java.ObservableFuture;%0Aimport io.vertx.rx.java.RxHelper;%0Aimport io.vertx.rxjava.core.AbstractVerticle;%0Aimport io.vertx.rxjava.ext.mongo.MongoClient
 ;%0Aim
@@ -403,32 +403,39 @@
 import io.vertx.
+rxjava.
 servicediscovery
@@ -465,24 +465,31 @@
 rt io.vertx.
+rxjava.
 servicedisco
@@ -516,16 +516,57 @@
 ervice;%0A
+import io.vertx.servicediscovery.Record;%0A
 import i
@@ -1710,40 +1710,112 @@
     
-StorageService storageService = 
+com.gofish.sentiment.storage.rxjava.StorageService storageService = com.gofish.sentiment.storage.rxjava.
 Stor
@@ -1931,16 +1931,30 @@
 s, vertx
+.getDelegate()
 , storag
@@ -1961,16 +1961,30 @@
 eService
+.getDelegate()
 , Storag
@@ -2002,16 +2002,17 @@
 DRESS);%0A
+%0A
         
@@ -2059,19 +2059,98 @@
 te(vertx
-);%0A
+, serviceDiscovery -%3E %7B%0A            LOG.info(%22Service Discovery intialised%22);%0A    
         
@@ -2248,28 +2248,42 @@
 ervice.class
-);%0A%0A
+.getName());%0A%0A    
         serv
@@ -2295,17 +2295,19 @@
 scovery.
-p
+rxP
 ublish(r
@@ -2315,162 +2315,48 @@
 cord
-, resultHandler -%3E %7B%0A            if (resultHandler.succeeded()) %7B%0A                startFuture.complete();%0A            %7D%0A            else %7B%0A               
+).subscribe(r -%3E startFuture.complete(),
  sta
@@ -2367,51 +2367,16 @@
 ture
-.
+::
 fail
-(resultHandler.cause());%0A            %7D
+);
 %0A   
@@ -2391,16 +2391,30 @@
 %0A    %7D%0A%0A
+    @Override%0A
     publ
@@ -2482,135 +2482,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
     
@@ -2580,17 +2580,19 @@
 scovery.
-u
+rxU
 npublish
@@ -2620,46 +2620,64 @@
 on()
-, recordUnpublishFuture.completer());%0A
+)%0A                .flatMapObservable(v -%3E %7B%0A            
     
@@ -2726,48 +2726,47 @@
 umer
-UnregisterFuture.complet
+Observable.toHandl
 er());%0A
-%0A
+      
         
 reco
@@ -2765,105 +2765,99 @@
     
-recordUnpublishFuture.compose(v -%3E messageConsumerUnregisterFuture).setHandler(v -%3E %7B%0A           
+      return messageConsumerObservable;%0A                %7D)%0A                .doOnNext(v -%3E %7B
  ser
@@ -2878,28 +2878,16 @@
 close();
-%0A           
  mongo.c
@@ -2897,182 +2897,72 @@
 e();
-%0A%0A            if (v.succeeded()) %7B%0A                stopFuture.complete();%0A            %7D%0A            else %7B%0A                stopFuture.fail(v.cause());%0A            %7D%0A        %7D
+ %7D)%0A                .subscribe(RxHelper.toSubscriber(stopFuture)
 );%0A 

Migrate to more rx focussed implementation