123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086 |
- using Microsoft.Azure.Cosmos;
- using Microsoft.Azure.Cosmos.Linq;
- using System;
- using System.Collections.Concurrent;
- using System.Collections.Generic;
- using System.Diagnostics;
- using System.IO;
- using System.Linq;
- using System.Linq.Expressions;
- using System.Net;
- using System.Reflection;
- using System.Text;
- using System.Text.Json;
- using System.Threading;
- using System.Threading.Tasks;
- using TEAMModelOS.SDK.Context.Attributes.Azure;
- using TEAMModelOS.SDK.Context.Exception;
- using TEAMModelOS.SDK.Helper.Common.CollectionHelper;
- using TEAMModelOS.SDK.Helper.Common.ReflectorExtensions;
- using TEAMModelOS.SDK.Module.AzureCosmosDB.Configuration;
- namespace TEAMModelOS.SDK.Module.AzureCosmosDBV3
- {
- public class CosmosModelInfo
- {
- public Container container { get; set; }
- public bool cache { get; set; }
- public bool monitor { get; set; } = false;
- public Type type { get; set; }
- }
- public class AzureCosmosDBV3Repository : IAzureCosmosDBV3Repository
- {
- private CosmosClient CosmosClient { get; set; }
- /// <summary>
- /// 线程安全的dict类型
- /// </summary>
- private Dictionary<string, CosmosModelInfo> DocumentCollectionDict { get; set; } = new Dictionary<string, CosmosModelInfo>();
- private string DatabaseId { get; set; }
- private int CollectionThroughput { get; set; }
- private Database database { get; set; }
- int pageSize = 200;
- private const string CacheCosmosPrefix = "cosmos:";
- private string[] ScanModel { get; set; }
- private const int timeoutSeconds = 86400;
- private const int ttl = 1;
- private string leaseId = "AleaseContainer";
- public AzureCosmosDBV3Repository(AzureCosmosDBOptions options, CosmosSerializer cosmosSerializer)
- {
- try
- {
- if (!string.IsNullOrEmpty(options.ConnectionString))
- {
- CosmosClient = CosmosDBV3ClientSingleton.getInstance(options.ConnectionString, options.ConnectionKey, cosmosSerializer).GetCosmosDBClient();
- }
- else
- {
- throw new BizException("请设置正确的AzureCosmosDB数据库配置信息!");
- }
- DatabaseId = options.Database;
- CollectionThroughput = options.CollectionThroughput;
- ScanModel = options.ScanModel;
- // InitializeDatabase().GetAwaiter().GetResult();
- }
- catch (CosmosException e)
- {
- // Dispose(true);
- throw new BizException(e.Message, 500, e.StackTrace);
- }
- }
- public AzureCosmosDBV3Repository(AzureCosmosDBOptions options)
- {
- try
- {
- if (!string.IsNullOrEmpty(options.ConnectionString))
- {
- CosmosClient = CosmosDBV3ClientSingleton.getInstance(options.ConnectionString, options.ConnectionKey, null).GetCosmosDBClient();
- }
- else
- {
- throw new BizException("请设置正确的AzureCosmosDB数据库配置信息!");
- }
- DatabaseId = options.Database;
- CollectionThroughput = options.CollectionThroughput;
- ScanModel = options.ScanModel;
- // InitializeDatabase().GetAwaiter().GetResult();
- }
- catch (CosmosException e)
- {
- // Dispose(true);
- throw new BizException(e.Message, 500, e.StackTrace);
- }
- }
- public async Task<Dictionary<string, CosmosModelInfo>> InitializeDatabase()
- {
- try
- {
- database = await CosmosClient.CreateDatabaseIfNotExistsAsync(DatabaseId, CollectionThroughput);
- FeedIterator<ContainerProperties> resultSetIterator = database.GetContainerQueryIterator<ContainerProperties>();
- while (resultSetIterator.HasMoreResults)
- {
- foreach (ContainerProperties container in await resultSetIterator.ReadNextAsync())
- {
- DocumentCollectionDict.TryAdd(container.Id, new CosmosModelInfo { container = database.GetContainer(container.Id), cache = false, monitor = false });
- }
- }
- bool isMonitor = false;
- //获取数据库所有的表
- List<Type> types = ReflectorExtensions.GetAllTypeAsAttribute<CosmosDBAttribute>(ScanModel);
- foreach (Type type in types)
- {
- string PartitionKey = GetPartitionKey(type);
- string CollectionName = "";
- int RU = 0;
- bool cache = false;
- bool monitor = false;
- IEnumerable<CosmosDBAttribute> attributes = type.GetCustomAttributes<CosmosDBAttribute>(true);
- if (!string.IsNullOrEmpty(attributes.First<CosmosDBAttribute>().Name))
- {
- CollectionName = attributes.First<CosmosDBAttribute>().Name;
- }
- else
- {
- CollectionName = type.Name;
- }
- if (attributes.First<CosmosDBAttribute>().Cache)
- {
- cache = attributes.First<CosmosDBAttribute>().Cache;
- }
- if (attributes.First<CosmosDBAttribute>().Monitor)
- {
- monitor = attributes.First<CosmosDBAttribute>().Monitor;
- if (monitor)
- {
- isMonitor = true;
- }
- }
- //else
- //{
- // cache = false;
- //}
- if (attributes.First<CosmosDBAttribute>().RU > 400)
- {
- RU = attributes.First<CosmosDBAttribute>().RU;
- }
- else
- {
- RU = CollectionThroughput;
- }
- //如果表存在于数据则检查RU是否变动,如果不存在则执行创建DocumentCollection
- if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo cosmosModelInfo))
- { //更新RU
- cosmosModelInfo.cache = cache;
- Container container = CosmosClient.GetDatabase(DatabaseId).GetContainer(cosmosModelInfo.container.Id);
- int? throughputResponse = await container.ReadThroughputAsync();
- if (throughputResponse < RU)
- {
- await CosmosClient.GetDatabase(DatabaseId).GetContainer(cosmosModelInfo.container.Id).ReplaceThroughputAsync(RU);
- }
- DocumentCollectionDict[CollectionName] = new CosmosModelInfo { container = container, cache = cache, monitor = monitor, type = type };
- }
- else
- {
- ContainerProperties containerProperties = new ContainerProperties { Id = CollectionName ,DefaultTimeToLive=-1 };
- if (!string.IsNullOrEmpty(PartitionKey))
- {
- containerProperties.PartitionKeyPath = "/" + PartitionKey;
- }
- if (RU > CollectionThroughput)
- {
- CollectionThroughput = RU;
- }
- Container containerWithConsistentIndexing = await database.CreateContainerIfNotExistsAsync(containerProperties, throughput: CollectionThroughput);
- DocumentCollectionDict.TryAdd(CollectionName, new CosmosModelInfo { container = containerWithConsistentIndexing, cache = cache, monitor = monitor, type = type });
- }
- }
- if (isMonitor)
- {
- ContainerProperties leaseProperties = new ContainerProperties { Id = leaseId, PartitionKeyPath = "/id", DefaultTimeToLive = -1 };
- Container leaseContainer = await database.CreateContainerIfNotExistsAsync(leaseProperties, throughput: CollectionThroughput);
- DocumentCollectionDict.TryAdd(leaseId, new CosmosModelInfo { container = leaseContainer, cache = false, monitor = false });
- }
- return DocumentCollectionDict;
- }
- catch (CosmosException e)
- {
- throw new BizException(e.Message, 500, e.StackTrace);
- }
- }
-
- private string GetPartitionKey<T>()
- {
- Type type = typeof(T);
- return GetPartitionKey(type);
- }
- private string GetPartitionKey(Type type)
- {
- PropertyInfo[] properties = type.GetProperties();
- List<PropertyInfo> attrProperties = new List<PropertyInfo>();
- foreach (PropertyInfo property in properties)
- {
- if (property.Name.Equals("PartitionKey"))
- {
- attrProperties.Add(property);
- break;
- }
- object[] attributes = property.GetCustomAttributes(true);
- foreach (object attribute in attributes) //2.通过映射,找到成员属性上关联的特性类实例,
- {
- if (attribute is PartitionKeyAttribute)
- {
- attrProperties.Add(property);
- }
- }
- }
- if (attrProperties.Count <= 0)
- {
- throw new BizException(type.Name + "has no PartitionKey !");
- }
- else
- {
- if (attrProperties.Count == 1)
- {
- return attrProperties[0].Name;
- }
- else { throw new BizException("PartitionKey can only be single!"); }
- }
- }
- private async Task<CosmosModelInfo> InitializeCollection<T>()
- {
- Type type = typeof(T);
- string partitionKey = GetPartitionKey<T>();
- string CollectionName;
- IEnumerable<CosmosDBAttribute> attributes = type.GetCustomAttributes<CosmosDBAttribute>(true);
- if (!string.IsNullOrEmpty(attributes.First<CosmosDBAttribute>().Name))
- {
- CollectionName = attributes.First<CosmosDBAttribute>().Name;
- }
- else
- {
- CollectionName = type.Name;
- }
- return await InitializeCollection(CollectionName, partitionKey);
- }
- private async Task<CosmosModelInfo> InitializeCollection(string CollectionName, string PartitionKey)
- {
- /////内存中已经存在这个表则直接返回
- if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo cosmosModelInfo))
- {
- return cosmosModelInfo;
- }///如果没有则尝试默认创建
- else
- {
- ContainerProperties containerProperties = new ContainerProperties { Id = CollectionName };
- if (!string.IsNullOrEmpty(PartitionKey))
- {
- containerProperties.PartitionKeyPath = "/" + PartitionKey;
- }
- Container containerWithConsistentIndexing = await database.CreateContainerIfNotExistsAsync(containerProperties, throughput: CollectionThroughput);
- CosmosModelInfo cosmosModel = new CosmosModelInfo { container = containerWithConsistentIndexing, cache = false };
- DocumentCollectionDict.TryAdd(CollectionName, cosmosModel);
- return cosmosModel;
- }
- }
- /// <summary>
- /// 按TTL删除
- /// </summary>
- /// <typeparam name="T"></typeparam>
- /// <param name="list"></param>
- /// <returns></returns>
- private async Task<List<T>> DeleteTTL<T>(List<T> list)where T:ID {
- CosmosModelInfo container = await InitializeCollection<T>();
- list.ForEach(x=> { x.ttl = ttl; });
- list= await SaveOrUpdateAll(list);
- if (container.cache && RedisHelper.Instance != null)
- {
- list.ForEach(async x => {
- await RedisHelper.HDelAsync(CacheCosmosPrefix + container.container.Id, x.id);
- });
- }
- return list;
- }
- public async Task<List<IdPk>> DeleteAll<T>(List<IdPk> ids) where T : ID
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- List<IdPk> idPks = new List<IdPk>();
- if (container.monitor)
- {
- List<T> list= await FindByDict<T>(new Dictionary<string, object>() { { "id", ids.Select(x => x.id).ToArray() } });
- list= await DeleteTTL(list);
- return ids;
- }
- else {
- int pages = (int)Math.Ceiling((double)ids.Count / pageSize);
- Stopwatch stopwatch = Stopwatch.StartNew();
- for (int i = 0; i < pages; i++)
- {
- List<IdPk> lists = ids.Skip((i) * pageSize).Take(pageSize).ToList();
- List<Task> tasks = new List<Task>(lists.Count);
- lists.ForEach(item =>
- {
- tasks.Add(container.container.DeleteItemStreamAsync(item.id, new PartitionKey(item.pk))
- .ContinueWith((Task<ResponseMessage> task) =>
- {
- using (ResponseMessage response = task.Result)
- {
- idPks.Add(new IdPk { id = item.id, pk = item.pk.ToString(), StatusCode = response.StatusCode });
- // if (!response.IsSuccessStatusCode)
- // {
- // }
- }
- }
- ));
- });
- await Task.WhenAll(tasks);
- if (container.cache && RedisHelper.Instance != null)
- {
- lists.ForEach(async x => {
- await RedisHelper.HDelAsync(CacheCosmosPrefix + container.container.Id, x.id);
- });
- }
- }
- stopwatch.Stop();
- return idPks;
- }
-
- }
- public async Task<List<IdPk>> DeleteAll<T>(Dictionary<string, object> dict) where T : ID
- {
- if (dict.Keys.Count > 0)
- {
- List<T> list = await FindByDict<T>(dict);
- return await DeleteAll(list);
- }
- else {
- throw new BizException("参数为空", 500);
- }
-
- }
- public async Task<List<IdPk>> DeleteAll<T>(List<T> enyites) where T : ID
- {
- Type type = typeof(T);
- string pk = GetPartitionKey<T>();
- CosmosModelInfo container = await InitializeCollection<T>();
- List<IdPk> idPks = new List<IdPk>();
- if (container.monitor)
- {
- enyites = await DeleteTTL(enyites);
- foreach (T t in enyites) {
- object o = type.GetProperty(pk).GetValue(t, null);
- idPks.Add(new IdPk { id = t.id, pk =o.ToString(), StatusCode = HttpStatusCode .NoContent});
- }
- return idPks;
- }
- else {
- int pages = (int)Math.Ceiling((double)enyites.Count / pageSize);
- Stopwatch stopwatch = Stopwatch.StartNew();
- for (int i = 0; i < pages; i++)
- {
- List<T> lists = enyites.Skip((i) * pageSize).Take(pageSize).ToList();
- List<KeyValuePair<PartitionKey, string>> itemsToInsert = new List<KeyValuePair<PartitionKey, string>>();
- lists.ForEach(x =>
- {
- object o = type.GetProperty(pk).GetValue(x, null);
- KeyValuePair<PartitionKey, string> keyValue = new KeyValuePair<PartitionKey, string>(new PartitionKey(o.ToString()), x.id);
- itemsToInsert.Add(keyValue);
- });
- List<Task> tasks = new List<Task>(lists.Count);
- itemsToInsert.ForEach(item =>
- {
- tasks.Add(container.container.DeleteItemStreamAsync(item.Value, item.Key)
- .ContinueWith((Task<ResponseMessage> task) =>
- {
- using (ResponseMessage response = task.Result)
- {
- idPks.Add(new IdPk { id = item.Value, pk = item.Key.ToString(), StatusCode = response.StatusCode });
- }
- }
- ));
- });
- await Task.WhenAll(tasks); if (container.cache && RedisHelper.Instance != null)
- {
- lists.ForEach(async x => {
- await RedisHelper.HDelAsync(CacheCosmosPrefix + container.container.Id, x.id);
- });
- }
- }
- stopwatch.Stop();
- return idPks;
- }
- }
- public async Task<IdPk> DeleteAsync<T>(IdPk idPk) where T : ID
- {
- return await DeleteAsync<T>(idPk.id, idPk.pk);
- }
- public async Task<IdPk> DeleteAsync<T>(string id, string pk) where T : ID
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- if (container.monitor)
- {
- List<T> list= await FindByDict<T>(new Dictionary<string, object>() { { "id", id } });
- if (list.Count > 0)
- {
- await DeleteTTL<T>(list);
- return new IdPk { id = id, pk = pk, StatusCode = HttpStatusCode.NoContent };
- }
- else {
- throw new BizException("未找到ID匹配的数据,删除失败");
- }
- }
- else {
- ResponseMessage response = await container.container.DeleteItemStreamAsync(id: id, partitionKey: new PartitionKey(pk));
- if (container.cache && RedisHelper.Instance != null)
- {
- await RedisHelper.HDelAsync(CacheCosmosPrefix + container.container.Id, id);
- }
- return new IdPk { id = id, pk = pk, StatusCode = response.StatusCode };
- }
-
- }
- public async Task<IdPk> DeleteAsync<T>(T entity) where T : ID
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- string partitionKey = GetPartitionKey<T>();
- Type type = typeof(T);
- object o = type.GetProperty(partitionKey).GetValue(entity, null);
- if (container.monitor)
- {
- await DeleteTTL<T>(new List<T>() { entity});
- return new IdPk { id = entity.id, pk = o.ToString(), StatusCode = HttpStatusCode.NoContent };
- }
- else {
- ResponseMessage response = await container.container.DeleteItemStreamAsync(id: entity.id, partitionKey: new PartitionKey(o.ToString()));
- if (container.cache && RedisHelper.Instance != null)
- {
- await RedisHelper.HDelAsync(CacheCosmosPrefix + container.container.Id, entity.id);
- }
- return new IdPk { id = entity.id, pk = o.ToString(), StatusCode = response.StatusCode };
- }
- }
- //public async Task<T> DeleteAsync<T>(string id) where T : ID
- //{
- // Container container = await InitializeCollection<T>();
- // ItemResponse<T> response = await container.DeleteItemAsync<T>(id: id, partitionKey: new PartitionKey(GetPartitionKey<T>()));
- // return response.Resource;
- //}
- public async Task<List<T>> FindAll<T>(List<string> propertys = null) where T : ID
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- StringBuilder sql;
- sql = SQLHelperParametric.GetSQLSelect(propertys);
- CosmosDbQuery cosmosDbQuery = new CosmosDbQuery { QueryText = sql.ToString() };
- FeedIterator<T> query = container.container.GetItemQueryIterator<T>(queryDefinition: cosmosDbQuery.CosmosQueryDefinition);
- return await ResultsFromFeedIterator(query);
- }
- private async Task<List<T>> ResultsFromFeedIterator<T>(FeedIterator<T> query, int? maxItemCount = null)
- {
- List<T> results = new List<T>();
- while (query.HasMoreResults)
- {
- foreach (T t in await query.ReadNextAsync())
- {
- results.Add(t);
- if (results.Count == maxItemCount)
- {
- return results;
- }
- }
- }
- return results;
- }
- private async Task<List<T>> ResultsFromFeedIterator<T>(FeedIterator<T> query, Func<List<T>, Task> batchAction, int itemsPerPage)
- {
- List<T> results = new List<T>();
- while (query.HasMoreResults)
- {
- if (results.Count() >= itemsPerPage)
- {
- await batchAction(results);
- results.Clear();
- }
- results.AddRange(await query.ReadNextAsync());
- }
- if (results.Count() > 0)
- {
- await batchAction(results);
- results.Clear();
- }
- return results;
- }
- public async Task<List<dynamic>> FindByDict(string CollectionName, Dictionary<string, object> dict, string partitionKey = null, List<string> propertys = null)
- {
- if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo container))
- {
- //StringBuilder sql = new StringBuilder("select value(c) from c");
- //SQLHelper.GetSQL(dict, ref sql);
- //CosmosDbQuery cosmosDbQuery = new CosmosDbQuery, int itemsPerPage = -1, int?
- //{
- // QueryText = sql.ToString()
- //};
- StringBuilder sql;
- sql = SQLHelperParametric.GetSQLSelect(propertys);
- CosmosDbQuery cosmosDbQuery = SQLHelperParametric.GetSQL(dict, sql);
- QueryRequestOptions queryRequestOptions = GetDefaultQueryRequestOptions(itemsPerPage: GetEffectivePageSize(-1, null));
- FeedIterator<dynamic> query = container.container.GetItemQueryIterator<dynamic>(queryDefinition: cosmosDbQuery.CosmosQueryDefinition, requestOptions: queryRequestOptions);
- return await ResultsFromFeedIterator(query);
- }
- else
- {
- throw new BizException("CollectionName named:" + CollectionName + " dose not exsit in Database!");
- }
- }
- public async Task<List<dynamic>> FindCountByDict(string CollectionName, Dictionary<string, object> dict, string partitionKey = null)
- {
- if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo container))
- {
- dict.Remove("@CURRPAGE");
- dict.Remove("@PAGESIZE");
- dict.Remove("@ASC");
- dict.Remove("@DESC");
- StringBuilder sql = new StringBuilder("select value count(c) from c");
- CosmosDbQuery cosmosDbQuery = SQLHelperParametric.GetSQL(dict, sql);
- QueryRequestOptions queryRequestOptions = GetDefaultQueryRequestOptions(itemsPerPage: GetEffectivePageSize(-1, null));
- FeedIterator<dynamic> query = container.container.GetItemQueryIterator<dynamic>(queryDefinition: cosmosDbQuery.CosmosQueryDefinition, requestOptions: queryRequestOptions);
- return await ResultsFromFeedIterator(query);
- }
- else
- {
- throw new BizException("CollectionName named:" + CollectionName + " dose not exsit in Database!");
- }
- }
- public async Task<List<T>> FindByParams<T>(Dictionary<string, object> dict, string partitionKey = null, List<string> propertys = null) where T : ID
- {
- return await FindByDict<T>(dict, partitionKey, propertys);
- }
- public async Task<List<T>> FindByDict<T>(Dictionary<string, object> dict, string partitionKey = null, List<string> propertys = null) where T : ID
- {
- StringBuilder sql;
- sql = SQLHelperParametric.GetSQLSelect(propertys);
- CosmosDbQuery cosmosDbQuery = SQLHelperParametric.GetSQL(dict, sql);
- QueryRequestOptions queryRequestOptions = GetDefaultQueryRequestOptions(itemsPerPage: GetEffectivePageSize(-1, null));
- return await ResultsFromQueryAndOptions<T>(cosmosDbQuery, queryRequestOptions);
- }
- private async Task<List<T>> ResultsFromQueryAndOptions<T>(CosmosDbQuery cosmosDbQuery, QueryRequestOptions queryOptions)
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- FeedIterator<T> query = container.container.GetItemQueryIterator<T>(
- queryDefinition: cosmosDbQuery.CosmosQueryDefinition,
- requestOptions: queryOptions);
- return await ResultsFromFeedIterator(query);
- }
- private int GetEffectivePageSize(int itemsPerPage, int? maxItemCount)
- {
- return itemsPerPage == -1 ? maxItemCount ?? itemsPerPage : Math.Min(maxItemCount ?? itemsPerPage, itemsPerPage);
- }
- private QueryRequestOptions GetDefaultQueryRequestOptions(int? itemsPerPage = null,
- int? maxBufferedItemCount = null,
- int? maxConcurrency = null)
- {
- QueryRequestOptions queryRequestOptions = new QueryRequestOptions
- {
- MaxItemCount = itemsPerPage == -1 ? 1000 : itemsPerPage,
- MaxBufferedItemCount = maxBufferedItemCount ?? 100,
- MaxConcurrency = maxConcurrency ?? 50
- };
- return queryRequestOptions;
- }
- private async Task<List<T>> ResultsFromQueryAndOptions<T>(CosmosDbQuery cosmosDbQuery, Func<List<T>, Task> batchAction, QueryRequestOptions queryOptions)
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- FeedIterator<T> query = container.container.GetItemQueryIterator<T>(
- queryDefinition: cosmosDbQuery.CosmosQueryDefinition,
- requestOptions: queryOptions);
- return await ResultsFromFeedIterator(query, batchAction, queryOptions.MaxItemCount ?? 0);
- }
- private QueryRequestOptions GetQueryRequestOptions(int itemsPerPage)
- {
- QueryRequestOptions queryRequestOptions = new QueryRequestOptions
- {
- MaxItemCount = itemsPerPage
- };
- return queryRequestOptions;
- }
- public async Task<List<T>> FindLinq<T>(Expression<Func<T, bool>> query = null, Expression<Func<T, object>> order = null, bool isDesc = false) where T : ID
- {
- //QueryRequestOptions queryRequestOptions = GetQueryRequestOptions(itemsPerPage);
- QueryRequestOptions queryRequestOptions = GetDefaultQueryRequestOptions(itemsPerPage: GetEffectivePageSize(-1, null));
- FeedIterator<T> feedIterator;
- CosmosModelInfo container = await InitializeCollection<T>();
- if (query == null)
- {
- if (order != null)
- {
- if (isDesc)
- {
- feedIterator = container.container
- .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions).OrderByDescending(order)
- .ToFeedIterator();
- }
- else
- {
- feedIterator = container.container
- .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions).OrderBy(order)
- .ToFeedIterator();
- }
- }
- else
- {
- feedIterator = container.container
- .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions)
- .ToFeedIterator();
- }
- }
- else
- {
- if (order != null)
- {
- if (isDesc)
- {
- feedIterator = container.container
- .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions)
- .Where(query).OrderByDescending(order)
- .ToFeedIterator();
- }
- else
- {
- feedIterator = container.container
- .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions)
- .Where(query).OrderBy(order)
- .ToFeedIterator();
- }
- }
- else
- {
- feedIterator = container.container
- .GetItemLinqQueryable<T>(requestOptions: queryRequestOptions)
- .Where(query)
- .ToFeedIterator();
- }
- }
- return await ResultsFromFeedIterator<T>(feedIterator);
- }
- public async Task<List<T>> FindSQL<T>(string sql, Dictionary<string, object> Parameters = null) where T : ID
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- QueryRequestOptions queryOptions = GetQueryRequestOptions(GetEffectivePageSize(-1, null));
- if (Parameters != null)
- {
- CosmosDbQuery cosmosDbQuery = new CosmosDbQuery
- {
- QueryText = sql,
- Parameters = Parameters
- };
- FeedIterator<T> feedIterator = container.container
- .GetItemQueryIterator<T>(cosmosDbQuery.CosmosQueryDefinition, requestOptions: queryOptions);
- return await ResultsFromFeedIterator(feedIterator);
- }
- else
- {
- QueryDefinition queryDefinition = new QueryDefinition(sql);
- return await ResultsFromFeedIterator<T>(container.container.GetItemQueryIterator<T>(queryDefinition));
- }
- }
- public async Task<T> Save<T>(T entity) where T : ID
- {
- try
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- ItemResponse<T> response = await container.container.CreateItemAsync<T>(entity);
- if (container.cache && RedisHelper.Instance != null)
- {
- if (!RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
- {
- await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
- }
- else
- {
- await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
- await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
- }
- }
- return response.Resource;
- }
- catch (Exception e)
- {
- throw new BizException(e.Message);
- }
- }
- public async Task<List<T>> SaveAll<T>(List<T> enyites) where T : ID
- {
- int pages = (int)Math.Ceiling((double)enyites.Count / pageSize);
- CosmosModelInfo container = await InitializeCollection<T>();
- bool flag = false;
- if (RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
- {
- flag = true;
- }
- string pk = GetPartitionKey<T>();
- Type type = typeof(T);
- Stopwatch stopwatch = Stopwatch.StartNew();
- for (int i = 0; i < pages; i++)
- {
- List<T> lists = enyites.Skip((i) * pageSize).Take(pageSize).ToList();
- List<KeyValuePair<PartitionKey, Stream>> itemsToInsert = new List<KeyValuePair<PartitionKey, Stream>>();
- lists.ForEach(async x =>
- {
- MemoryStream stream = new MemoryStream();
- await JsonSerializer.SerializeAsync(stream, x, new JsonSerializerOptions { IgnoreNullValues = true });
- object o = type.GetProperty(pk).GetValue(x, null);
- KeyValuePair<PartitionKey, Stream> keyValue = new KeyValuePair<PartitionKey, Stream>(new PartitionKey(o.ToString()), stream);
- itemsToInsert.Add(keyValue);
- });
- List<Task> tasks = new List<Task>(lists.Count);
- itemsToInsert.ForEach(item =>
- {
- tasks.Add(container.container.CreateItemStreamAsync(item.Value, item.Key)
- .ContinueWith((Task<ResponseMessage> task) =>
- {
- using (ResponseMessage response = task.Result)
- {
- if (!response.IsSuccessStatusCode)
- {
- }
- }
- }
- ));
- });
- await Task.WhenAll(tasks);
- if (container.cache && RedisHelper.Instance != null)
- {
- lists.ForEach(async x => {
- await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, x.id, x);
- });
- }
- }
- if (container.cache && RedisHelper.Instance != null && !flag)
- {
- await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
- }
- stopwatch.Stop();
- return enyites;
- }
- public async Task<T> SaveOrUpdate<T>(T entity) where T : ID
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- ItemResponse<T> response = await container.container.UpsertItemAsync(item: entity);
- if (container.cache && RedisHelper.Instance != null)
- {
- if (!RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
- {
- await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
- }
- else
- {
- await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
- await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
- }
- }
- return response.Resource;
- }
- public async Task<List<T>> SaveOrUpdateAll<T>(List<T> enyites) where T : ID
- {
- //await Task.Run(() => Parallel.ForEach(entities, new ParallelOptions { MaxDegreeOfParallelism = 2 }, (item) =>
- //{
- // Task.WaitAll(Update(item));
- //}));
- int pages = (int)Math.Ceiling((double)enyites.Count / pageSize);
- CosmosModelInfo container = await InitializeCollection<T>();
- bool flag = false;
- if (RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
- {
- flag = true;
- }
- string pk = GetPartitionKey<T>();
- Type type = typeof(T);
- Stopwatch stopwatch = Stopwatch.StartNew();
- for (int i = 0; i < pages; i++)
- {
- List<T> lists = enyites.Skip((i) * pageSize).Take(pageSize).ToList();
- List<KeyValuePair<PartitionKey, Stream>> itemsToInsert = new List<KeyValuePair<PartitionKey, Stream>>();
- lists.ForEach(async x =>
- {
- MemoryStream stream = new MemoryStream();
- await JsonSerializer.SerializeAsync(stream, x, new JsonSerializerOptions { IgnoreNullValues = true });
- object o = type.GetProperty(pk).GetValue(x, null);
- KeyValuePair<PartitionKey, Stream> keyValue = new KeyValuePair<PartitionKey, Stream>(new PartitionKey(o.ToString()), stream);
- itemsToInsert.Add(keyValue);
- });
- List<Task> tasks = new List<Task>(lists.Count);
- itemsToInsert.ForEach(item =>
- {
- tasks.Add(container.container.UpsertItemStreamAsync(item.Value, item.Key)
- .ContinueWith((Task<ResponseMessage> task) =>
- {
- //using (ResponseMessage response = task.Result)
- //{
- // if (!response.IsSuccessStatusCode)
- // {
- // }
- //}
- }
- ));
- });
- await Task.WhenAll(tasks);
- if (container.cache && RedisHelper.Instance != null)
- {
- lists.ForEach(async x => {
- await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, x.id, x);
- });
- }
- }
- if (container.cache && RedisHelper.Instance != null && !flag)
- {
- await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
- }
- stopwatch.Stop();
- return enyites;
- }
- public async Task<T> Update<T>(T entity) where T : ID
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- string pk = GetPartitionKey<T>();
- object o = typeof(T).GetProperty(pk).GetValue(entity, null);
- ItemResponse<T> response = await container.container.ReplaceItemAsync(entity, entity.id, new PartitionKey(o.ToString()));
- if (container.cache && RedisHelper.Instance != null)
- {
- if (!RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
- {
- await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
- }
- else
- {
- await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, entity.id, entity);
- await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
- }
- }
- return response.Resource;
- }
- internal class Item
- {
- public string id { get; set; }
- public string pk { get; set; }
- public MemoryStream stream { get; set; }
- }
- public async Task<List<T>> UpdateAll<T>(List<T> enyites) where T : ID
- {
- //await Task.Run(() => Parallel.ForEach(entities, new ParallelOptions { MaxDegreeOfParallelism = 2 }, (item) =>
- //{
- // Task.WaitAll(Update(item));
- //}));
- int pages = (int)Math.Ceiling((double)enyites.Count / pageSize);
- CosmosModelInfo container = await InitializeCollection<T>();
- bool flag = false;
- if (RedisHelper.Exists(CacheCosmosPrefix + container.container.Id))
- {
- flag = true;
- }
- string pk = GetPartitionKey<T>();
- Type type = typeof(T);
- Stopwatch stopwatch = Stopwatch.StartNew();
- for (int i = 0; i < pages; i++)
- {
- List<T> lists = enyites.Skip((i) * pageSize).Take(pageSize).ToList();
- List<Item> itemsToInsert = new List<Item>();
- lists.ForEach(async x =>
- {
- MemoryStream stream = new MemoryStream();
- await JsonSerializer.SerializeAsync(stream, x, new JsonSerializerOptions { IgnoreNullValues = true });
- object o = type.GetProperty(pk).GetValue(x, null);
- Item keyValue = new Item { id = x.id, pk = o.ToString(), stream = stream };
- itemsToInsert.Add(keyValue);
- });
- List<Task> tasks = new List<Task>(lists.Count);
- itemsToInsert.ForEach(item =>
- {
- tasks.Add(container.container.ReplaceItemStreamAsync(item.stream, item.id, new PartitionKey(item.pk))
- .ContinueWith((Task<ResponseMessage> task) =>
- {
- //using (ResponseMessage response = task.Result)
- //{
- // if (!response.IsSuccessStatusCode)
- // {
- // }
- //}
- }
- ));
- });
- await Task.WhenAll(tasks);
- if (container.cache && RedisHelper.Instance != null)
- {
- lists.ForEach(async x => {
- await RedisHelper.HSetAsync(CacheCosmosPrefix + container.container.Id, x.id, x);
- });
- }
- }
- if (container.cache && RedisHelper.Instance != null && !flag)
- {
- await RedisHelper.ExpireAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds);
- }
- stopwatch.Stop();
- return enyites;
- }
- //public void Dispose()
- //{
- // Dispose(true);
- //}
- //protected virtual void Dispose(bool disposing)
- //{
- // if (disposing)
- // {
- // CosmosClient?.Dispose();
- // }
- //}
- private async Task<T> FindByIdAsSql<T>(string id) where T : ID
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- CosmosDbQuery cosmosDbQuery = new CosmosDbQuery
- {
- QueryText = @"SELECT *
- FROM c
- WHERE c.id = @id",
- Parameters = new Dictionary<string, object>
- {
- { "@id",id}
- }
- };
- FeedIterator<T> feedIterator = container.container
- .GetItemQueryIterator<T>(cosmosDbQuery.CosmosQueryDefinition);
- return (await ResultsFromFeedIterator(feedIterator)).SingleOrDefault();
- }
- public async Task<T> FindByIdPk<T>(string id, string pk) where T : ID
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- ItemResponse<T> response = await container.container.ReadItemAsync<T>(id: id, partitionKey: new PartitionKey(pk));
- return response.Resource;
- }
- public async Task<T> FindById<T>(string id, bool cache = true) where T : ID
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- if (container.cache && RedisHelper.Instance != null && cache==true)
- {
- return await RedisHelper.CacheShellAsync(CacheCosmosPrefix + container.container.Id, id, timeoutSeconds, () => { return FindByIdAsSql<T>(id); });
- }
- else
- {
- return await FindByIdAsSql<T>(id);
- }
- }
- public async Task<List<T>> FindByIds<T>(List<string> ids) where T : ID
- {
- CosmosModelInfo container = await InitializeCollection<T>();
- if (container.cache && RedisHelper.Instance != null)
- {
- List<T> list = new List<T>();
- List<string> NotIn = new List<string>();
- foreach (string id in ids)
- {
- if (!await RedisHelper.HExistsAsync(CacheCosmosPrefix + container.container.Id, id))
- {
- NotIn.Add(id);
- }
- else
- {
- list.Add(await RedisHelper.HGetAsync<T>(CacheCosmosPrefix + container.container.Id, id));
- }
- }
- if (NotIn.IsNotEmpty())
- {
- List<T> noInList = await FindByDict<T>(new Dictionary<string, object> { { "id", NotIn.ToArray() } });
- noInList.ForEach(x => { RedisHelper.HSet(CacheCosmosPrefix + container.container.Id, x.id, x); RedisHelper.Expire(CacheCosmosPrefix + container.container.Id, timeoutSeconds); });
- list.AddRange(noInList);
- }
- return list;
- }
- else
- {
- return await FindByDict<T>(new Dictionary<string, object> { { "id", ids.ToArray() } });
- }
- }
- public async Task<dynamic> FindById(string CollectionName, string id)
- {
- if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo container))
- {
- if (container.cache && RedisHelper.Instance != null)
- {
- return await RedisHelper.CacheShellAsync(CacheCosmosPrefix + container.container.Id, id, timeoutSeconds, () => { return FindByDict(CollectionName, new Dictionary<string, object> { { "id", id } }); });
- }
- else
- {
- return await FindByDict(CollectionName, new Dictionary<string, object> { { "id", id } });
- }
- }
- else
- {
- throw new BizException("CollectionName named:" + CollectionName + " dose not exsit in Database!");
- }
- }
- public async Task<List<dynamic>> FindByIds(string CollectionName, List<string> ids)
- {
- if (DocumentCollectionDict.TryGetValue(CollectionName, out CosmosModelInfo container))
- {
- if (container.cache && RedisHelper.Instance != null)
- {
- return await RedisHelper.CacheShellAsync(CacheCosmosPrefix + container.container.Id, timeoutSeconds, () => { return FindByDict(CollectionName, new Dictionary<string, object> { { "id", ids.ToArray() } }); });
- }
- else
- {
- return await FindByDict(CollectionName, new Dictionary<string, object> { { "id", ids.ToArray() } });
- }
- }
- else
- {
- throw new BizException("CollectionName named:" + CollectionName + " dose not exsit in Database!");
- }
- }
- }
- }
|