package org.cstamas.vertx.content.examples;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.Message;
import io.vertx.core.file.AsyncFile;
import io.vertx.core.file.OpenOptions;
import io.vertx.core.json.JsonObject;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.core.streams.Pump;
import io.vertx.core.streams.ReadStream;
import org.cstamas.vertx.content.ContentManager;

/**
 * Content receiver verticle, that receives the file and saves it under path it got from sender appending it with
 * {@code
 * .received}.
 */
public class ContentReceiverVerticle
    extends AbstractVerticle
    implements Handler<Message<JsonObject>>
{
  private static final Logger log = LoggerFactory.getLogger(ContentReceiverVerticle.class);

  public static final String ADDRESS = "content.receiver";

  private ContentManager contentManager;

  @Override
  public void start(final Future<Void> startFuture) throws Exception {
    contentManager = Utility.create(vertx, config());
    vertx.eventBus().consumer(ADDRESS, this);
    super.start(startFuture);
  }

  @Override
  public void handle(final Message<JsonObject> message) {
    JsonObject contentHandle = message.body();
    String path = contentHandle.getString("path") + ".received";
    log.info("R: " + path);
    contentManager.receive(
        contentHandle,
        streamResult -> {
          if (streamResult.failed()) {
            log.error("R: ", streamResult.cause());
            message.fail(500, streamResult.cause().getMessage());
          }
          else {
            vertx.fileSystem().open(
                path,
                new OpenOptions(),
                oh -> {
                  if (oh.failed()) {
                    log.error("E:", oh.cause());
                    message.fail(500, oh.cause().getMessage());
                  }
                  else {
                    ReadStream<Buffer> stream = streamResult.result();
                    AsyncFile file = oh.result();
                    stream.endHandler(
                        h -> {
                          file.flush().end();
                          message.reply(new JsonObject().put("status", 201));
                        }
                    );
                    Pump.pump(stream, file).start();
                  }
                }
            );
          }
        }
    );

  }
}

@@ -1422,75 +1422,84 @@
     
-contentManager.receive(%0A        contentHandle,%0A        streamResult
+vertx.fileSystem().open(%0A        path,%0A        new OpenOptions(),%0A        oh
  -%3E 
@@ -1514,28 +1514,18 @@
     if (
-streamResult
+oh
 .failed(
@@ -1556,26 +1556,15 @@
 or(%22
-R: %22, streamResult
+E:%22, oh
 .cau
@@ -1600,28 +1600,18 @@
 il(500, 
-streamResult
+oh
 .cause()
@@ -1671,88 +1671,61 @@
     
-vertx.fileSystem().open(%0A                path,%0A                new OpenOptions()
+contentManager.receive(%0A                contentHandle
 ,%0A  
@@ -1730,34 +1730,44 @@
                 
-oh
+streamResult
  -%3E %7B%0A          
@@ -1770,34 +1770,44 @@
             if (
-oh
+streamResult
 .failed()) %7B%0A   
@@ -1826,39 +1826,50 @@
      log.error(%22
-E:%22, oh
+R: %22, streamResult
 .cause());%0A     
@@ -1893,34 +1893,44 @@
 ssage.fail(500, 
-oh
+streamResult
 .cause().getMess
@@ -2294,16 +2294,34 @@
 s%22, 201)
+.put(%22path%22, path)
 );%0A     
@@ -2499,15 +2499,14 @@
 %0A    );%0A
-%0A
   %7D%0A%7D%0A

Fix receiver,