package com.emc.mongoose.load.step.linear;

import com.emc.mongoose.env.Extension;
import com.emc.mongoose.exception.OmgShootMyFootException;
import com.emc.mongoose.data.DataInput;
import com.emc.mongoose.item.io.IoType;
import com.emc.mongoose.item.Item;
import com.emc.mongoose.item.ItemFactory;
import com.emc.mongoose.item.ItemInfoFileOutput;
import com.emc.mongoose.item.ItemType;
import com.emc.mongoose.load.controller.LoadController;
import com.emc.mongoose.load.generator.LoadGenerator;
import com.emc.mongoose.load.step.local.LoadStepLocalBase;
import com.emc.mongoose.storage.driver.StorageDriver;
import com.emc.mongoose.load.controller.LoadControllerImpl;
import com.emc.mongoose.load.generator.LoadGeneratorBuilderImpl;
import com.emc.mongoose.logging.LogUtil;
import com.emc.mongoose.logging.Loggers;

import com.github.akurilov.commons.io.Output;
import com.github.akurilov.commons.reflection.TypeUtil;
import com.github.akurilov.commons.system.SizeInBytes;
import com.github.akurilov.commons.concurrent.throttle.RateThrottle;

import com.github.akurilov.confuse.Config;

import org.apache.logging.log4j.Level;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CancellationException;

public class LinearLoadStepLocal
extends LoadStepLocalBase {

	public LinearLoadStepLocal(
		final Config baseConfig, final List<Extension> extensions, final List<Map<String, Object>> contexts
	) {
		super(baseConfig, extensions, contexts);
	}

	@Override
	public String getTypeName() {
		return LinearLoadStepExtension.TYPE;
	}

	@Override
	protected void init() {

		final String autoStepId = "linear_" + LogUtil.getDateTimeStamp();
		if(config.boolVal("load-step-idAutoGenerated")) {
			config.val("load-step-id", autoStepId);
		}

		final Config loadConfig = config.configVal("load");
		final Config stepConfig = loadConfig.configVal("step");
		final IoType ioType = IoType.valueOf(loadConfig.stringVal("type").toUpperCase());
		final int concurrency = stepConfig.intVal("limit-concurrency");
		final Config outputConfig = config.configVal("output");
		final Config metricsConfig = outputConfig.configVal("metrics");
		final SizeInBytes itemDataSize;
		final Object itemDataSizeRaw = config.val("item-data-size");
		if(itemDataSizeRaw instanceof String) {
			itemDataSize = new SizeInBytes((String) itemDataSizeRaw);
		} else {
			itemDataSize = new SizeInBytes(TypeUtil.typeConvert(itemDataSizeRaw, long.class));
		}
		final int originIndex = 0;
		final boolean outputColorFlag = outputConfig.boolVal("color");
		initMetrics(originIndex, ioType, concurrency, metricsConfig, itemDataSize, outputColorFlag);

		final Config itemConfig = config.configVal("item");
		final Config storageConfig = config.configVal("storage");
		final Config dataConfig = itemConfig.configVal("data");
		final Config dataInputConfig = dataConfig.configVal("input");
		final Config limitConfig = stepConfig.configVal("limit");
		final Config dataLayerConfig = dataInputConfig.configVal("layer");

		final String testStepId = stepConfig.stringVal("id");

		try {

			final Object dataLayerSizeRaw = dataLayerConfig.val("size");
			final SizeInBytes dataLayerSize;
			if(dataLayerSizeRaw instanceof String) {
				dataLayerSize = new SizeInBytes((String) dataLayerSizeRaw);
			} else {
				dataLayerSize = new SizeInBytes(TypeUtil.typeConvert(dataLayerSizeRaw, int.class));
			}

			final DataInput dataInput = DataInput.instance(
				dataInputConfig.stringVal("file"), dataInputConfig.stringVal("seed"), dataLayerSize,
				dataLayerConfig.intVal("cache")
			);

			try {

				final StorageDriver driver = StorageDriver.instance(
					extensions, loadConfig, storageConfig, dataInput, dataConfig.boolVal("verify"), testStepId
				);
				drivers.add(driver);

				final ItemType itemType = ItemType.valueOf(
					itemConfig.stringVal("type").toUpperCase()
				);
				final ItemFactory<Item> itemFactory = ItemType.getItemFactory(itemType);
				final double rateLimit = stepConfig.doubleVal("limit-rate");

				try {
					final LoadGenerator generator = new LoadGeneratorBuilderImpl<>()
						.itemConfig(itemConfig)
						.loadConfig(loadConfig)
						.limitConfig(limitConfig)
						.itemType(itemType)
						.itemFactory((ItemFactory) itemFactory)
						.storageDriver(driver)
						.authConfig(storageConfig.configVal("auth"))
						.originIndex(0)
						.addThrottle(rateLimit > 0 ? new RateThrottle(rateLimit) : null)
						.build();
					generators.add(generator);

					final LoadController controller = new LoadControllerImpl<>(
						testStepId, generator, driver, metricsContexts.get(0), limitConfig,
						outputConfig.boolVal("metrics-trace-persist"),
						loadConfig.intVal("batch-size"),
						loadConfig.intVal("generator-recycle-limit")
					);
					controllers.add(controller);

					final String itemOutputFile = itemConfig.stringVal("output-file");
					if(itemOutputFile != null && itemOutputFile.length() > 0) {
						final Path itemOutputPath = Paths.get(itemOutputFile);
						if(Files.exists(itemOutputPath)) {
							Loggers.ERR.warn("Items output file \"{}\" already exists", itemOutputPath);
						}
						try {
							final Output<? extends Item> itemOutput = new ItemInfoFileOutput<>(itemOutputPath);
							controller.ioResultsOutput(itemOutput);
						} catch(final IOException e) {
							LogUtil.exception(
								Level.ERROR, e,
								"Failed to initialize the item output, the processed items " +
									"info won't be persisted"
							);
						}
					}
				} catch(final OmgShootMyFootException e) {
					throw new IllegalStateException(
						"Failed to initialize the load generator", e
					);
				}
			} catch(final OmgShootMyFootException e) {
				throw new IllegalStateException("Failed to initialize the storage driver", e);
			} catch(final InterruptedException e) {
				throw new CancellationException();
			}
		} catch(final IOException e) {
			throw new IllegalStateException("Failed to initialize the data input", e);
		}
	}
}

@@ -490,16 +490,77 @@
 erator;%0A
+import com.emc.mongoose.load.generator.LoadGeneratorBuilder;%0A
 import c
@@ -4208,16 +4208,23 @@
 enerator
+Builder
  generat
@@ -4225,16 +4225,23 @@
 enerator
+Builder
  = new L
@@ -4537,27 +4537,17 @@
 x(0)
+;
 %0A%09%09%09%09%09
-%09.addThrottle
+if
 (rat
@@ -4560,53 +4560,137 @@
  %3E 0
- ? new RateThrottle(rateLimit) : null)%0A%09%09%09%09%09%09
+) %7B%0A%09%09%09%09%09%09generatorBuilder.addThrottle(new RateThrottle(rateLimit));%0A%09%09%09%09%09%7D%0A%09%09%09%09%09final LoadGenerator generator = generatorBuilder
 .bui

save [skip ci]