/*
 * Knot.x - Reactive microservice assembler - Templating Engine Verticle
 *
 * Copyright (C) 2016 Cognifide Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.cognifide.knotx.engine;

import com.cognifide.knotx.api.RenderRequest;
import com.cognifide.knotx.api.RenderResponse;
import com.cognifide.knotx.engine.impl.TemplateEngine;

import io.vertx.core.Context;
import io.vertx.core.Vertx;
import io.vertx.core.http.HttpClientOptions;
import io.vertx.core.json.JsonObject;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.rxjava.core.AbstractVerticle;
import io.vertx.rxjava.core.eventbus.EventBus;
import io.vertx.rxjava.core.eventbus.Message;
import io.vertx.rxjava.core.http.HttpClient;
import rx.Observable;

public class TemplateEngineVerticle extends AbstractVerticle {

  private static final Logger LOGGER = LoggerFactory.getLogger(TemplateEngineVerticle.class);

  private TemplateEngine templateEngine;

  private TemplateEngineConfiguration configuration;

  private String address;

  private HttpClient httpClient;

  @Override
  public void init(Vertx vertx, Context context) {
    super.init(vertx, context);
    JsonObject config = config().getJsonObject("config");
    this.address = config.getString("address");

    configuration = new TemplateEngineConfiguration(config);

    final JsonObject clientOptions = configuration.getClientOptions();
    httpClient = clientOptions.isEmpty() ?
        this.vertx.createHttpClient() : this.vertx.createHttpClient(new HttpClientOptions(clientOptions));
    templateEngine = new TemplateEngine(httpClient, configuration);

  }

  @Override
  public void start() throws Exception {
    LOGGER.debug("Registered <{}>", this.getClass().getSimpleName());

    EventBus eventBus = vertx.eventBus();
    Observable<Message<JsonObject>> messageObservable = eventBus.<JsonObject>consumer(address).toObservable();

    messageObservable
        .doOnNext(this::traceMessage)
        .subscribe(
            msg -> templateEngine.process(new RenderRequest(msg.body()))
                .subscribe(
                    result -> msg.reply(RenderResponse.success(result).toJsonObject()),
                    error -> {
                      LOGGER.error("Error happened", error);
                      msg.reply(RenderResponse.error(error.getMessage()).toJsonObject());
                    }
                )
        );
  }

  @Override
  public void stop() throws Exception {
    LOGGER.debug("Stopping TemplateEngineVerticle and stopping http client");
    httpClient.close();
  }

  private void traceMessage(Message<JsonObject> message) {
    if (LOGGER.isTraceEnabled()) {
      LOGGER.trace("Got message from <{}> with value <{}>", message.replyAddress(), message.body().encodePrettily());
    }
  }
}

@@ -2442,17 +2442,16 @@
 able();%0A
-%0A
     mess
@@ -2511,39 +2511,24 @@
        .
-subscribe(%0A            
+flatMap(
 msg -%3E t
@@ -2596,44 +2596,18 @@
     
-    .subscribe(%0A                    
+.doOnNext(
 resu
@@ -2668,17 +2668,17 @@
 bject())
-,
+)
 %0A       
@@ -2682,24 +2682,27 @@
         
-        
+.doOnError(
 error -%3E
@@ -2704,24 +2704,16 @@
 or -%3E %7B%0A
-        
         
@@ -2771,24 +2771,16 @@
         
-   
-     
 msg.repl
@@ -2855,26 +2855,11 @@
     
-        %7D%0A        
+%7D)%0A
     
@@ -2872,16 +2872,27 @@
         
+.subscribe(
 );%0A  %7D%0A%0A

Change template engine subscription