DBComponentSystem.cs 7.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248
  1. using System;
  2. using System.Collections.Generic;
  3. using System.Linq.Expressions;
  4. using MongoDB.Driver;
  5. namespace ET
  6. {
  7. public class DBComponentAwakeSystem : AwakeSystem<DBComponent, string, string, int>
  8. {
  9. public override void Awake(DBComponent self, string dbConnection, string dbName, int zone)
  10. {
  11. self.mongoClient = new MongoClient(dbConnection);
  12. self.database = self.mongoClient.GetDatabase(dbName);
  13. }
  14. }
  15. public class DBComponentDestroySystem: DestroySystem<DBComponent>
  16. {
  17. public override void Destroy(DBComponent self)
  18. {
  19. }
  20. }
  21. public static class DBComponentSystem
  22. {
  23. #region Query
  24. public static async ETTask<T> Query<T>(this DBComponent self, long id, string collection = null) where T : Entity
  25. {
  26. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, id % DBComponent.TaskCount))
  27. {
  28. IAsyncCursor<T> cursor = await self.GetCollection<T>(collection).FindAsync(d => d.Id == id);
  29. return await cursor.FirstOrDefaultAsync();
  30. }
  31. }
  32. public static async ETTask<List<T>> Query<T>(this DBComponent self, Expression<Func<T, bool>> filter, string collection = null)
  33. where T : Entity
  34. {
  35. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, RandomHelper.RandInt64() % DBComponent.TaskCount))
  36. {
  37. IAsyncCursor<T> cursor = await self.GetCollection<T>(collection).FindAsync(filter);
  38. return await cursor.ToListAsync();
  39. }
  40. }
  41. public static async ETTask<List<T>> Query<T>(this DBComponent self, long taskId, Expression<Func<T, bool>> filter, string collection = null)
  42. where T : Entity
  43. {
  44. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, taskId % DBComponent.TaskCount))
  45. {
  46. IAsyncCursor<T> cursor = await self.GetCollection<T>(collection).FindAsync(filter);
  47. return await cursor.ToListAsync();
  48. }
  49. }
  50. public static async ETTask Query(this DBComponent self, long id, List<string> collectionNames, List<Entity> result)
  51. {
  52. if (collectionNames == null || collectionNames.Count == 0)
  53. {
  54. return;
  55. }
  56. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, id % DBComponent.TaskCount))
  57. {
  58. foreach (string collectionName in collectionNames)
  59. {
  60. IAsyncCursor<Entity> cursor = await self.GetCollection(collectionName).FindAsync(d => d.Id == id);
  61. Entity e = await cursor.FirstOrDefaultAsync();
  62. if (e == null)
  63. {
  64. continue;
  65. }
  66. result.Add(e);
  67. }
  68. }
  69. }
  70. public static async ETTask<List<T>> QueryJson<T>(this DBComponent self, string json, string collection = null) where T : Entity
  71. {
  72. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, RandomHelper.RandInt64() % DBComponent.TaskCount))
  73. {
  74. FilterDefinition<T> filterDefinition = new JsonFilterDefinition<T>(json);
  75. IAsyncCursor<T> cursor = await self.GetCollection<T>(collection).FindAsync(filterDefinition);
  76. return await cursor.ToListAsync();
  77. }
  78. }
  79. public static async ETTask<List<T>> QueryJson<T>(this DBComponent self, long taskId, string json, string collection = null) where T : Entity
  80. {
  81. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, RandomHelper.RandInt64() % DBComponent.TaskCount))
  82. {
  83. FilterDefinition<T> filterDefinition = new JsonFilterDefinition<T>(json);
  84. IAsyncCursor<T> cursor = await self.GetCollection<T>(collection).FindAsync(filterDefinition);
  85. return await cursor.ToListAsync();
  86. }
  87. }
  88. #endregion
  89. #region Insert
  90. public static async ETTask InsertBatch<T>(this DBComponent self, IEnumerable<T> list, string collection = null) where T: Entity
  91. {
  92. if (collection == null)
  93. {
  94. collection = typeof (T).Name;
  95. }
  96. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, RandomHelper.RandInt64() % DBComponent.TaskCount))
  97. {
  98. await self.GetCollection(collection).InsertManyAsync(list);
  99. }
  100. }
  101. #endregion
  102. #region Save
  103. public static async ETTask Save<T>(this DBComponent self, T entity, string collection = null) where T : Entity
  104. {
  105. if (entity == null)
  106. {
  107. Log.Error($"save entity is null: {typeof (T).Name}");
  108. return;
  109. }
  110. if (collection == null)
  111. {
  112. collection = entity.GetType().Name;
  113. }
  114. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, entity.Id % DBComponent.TaskCount))
  115. {
  116. await self.GetCollection(collection).ReplaceOneAsync(d => d.Id == entity.Id, entity, new ReplaceOptions { IsUpsert = true });
  117. }
  118. }
  119. public static async ETTask Save<T>(this DBComponent self, long taskId, T entity, string collection = null) where T : Entity
  120. {
  121. if (entity == null)
  122. {
  123. Log.Error($"save entity is null: {typeof (T).Name}");
  124. return;
  125. }
  126. if (collection == null)
  127. {
  128. collection = entity.GetType().Name;
  129. }
  130. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, taskId % DBComponent.TaskCount))
  131. {
  132. await self.GetCollection(collection).ReplaceOneAsync(d => d.Id == entity.Id, entity, new ReplaceOptions { IsUpsert = true });
  133. }
  134. }
  135. public static async ETTask Save(this DBComponent self, long id, List<Entity> entities)
  136. {
  137. if (entities == null)
  138. {
  139. Log.Error($"save entity is null");
  140. return;
  141. }
  142. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, id % DBComponent.TaskCount))
  143. {
  144. foreach (Entity entity in entities)
  145. {
  146. if (entity == null)
  147. {
  148. continue;
  149. }
  150. await self.GetCollection(entity.GetType().Name)
  151. .ReplaceOneAsync(d => d.Id == entity.Id, entity, new ReplaceOptions { IsUpsert = true });
  152. }
  153. }
  154. }
  155. public static async ETTask SaveNotWait<T>(this DBComponent self, T entity, long taskId = 0, string collection = null) where T : Entity
  156. {
  157. if (taskId == 0)
  158. {
  159. await self.Save(entity, collection);
  160. return;
  161. }
  162. await self.Save(taskId, entity, collection);
  163. }
  164. #endregion
  165. #region Remove
  166. public static async ETTask<long> Remove<T>(this DBComponent self, long id, string collection = null) where T : Entity
  167. {
  168. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, id % DBComponent.TaskCount))
  169. {
  170. DeleteResult result = await self.GetCollection<T>(collection).DeleteOneAsync(d => d.Id == id);
  171. return result.DeletedCount;
  172. }
  173. }
  174. public static async ETTask<long> Remove<T>(this DBComponent self, long taskId, long id, string collection = null) where T : Entity
  175. {
  176. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, taskId % DBComponent.TaskCount))
  177. {
  178. DeleteResult result = await self.GetCollection<T>(collection).DeleteOneAsync(d => d.Id == id);
  179. return result.DeletedCount;
  180. }
  181. }
  182. public static async ETTask<long> Remove<T>(this DBComponent self, Expression<Func<T, bool>> filter, string collection = null) where T : Entity
  183. {
  184. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, RandomHelper.RandInt64() % DBComponent.TaskCount))
  185. {
  186. DeleteResult result = await self.GetCollection<T>(collection).DeleteManyAsync(filter);
  187. return result.DeletedCount;
  188. }
  189. }
  190. public static async ETTask<long> Remove<T>(this DBComponent self, long taskId, Expression<Func<T, bool>> filter, string collection = null)
  191. where T : Entity
  192. {
  193. using (await CoroutineLockComponent.Instance.Wait(CoroutineLockType.DB, taskId % DBComponent.TaskCount))
  194. {
  195. DeleteResult result = await self.GetCollection<T>(collection).DeleteManyAsync(filter);
  196. return result.DeletedCount;
  197. }
  198. }
  199. #endregion
  200. }
  201. }