using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.IO; using System.Linq; using System.Threading.Tasks; using ZeroLevel.Services.FileSystem; using ZeroLevel.Services.PartitionStorage.Interfaces; namespace ZeroLevel.Services.PartitionStorage { public class Store : IStore { private readonly StoreOptions _options; public Store(StoreOptions options) { if (options == null) throw new ArgumentNullException(nameof(options)); _options = options; if (Directory.Exists(_options.RootFolder) == false) { Directory.CreateDirectory(_options.RootFolder); } } public void RemovePartition(TMeta info) { var partition = CreateAccessor(info); partition.DropData(); FSUtils.RemoveFolder(partition.GetCatalogPath()); } public IStorePartitionAccessor CreateAccessor(TMeta info) { return new StorePartitionAccessor(_options, info); } public IStorePartitionBuilder CreateBuilder(TMeta info) { return new StorePartitionBuilder(_options, info); } public IStorePartitionMergeBuilder CreateMergeAccessor(TMeta info, Func> decompressor) { return new StoreMergePartitionAccessor(_options, info, decompressor); } public async Task> Search(StoreSearchRequest searchRequest) { var result = new StoreSearchResult(); var results = new ConcurrentDictionary>>(); if (searchRequest.PartitionSearchRequests?.Any() ?? false) { var partitionsSearchInfo = searchRequest .PartitionSearchRequests .ToDictionary(r => r.Info, r => r.Keys); var options = new ParallelOptions { MaxDegreeOfParallelism = _options.MaxDegreeOfParallelism }; await Parallel.ForEachAsync(partitionsSearchInfo, options, async (pair, _) => { using (var accessor = CreateAccessor(pair.Key)) { results[pair.Key] = accessor .Find(pair.Value) .ToArray(); } }); } result.Results = results; return result; } } }