package com.gentics.mesh.storage;

import static com.gentics.mesh.core.rest.error.Errors.error;
import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;

import java.io.File;
import java.nio.file.NoSuchFileException;
import java.util.concurrent.atomic.AtomicReference;

import javax.inject.Inject;
import javax.inject.Singleton;

import com.gentics.mesh.Mesh;
import com.gentics.mesh.core.data.node.field.BinaryGraphField;
import com.gentics.mesh.etc.config.MeshUploadOptions;
import com.gentics.mesh.util.RxUtil;

import io.reactivex.Completable;
import io.reactivex.Observable;
import io.reactivex.Single;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.file.OpenOptions;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.reactivex.core.file.AsyncFile;
import io.vertx.reactivex.core.file.FileSystem;
import io.vertx.reactivex.core.streams.Pump;

@Singleton
public class LocalBinaryStorage extends AbstractBinaryStorage {

	private static final Logger log = LoggerFactory.getLogger(LocalBinaryStorage.class);

	@Inject
	public LocalBinaryStorage() {
	}

	@Override
	public Completable store(Observable<Buffer> stream, String uuid) {
		return Completable.defer(() -> {
			FileSystem fileSystem = FileSystem.newInstance(Mesh.vertx().fileSystem());
			String path = getFilePath(uuid);
			log.debug("Saving data for field to path {" + path + "}");
			MeshUploadOptions uploadOptions = Mesh.mesh().getOptions().getUploadOptions();
			File uploadFolder = new File(uploadOptions.getDirectory(), getSegmentedPath(uuid));

			if (!uploadFolder.exists()) {
				if (!uploadFolder.mkdirs()) {
					log.error("Failed to create target folder {" + uploadFolder.getAbsolutePath() + "}");
					throw error(BAD_REQUEST, "node_error_upload_failed");
				}

				if (log.isDebugEnabled()) {
					log.debug("Created folder {" + uploadFolder.getAbsolutePath() + "}");
				}
			}

			File targetFile = new File(uploadFolder, uuid + ".bin");
			return fileSystem.rxOpen(targetFile.getAbsolutePath(), new OpenOptions()).map(file -> {
				Observable<AsyncFile> s = stream.map(buffer -> {
					file.write(new io.vertx.reactivex.core.buffer.Buffer(buffer));
					return buffer;
				}).map(buffer -> file);
				return s.ignoreElements().andThen(Single.just(file)).doOnSuccess(AsyncFile::flush).doFinally(() -> file.close());
			}).flatMap(e -> e).toCompletable();
		});
	}

	/**
	 * Return the absolute path to the binary data for the given uuid.
	 * 
	 * @param binaryUuid
	 * @return
	 */
	public static String getFilePath(String binaryUuid) {
		File folder = new File(Mesh.mesh().getOptions().getUploadOptions().getDirectory(), getSegmentedPath(binaryUuid));
		File binaryFile = new File(folder, binaryUuid + ".bin");
		return binaryFile.getAbsolutePath();
	}

	@Override
	public boolean exists(BinaryGraphField field) {
		String uuid = field.getBinary().getUuid();
		return new File(getFilePath(uuid)).exists();
	}

	@Override
	public Observable<Buffer> read(String binaryUuid) {
		String path = getFilePath(binaryUuid);
		Observable<Buffer> obs = FileSystem.newInstance(Mesh.vertx().fileSystem()).rxOpen(path, new OpenOptions()).toObservable()
				.map(file -> file.getDelegate()).flatMap(RxUtil::toBufferObs);
		return obs;
	}

	/**
	 * Generate the segmented path for the given binary uuid.
	 * 
	 * @param binaryUuid
	 * @return
	 */
	public static String getSegmentedPath(String binaryUuid) {
		String partA = binaryUuid.substring(0, 2);
		String partB = binaryUuid.substring(2, 4);
		StringBuffer buffer = new StringBuffer();
		buffer.append(File.separator);
		buffer.append(partA);
		buffer.append(File.separator);
		buffer.append(partB);
		buffer.append(File.separator);
		return buffer.toString();
	}

	@Override
	public Completable delete(String binaryUuid) {
		String path = getFilePath(binaryUuid);
		return FileSystem.newInstance(Mesh.vertx().fileSystem())

				.rxDelete(path)
				// Don't fail if the file is not even in the local storage
				.onErrorComplete(e -> {
					Throwable cause = e.getCause();
					if (cause != null) {
						return cause instanceof NoSuchFileException;
					} else {
						return e instanceof NoSuchFileException;
					}
				});
	}

}

@@ -2066,345 +2066,230 @@
 ()).
-map(file -%3E %7B%0A%09%09%09%09Observable%3CAsyncFile%3E s = stream.map(buffer -%3E %7B%0A%09%09%09%09%09file.write(new io.vertx.reactivex.core.buffer.Buffer(buffer));%0A%09%09%09%09%09return buffer;%0A%09%09%09%09%7D).map(buffer -%3E fil
+flatMapCompletable(file -%3E %7B%0A%09%09%09%09return stream%0A%09%09%09%09%09.map(io.vertx.reactivex.core.buffer.Buffer::new)%0A%09%09%09%09%09.doOnNext(file::writ
 e)
-;
 %0A%09%09%09%09
-return s.ignoreElements().andThen(Single.just(file)).doOnSuccess(AsyncFile::flush).doFinally(() -%3E 
+%09.doOnComplete(file::flush)%0A%09%09%09%09%09.doOnTerminate(
 file
-.
+::
 close
-());%0A%09%09%09%7D).flatMap(e -%3E e).toCompletable(
+)%0A%09%09%09%09%09.ignoreElements();%0A%09%09%09%7D
 );%0A%09
@@ -3080,40 +3080,8 @@
 %09%09%09%09
-.map(file -%3E file.getDelegate())
 .fla

Rewrite RxJava2 file write code