Skip to content

Commit 017b3e0

Browse files
xuejmnetxuejmnet
committed
#11 支持单个查询的追踪 first firstordefault last lastordefault single singleordefault....
1 parent a66a19d commit 017b3e0

9 files changed

+121
-27
lines changed

‎src/ShardingCore/Sharding/ShardingQueryExecutors/DefaultShardingQueryExecutor.cs‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -79,17 +79,17 @@ public TResult Execute<TResult>(ICurrentDbContext currentContext, Expression que
7979
{
8080

8181
case nameof(Enumerable.First):
82-
return GenericMergeExecute<TResult>(typeof(FirstAsyncInMemoryMergeEngine<>), shardingDbContext, methodCallExpression, async, cancellationToken);
82+
return GenericShardingDbContextMergeExecute<TResult>(typeof(FirstAsyncInMemoryMergeEngine<,>), shardingDbContext, methodCallExpression, async, cancellationToken);
8383
case nameof(Enumerable.FirstOrDefault):
84-
return GenericMergeExecute<TResult>(typeof(FirstOrDefaultAsyncInMemoryMergeEngine<>), shardingDbContext, methodCallExpression, async, cancellationToken);
84+
return GenericShardingDbContextMergeExecute<TResult>(typeof(FirstOrDefaultAsyncInMemoryMergeEngine<,>), shardingDbContext, methodCallExpression, async, cancellationToken);
8585
case nameof(Enumerable.Last):
86-
return GenericMergeExecute<TResult>(typeof(LastAsyncInMemoryMergeEngine<>), shardingDbContext, methodCallExpression, async, cancellationToken);
86+
return GenericShardingDbContextMergeExecute<TResult>(typeof(LastAsyncInMemoryMergeEngine<,>), shardingDbContext, methodCallExpression, async, cancellationToken);
8787
case nameof(Enumerable.LastOrDefault):
88-
return GenericMergeExecute<TResult>(typeof(LastOrDefaultAsyncInMemoryMergeEngine<>), shardingDbContext, methodCallExpression, async, cancellationToken);
88+
return GenericShardingDbContextMergeExecute<TResult>(typeof(LastOrDefaultAsyncInMemoryMergeEngine<,>), shardingDbContext, methodCallExpression, async, cancellationToken);
8989
case nameof(Enumerable.Single):
90-
return GenericMergeExecute<TResult>(typeof(SingleAsyncInMemoryMergeEngine<>), shardingDbContext, methodCallExpression, async, cancellationToken);
90+
return GenericShardingDbContextMergeExecute<TResult>(typeof(SingleAsyncInMemoryMergeEngine<,>), shardingDbContext, methodCallExpression, async, cancellationToken);
9191
case nameof(Enumerable.SingleOrDefault):
92-
return GenericMergeExecute<TResult>(typeof(SingleOrDefaultAsyncInMemoryMergeEngine<>), shardingDbContext, methodCallExpression, async, cancellationToken);
92+
return GenericShardingDbContextMergeExecute<TResult>(typeof(SingleOrDefaultAsyncInMemoryMergeEngine<,>), shardingDbContext, methodCallExpression, async, cancellationToken);
9393
case nameof(Enumerable.Count):
9494
return EnsureMergeExecute<TResult>(typeof(CountAsyncInMemoryMergeEngine<>), shardingDbContext, methodCallExpression, async, cancellationToken);
9595
case nameof(Enumerable.LongCount):
@@ -144,11 +144,11 @@ private TResult EnumerableExecute<TResult>(IShardingDbContext shardingDbContext,
144144
}
145145

146146

147-
private TResult GenericMergeExecute<TResult>(Type streamMergeEngineType, IShardingDbContext shardingDbContext, MethodCallExpression query, bool async, CancellationToken cancellationToken)
147+
private TResult GenericShardingDbContextMergeExecute<TResult>(Type streamMergeEngineType, IShardingDbContext shardingDbContext, MethodCallExpression query, bool async, CancellationToken cancellationToken)
148148
{
149149
var queryEntityType = query.GetQueryEntityType();
150150
var resultEntityType = query.GetResultType();
151-
streamMergeEngineType = streamMergeEngineType.MakeGenericType(queryEntityType);
151+
streamMergeEngineType = streamMergeEngineType.MakeGenericType(shardingDbContext.GetType(),queryEntityType);
152152
var streamEngine = Activator.CreateInstance(streamMergeEngineType, query, shardingDbContext);
153153
var methodName = async ? nameof(IGenericMergeResult.MergeResultAsync) : nameof(IGenericMergeResult.MergeResult);
154154
var streamEngineMethod = streamMergeEngineType.GetMethod(methodName);

‎src/ShardingCore/Sharding/StreamMergeEngines/Abstractions/AbstractInMemoryAsyncMergeEngine.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,7 @@ public AbstractInMemoryAsyncMergeEngine(MethodCallExpression methodCallExpressio
5858

5959

6060
_mergeContext = ((IStreamMergeContextFactory)ShardingContainer.GetService(typeof(IStreamMergeContextFactory<>).GetGenericType0(shardingDbContext.GetType()))).Create(_queryable, shardingDbContext);
61-
_parllelDbbContexts = new List<DbContext>();
61+
_parllelDbbContexts = new LinkedList<DbContext>();
6262
}
6363
/// <summary>
6464
/// 合并queryable

‎src/ShardingCore/Sharding/StreamMergeEngines/Abstractions/AbstractTrackGenericMethodCallWhereInMemoryAsyncMergeEngine.cs‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Linq.Expressions;
4+
using System.Text;
5+
using System.Threading;
6+
using System.Threading.Tasks;
7+
using Microsoft.EntityFrameworkCore;
8+
using ShardingCore.Core.TrackerManagers;
9+
using ShardingCore.Extensions;
10+
using ShardingCore.Sharding.Abstractions;
11+
using ShardingCore.Sharding.StreamMergeEngines.Abstractions.AbstractGenericExpressionMergeEngines;
12+
13+
namespace ShardingCore.Sharding.StreamMergeEngines.Abstractions
14+
{
15+
/*
16+
* @Author: xjm
17+
* @Description:
18+
* @Date: 2021/9/24 10:16:28
19+
* @Ver: 1.0
20+
* @Email: [email protected]
21+
*/
22+
public abstract class AbstractTrackGenericMethodCallWhereInMemoryAsyncMergeEngine<TShardingDbContext,TEntity> : AbstractGenericMethodCallWhereInMemoryAsyncMergeEngine<TEntity> where TShardingDbContext:DbContext,IShardingDbContext
23+
{
24+
private readonly ITrackerManager<TShardingDbContext> _trackerManager;
25+
protected AbstractTrackGenericMethodCallWhereInMemoryAsyncMergeEngine(MethodCallExpression methodCallExpression, IShardingDbContext shardingDbContext) : base(methodCallExpression, shardingDbContext)
26+
{
27+
_trackerManager = ShardingContainer.GetService<ITrackerManager<TShardingDbContext>>();
28+
}
29+
30+
private bool IsUseTrack => GetIsUseTracker();
31+
32+
private bool GetIsUseTracker()
33+
{
34+
if (GetStreamMergeContext().IsNoTracking.HasValue)
35+
{
36+
return !GetStreamMergeContext().IsNoTracking.Value;
37+
}
38+
else
39+
{
40+
return ((DbContext)GetStreamMergeContext().GetShardingDbContext()).ChangeTracker.QueryTrackingBehavior ==
41+
QueryTrackingBehavior.TrackAll;
42+
}
43+
}
44+
public override TResult MergeResult<TResult>()
45+
{
46+
var current = DoMergeResult<TResult>();
47+
if (current != null)
48+
{
49+
if (IsUseTrack && _trackerManager.EntityUseTrack(current.GetType()))
50+
{
51+
var c = (object)current;
52+
var genericDbContext = GetStreamMergeContext().GetShardingDbContext().CreateGenericDbContext(c);
53+
var attachedEntity = genericDbContext.GetAttachedEntity(c);
54+
if (attachedEntity == null)
55+
genericDbContext.Attach(current);
56+
else
57+
{
58+
return (TResult)attachedEntity;
59+
}
60+
}
61+
62+
}
63+
return current;
64+
}
65+
66+
public override async Task<TResult> MergeResultAsync<TResult>(CancellationToken cancellationToken = new CancellationToken())
67+
{
68+
cancellationToken.ThrowIfCancellationRequested();
69+
var current = await DoMergeResultAsync<TResult>(cancellationToken);
70+
if (current != null)
71+
{
72+
if (IsUseTrack && _trackerManager.EntityUseTrack(current.GetType()))
73+
{
74+
var c = (object)current;
75+
var genericDbContext = GetStreamMergeContext().GetShardingDbContext().CreateGenericDbContext(c);
76+
var attachedEntity = genericDbContext.GetAttachedEntity(c);
77+
if (attachedEntity == null)
78+
genericDbContext.Attach(current);
79+
else
80+
{
81+
return (TResult)attachedEntity;
82+
}
83+
}
84+
}
85+
return current;
86+
}
87+
public abstract TResult DoMergeResult<TResult>();
88+
89+
public abstract Task<TResult> DoMergeResultAsync<TResult>(
90+
CancellationToken cancellationToken = new CancellationToken());
91+
92+
93+
}
94+
}

‎src/ShardingCore/Sharding/StreamMergeEngines/FirstAsyncInMemoryMergeEngine.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,18 +21,18 @@ namespace ShardingCore.Sharding.StreamMergeEngines
2121
* @Ver: 1.0
2222
* @Email: [email protected]
2323
*/
24-
public class FirstAsyncInMemoryMergeEngine<TEntity>: AbstractGenericMethodCallWhereInMemoryAsyncMergeEngine<TEntity>
24+
public class FirstAsyncInMemoryMergeEngine<TShardingDbContext,TEntity>: AbstractTrackGenericMethodCallWhereInMemoryAsyncMergeEngine<TShardingDbContext,TEntity> where TShardingDbContext:DbContext,IShardingDbContext
2525
{
2626
public FirstAsyncInMemoryMergeEngine(MethodCallExpression methodCallExpression, IShardingDbContext shardingDbContext) : base(methodCallExpression, shardingDbContext)
2727
{
2828
}
2929

30-
public override TResult MergeResult<TResult>()
30+
public override TResult DoMergeResult<TResult>()
3131
{
3232
return AsyncHelper.RunSync(() => MergeResultAsync<TResult>());
3333
}
3434

35-
public override async Task<TResult> MergeResultAsync<TResult>(CancellationToken cancellationToken = new CancellationToken())
35+
public override async Task<TResult> DoMergeResultAsync<TResult>(CancellationToken cancellationToken = new CancellationToken())
3636
{
3737
var result = await base.ExecuteAsync( queryable => ((IQueryable<TResult>)queryable).FirstAsync(cancellationToken), cancellationToken);
3838
var q = result.Where(o => o != null&&o.QueryResult!=null).Select(o=>o.QueryResult).AsQueryable();

‎src/ShardingCore/Sharding/StreamMergeEngines/FirstOrDefaultAsyncInMemoryMergeEngine.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,18 +24,18 @@ namespace ShardingCore.Sharding.StreamMergeEngines
2424
* @Ver: 1.0
2525
* @Email: [email protected]
2626
*/
27-
public class FirstOrDefaultAsyncInMemoryMergeEngine<TEntity> : AbstractGenericMethodCallWhereInMemoryAsyncMergeEngine<TEntity>
27+
public class FirstOrDefaultAsyncInMemoryMergeEngine<TShardingDbContext,TEntity> : AbstractTrackGenericMethodCallWhereInMemoryAsyncMergeEngine<TShardingDbContext, TEntity> where TShardingDbContext : DbContext, IShardingDbContext
2828
{
2929
public FirstOrDefaultAsyncInMemoryMergeEngine(MethodCallExpression methodCallExpression, IShardingDbContext shardingDbContext) : base(methodCallExpression, shardingDbContext)
3030
{
3131
}
3232

33-
public override TResult MergeResult<TResult>()
33+
public override TResult DoMergeResult<TResult>()
3434
{
3535
return AsyncHelper.RunSync(() => MergeResultAsync<TResult>());
3636
}
3737

38-
public override async Task<TResult> MergeResultAsync<TResult>(CancellationToken cancellationToken = new CancellationToken())
38+
public override async Task<TResult> DoMergeResultAsync<TResult>(CancellationToken cancellationToken = new CancellationToken())
3939
{
4040
var result = await base.ExecuteAsync( queryable => ((IQueryable<TResult>)queryable).FirstOrDefaultAsync(cancellationToken), cancellationToken);
4141
var q = result.Where(o => o != null&&o.QueryResult!=null).Select(o=>o.QueryResult).AsQueryable();

‎src/ShardingCore/Sharding/StreamMergeEngines/LastAsyncInMemoryMergeEngine.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,18 +21,18 @@ namespace ShardingCore.Sharding.StreamMergeEngines
2121
* @Ver: 1.0
2222
* @Email: [email protected]
2323
*/
24-
public class LastAsyncInMemoryMergeEngine<TEntity>: AbstractGenericMethodCallWhereInMemoryAsyncMergeEngine<TEntity>
24+
public class LastAsyncInMemoryMergeEngine<TShardingDbContext,TEntity> : AbstractTrackGenericMethodCallWhereInMemoryAsyncMergeEngine<TShardingDbContext, TEntity> where TShardingDbContext : DbContext, IShardingDbContext
2525
{
2626
public LastAsyncInMemoryMergeEngine(MethodCallExpression methodCallExpression, IShardingDbContext shardingDbContext) : base(methodCallExpression, shardingDbContext)
2727
{
2828
}
2929

30-
public override TResult MergeResult<TResult>()
30+
public override TResult DoMergeResult<TResult>()
3131
{
3232
return AsyncHelper.RunSync(() => MergeResultAsync<TResult>());
3333
}
3434

35-
public override async Task<TResult> MergeResultAsync<TResult>(CancellationToken cancellationToken = new CancellationToken())
35+
public override async Task<TResult> DoMergeResultAsync<TResult>(CancellationToken cancellationToken = new CancellationToken())
3636
{
3737

3838
var result = await base.ExecuteAsync( queryable => ((IQueryable<TResult>)queryable).LastAsync(cancellationToken), cancellationToken);

‎src/ShardingCore/Sharding/StreamMergeEngines/LastOrDefaultAsyncInMemoryMergeEngine.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,18 +21,18 @@ namespace ShardingCore.Sharding.StreamMergeEngines
2121
* @Ver: 1.0
2222
* @Email: [email protected]
2323
*/
24-
public class LastOrDefaultAsyncInMemoryMergeEngine<TEntity>: AbstractGenericMethodCallWhereInMemoryAsyncMergeEngine<TEntity>
24+
public class LastOrDefaultAsyncInMemoryMergeEngine<TShardingDbContext,TEntity> : AbstractTrackGenericMethodCallWhereInMemoryAsyncMergeEngine<TShardingDbContext, TEntity> where TShardingDbContext : DbContext, IShardingDbContext
2525
{
2626
public LastOrDefaultAsyncInMemoryMergeEngine(MethodCallExpression methodCallExpression, IShardingDbContext shardingDbContext) : base(methodCallExpression, shardingDbContext)
2727
{
2828
}
2929

30-
public override TResult MergeResult<TResult>()
30+
public override TResult DoMergeResult<TResult>()
3131
{
3232
return AsyncHelper.RunSync(() => MergeResultAsync<TResult>());
3333
}
3434

35-
public override async Task<TResult> MergeResultAsync<TResult>(CancellationToken cancellationToken = new CancellationToken())
35+
public override async Task<TResult> DoMergeResultAsync<TResult>(CancellationToken cancellationToken = new CancellationToken())
3636
{
3737
var result = await base.ExecuteAsync( queryable => ((IQueryable<TResult>)queryable).LastOrDefaultAsync(cancellationToken), cancellationToken);
3838
var q = result.Where(o => o != null&&o.QueryResult!=null).Select(o=>o.QueryResult).AsQueryable();

‎src/ShardingCore/Sharding/StreamMergeEngines/SingleAsyncInMemoryMergeEngine.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,18 +21,18 @@ namespace ShardingCore.Sharding.StreamMergeEngines
2121
* @Ver: 1.0
2222
* @Email: [email protected]
2323
*/
24-
public class SingleAsyncInMemoryMergeEngine<TEntity>: AbstractGenericMethodCallWhereInMemoryAsyncMergeEngine<TEntity>
24+
public class SingleAsyncInMemoryMergeEngine<TShardingDbContext,TEntity> : AbstractTrackGenericMethodCallWhereInMemoryAsyncMergeEngine<TShardingDbContext, TEntity> where TShardingDbContext : DbContext, IShardingDbContext
2525
{
2626
public SingleAsyncInMemoryMergeEngine(MethodCallExpression methodCallExpression, IShardingDbContext shardingDbContext) : base(methodCallExpression, shardingDbContext)
2727
{
2828
}
2929

30-
public override TResult MergeResult<TResult>()
30+
public override TResult DoMergeResult<TResult>()
3131
{
3232
return AsyncHelper.RunSync(() => MergeResultAsync<TResult>());
3333
}
3434

35-
public override async Task<TResult> MergeResultAsync<TResult>(CancellationToken cancellationToken = new CancellationToken())
35+
public override async Task<TResult> DoMergeResultAsync<TResult>(CancellationToken cancellationToken = new CancellationToken())
3636
{
3737
var result = await base.ExecuteAsync( queryable => ((IQueryable<TResult>)queryable).SingleAsync(cancellationToken), cancellationToken);
3838
var q = result.Where(o => o != null&&o.QueryResult!=null).Select(o=>o.QueryResult).AsQueryable();

‎src/ShardingCore/Sharding/StreamMergeEngines/SingleOrDefaultAsyncInMemoryMergeEngine.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,19 +21,19 @@ namespace ShardingCore.Sharding.StreamMergeEngines
2121
* @Ver: 1.0
2222
* @Email: [email protected]
2323
*/
24-
public class SingleOrDefaultAsyncInMemoryMergeEngine<TEntity>: AbstractGenericMethodCallWhereInMemoryAsyncMergeEngine<TEntity>
24+
public class SingleOrDefaultAsyncInMemoryMergeEngine<TShardingDbContext,TEntity> : AbstractTrackGenericMethodCallWhereInMemoryAsyncMergeEngine<TShardingDbContext, TEntity> where TShardingDbContext : DbContext, IShardingDbContext
2525
{
2626
public SingleOrDefaultAsyncInMemoryMergeEngine(MethodCallExpression methodCallExpression, IShardingDbContext shardingDbContext) : base(methodCallExpression, shardingDbContext)
2727
{
2828
}
2929

30-
public override TResult MergeResult<TResult>()
30+
public override TResult DoMergeResult<TResult>()
3131
{
3232

3333
return AsyncHelper.RunSync(() => MergeResultAsync<TResult>());
3434
}
3535

36-
public override async Task<TResult> MergeResultAsync<TResult>(CancellationToken cancellationToken = new CancellationToken())
36+
public override async Task<TResult> DoMergeResultAsync<TResult>(CancellationToken cancellationToken = new CancellationToken())
3737
{
3838
var result = await base.ExecuteAsync( queryable => ((IQueryable<TResult>)queryable).SingleOrDefaultAsync(cancellationToken), cancellationToken);
3939
var q = result.Where(o => o != null&&o.QueryResult!=null).Select(o=>o.QueryResult).AsQueryable();

0 commit comments

Comments
 (0)