content: package module;
  
  import io.activej.csp.process.frame.FrameFormat;
  import io.activej.csp.process.frame.FrameFormats;
  import io.activej.dataflow.calcite.inject.CalciteServerModule;
  import io.activej.dataflow.inject.DatasetId;
  import io.activej.datastream.supplier.StreamSupplier;
  import io.activej.datastream.supplier.StreamSupplierWithResult;
  import io.activej.fs.FileSystem;
  import io.activej.fs.IFileSystem;
  import io.activej.inject.annotation.Named;
  import io.activej.inject.annotation.Provides;
  import io.activej.inject.annotation.Transient;
  import io.activej.inject.module.AbstractModule;
  import io.activej.multilog.IMultilog;
  import io.activej.multilog.LogFile;
  import io.activej.multilog.LogNamingScheme;
  import io.activej.multilog.Multilog;
  import io.activej.promise.Promise;
  import io.activej.reactor.Reactor;
  import io.activej.serializer.BinarySerializer;
  import misc.LogItem;
  
  import java.io.IOException;
  import java.nio.file.Files;
  import java.nio.file.Path;
  import java.util.concurrent.Executor;
  import java.util.concurrent.ThreadLocalRandom;
  
  import static module.MultilogDataflowSchemaModule.LOG_ITEM_TABLE_NAME;
  
  public class MultilogDataflowServerModule extends AbstractModule {
  
  	private MultilogDataflowServerModule() {
  	}
  
  	public static MultilogDataflowServerModule create() {
  		return new MultilogDataflowServerModule();
  	}
  
  	@Override
  	protected void configure() {
  		install(CalciteServerModule.create());
  		install(MultilogDataflowSchemaModule.create());
  	}
  
  	@Provides
  	@Transient
  	@DatasetId(LOG_ITEM_TABLE_NAME)
  	Promise<StreamSupplier<LogItem>> logItemDataset(@Named("Dataflow") Reactor reactor, IMultilog<LogItem> logItemMultilog, @Named("partition") String partition) {
  		assert reactor.inReactorThread();
  		return logItemMultilog.read(partition, new LogFile("", 0), 0L, null)
  			.map(StreamSupplierWithResult::getSupplier);
  	}
  
  	@Provides
  	@Named("partition")
  	String generateLogPartition() {
  		int partitionId = ThreadLocalRandom.current().nextInt(Integer.MAX_VALUE);
  		return "partition" + partitionId;
  	}
  
  	@Provides
  	IMultilog<LogItem> multilog(@Named("Dataflow") Reactor reactor, IFileSystem fs, FrameFormat frameFormat, BinarySerializer<LogItem> logItemSerializer, LogNamingScheme namingScheme) {
  		return Multilog.create(reactor, fs, frameFormat, logItemSerializer, namingScheme);
  	}
  
  	@Provides
  	IFileSystem fs(@Named("Dataflow") Reactor reactor, Executor executor) throws IOException {
  		Path multilogPath = Files.createTempDirectory("multilog");
  		return FileSystem.create(reactor, executor, multilogPath);
  	}
  
  	@Provides
  	FrameFormat frameFormat() {
  		return FrameFormats.lz4();
  	}
  
  	@Provides
  	LogNamingScheme logNamingScheme() {
  		return LogNamingScheme.NAME_PARTITION_REMAINDER;
  	}
  }
  