Skip to content

丁松杰 / Pole

Go to a project
Commit 1fffdf16 by dingsongjie
Options

完善逻辑与测试

parent e519b36d
Inline Side-by-side
Showing with 123 additions and 82 deletions
samples/apis/Backet.Api/Grains/BacketGrain.cs
...@@ -5,6 +5,7 @@ using Pole.Core.Grains; ...@@ -5,6 +5,7 @@ using Pole.Core.Grains;
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace Backet.Api.Grains namespace Backet.Api.Grains
......
samples/apis/SagasTest.Api/Activities/Transaction1CompensateErrorActivity.cs
...@@ -4,6 +4,7 @@ using System; ...@@ -4,6 +4,7 @@ using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Net.Http; using System.Net.Http;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace SagasTest.Api.Activities namespace SagasTest.Api.Activities
...@@ -15,12 +16,12 @@ namespace SagasTest.Api.Activities ...@@ -15,12 +16,12 @@ namespace SagasTest.Api.Activities
{ {
this.httpClientFactory = httpClientFactory; this.httpClientFactory = httpClientFactory;
} }
public Task Compensate(Transaction1Dto data) public Task Compensate(Transaction1Dto data,CancellationToken cancellationToken)
{ {
throw new NotImplementedException(); throw new NotImplementedException();
} }
public async Task<ActivityExecuteResult> Execute(Transaction1Dto data) public async Task<ActivityExecuteResult> Execute(Transaction1Dto data, CancellationToken cancellationToken)
{ {
var httpclient = httpClientFactory.CreateClient(); var httpclient = httpClientFactory.CreateClient();
httpclient.BaseAddress = new Uri("http://localhost:5000"); httpclient.BaseAddress = new Uri("http://localhost:5000");
......
samples/apis/SagasTest.Api/Activities/Transaction1ExceptionActivity.cs
...@@ -4,6 +4,7 @@ using System; ...@@ -4,6 +4,7 @@ using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Net.Http; using System.Net.Http;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace SagasTest.Api.Activities namespace SagasTest.Api.Activities
...@@ -16,14 +17,14 @@ namespace SagasTest.Api.Activities ...@@ -16,14 +17,14 @@ namespace SagasTest.Api.Activities
this.httpClientFactory = httpClientFactory; this.httpClientFactory = httpClientFactory;
} }
public async Task Compensate(Transaction1Dto data) public async Task Compensate(Transaction1Dto data, CancellationToken cancellationToken)
{ {
var httpclient = httpClientFactory.CreateClient(); var httpclient = httpClientFactory.CreateClient();
httpclient.BaseAddress = new Uri("http://localhost:5000"); httpclient.BaseAddress = new Uri("http://localhost:5000");
var result = await httpclient.GetAsync("api/OutGoingMock/Transaction1RollBack"); var result = await httpclient.GetAsync("api/OutGoingMock/Transaction1RollBack");
} }
public Task<ActivityExecuteResult> Execute(Transaction1Dto data) public Task<ActivityExecuteResult> Execute(Transaction1Dto data, CancellationToken cancellationToken)
{ {
throw new NotImplementedException(); throw new NotImplementedException();
} }
......
samples/apis/SagasTest.Api/Activities/Transaction1OkActivity.cs
...@@ -4,6 +4,7 @@ using System; ...@@ -4,6 +4,7 @@ using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Net.Http; using System.Net.Http;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace SagasTest.Api.Activities namespace SagasTest.Api.Activities
...@@ -15,18 +16,18 @@ namespace SagasTest.Api.Activities ...@@ -15,18 +16,18 @@ namespace SagasTest.Api.Activities
{ {
this.httpClientFactory = httpClientFactory; this.httpClientFactory = httpClientFactory;
} }
public async Task Compensate(Transaction1Dto data) public async Task Compensate(Transaction1Dto data, CancellationToken cancellationToken)
{ {
var httpclient = httpClientFactory.CreateClient(); var httpclient = httpClientFactory.CreateClient();
httpclient.BaseAddress = new Uri("http://localhost:5000"); httpclient.BaseAddress = new Uri("http://localhost:5000");
var result = await httpclient.GetAsync("api/OutGoingMock/Transaction1RollBack"); var result = await httpclient.GetAsync("api/OutGoingMock/Transaction1RollBack");
} }
public async Task<ActivityExecuteResult> Execute(Transaction1Dto data) public async Task<ActivityExecuteResult> Execute(Transaction1Dto data, CancellationToken cancellationToken)
{ {
var httpclient = httpClientFactory.CreateClient(); var httpclient = httpClientFactory.CreateClient();
httpclient.BaseAddress = new Uri("http://localhost:5000"); httpclient.BaseAddress = new Uri("http://localhost:5000");
var result = await httpclient.GetAsync("api/OutGoingMock/Transaction1Ok"); var result = await httpclient.GetAsync("api/OutGoingMock/Transaction1Ok", cancellationToken);
return ActivityExecuteResult.Success; return ActivityExecuteResult.Success;
} }
} }
......
samples/apis/SagasTest.Api/Activities/Transaction1ReturnFalseActivity.cs
...@@ -4,6 +4,7 @@ using System; ...@@ -4,6 +4,7 @@ using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Net.Http; using System.Net.Http;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace SagasTest.Api.Activities namespace SagasTest.Api.Activities
...@@ -16,14 +17,14 @@ namespace SagasTest.Api.Activities ...@@ -16,14 +17,14 @@ namespace SagasTest.Api.Activities
this.httpClientFactory = httpClientFactory; this.httpClientFactory = httpClientFactory;
} }
public async Task Compensate(Transaction1Dto data) public async Task Compensate(Transaction1Dto data, CancellationToken cancellationToken)
{ {
var httpclient = httpClientFactory.CreateClient(); var httpclient = httpClientFactory.CreateClient();
httpclient.BaseAddress = new Uri("http://localhost:5000"); httpclient.BaseAddress = new Uri("http://localhost:5000");
var result = await httpclient.GetAsync("api/OutGoingMock/Transaction1RollBack"); var result = await httpclient.GetAsync("api/OutGoingMock/Transaction1RollBack");
} }
public async Task<ActivityExecuteResult> Execute(Transaction1Dto data) public async Task<ActivityExecuteResult> Execute(Transaction1Dto data, CancellationToken cancellationToken)
{ {
var httpclient = httpClientFactory.CreateClient(); var httpclient = httpClientFactory.CreateClient();
httpclient.BaseAddress = new Uri("http://localhost:5000"); httpclient.BaseAddress = new Uri("http://localhost:5000");
......
samples/apis/SagasTest.Api/Activities/Transaction2CompensateErrorActivity.cs
...@@ -4,6 +4,7 @@ using System; ...@@ -4,6 +4,7 @@ using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Net.Http; using System.Net.Http;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace SagasTest.Api.Activities namespace SagasTest.Api.Activities
...@@ -15,12 +16,12 @@ namespace SagasTest.Api.Activities ...@@ -15,12 +16,12 @@ namespace SagasTest.Api.Activities
{ {
this.httpClientFactory = httpClientFactory; this.httpClientFactory = httpClientFactory;
} }
public Task Compensate(Transaction2Dto data) public Task Compensate(Transaction2Dto data, CancellationToken cancellationToken)
{ {
throw new NotImplementedException(); throw new NotImplementedException();
} }
public async Task<ActivityExecuteResult> Execute(Transaction2Dto data) public async Task<ActivityExecuteResult> Execute(Transaction2Dto data, CancellationToken cancellationToken)
{ {
var httpclient = httpClientFactory.CreateClient(); var httpclient = httpClientFactory.CreateClient();
httpclient.BaseAddress = new Uri("http://localhost:5000"); httpclient.BaseAddress = new Uri("http://localhost:5000");
......
samples/apis/SagasTest.Api/Activities/Transaction2ExceptionActivity.cs
...@@ -4,6 +4,7 @@ using System; ...@@ -4,6 +4,7 @@ using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Net.Http; using System.Net.Http;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace SagasTest.Api.Activities namespace SagasTest.Api.Activities
...@@ -16,14 +17,14 @@ namespace SagasTest.Api.Activities ...@@ -16,14 +17,14 @@ namespace SagasTest.Api.Activities
this.httpClientFactory = httpClientFactory; this.httpClientFactory = httpClientFactory;
} }
public async Task Compensate(Transaction2Dto data) public async Task Compensate(Transaction2Dto data, CancellationToken cancellationToken)
{ {
var httpclient = httpClientFactory.CreateClient(); var httpclient = httpClientFactory.CreateClient();
httpclient.BaseAddress = new Uri("http://localhost:5000"); httpclient.BaseAddress = new Uri("http://localhost:5000");
var result = await httpclient.GetAsync("api/OutGoingMock/Transaction2RollBack"); var result = await httpclient.GetAsync("api/OutGoingMock/Transaction2RollBack");
} }
public Task<ActivityExecuteResult> Execute(Transaction2Dto data) public Task<ActivityExecuteResult> Execute(Transaction2Dto data, CancellationToken cancellationToken)
{ {
throw new NotImplementedException(); throw new NotImplementedException();
} }
......
samples/apis/SagasTest.Api/Activities/Transaction2OkActivity.cs
...@@ -4,6 +4,7 @@ using System; ...@@ -4,6 +4,7 @@ using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Net.Http; using System.Net.Http;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace SagasTest.Api.Activities namespace SagasTest.Api.Activities
...@@ -15,14 +16,14 @@ namespace SagasTest.Api.Activities ...@@ -15,14 +16,14 @@ namespace SagasTest.Api.Activities
{ {
this.httpClientFactory = httpClientFactory; this.httpClientFactory = httpClientFactory;
} }
public async Task Compensate(Transaction2Dto data) public async Task Compensate(Transaction2Dto data, CancellationToken cancellationToken)
{ {
var httpclient = httpClientFactory.CreateClient(); var httpclient = httpClientFactory.CreateClient();
httpclient.BaseAddress = new Uri("http://localhost:5000"); httpclient.BaseAddress = new Uri("http://localhost:5000");
var result = await httpclient.GetAsync("api/OutGoingMock/Transaction2RollBack"); var result = await httpclient.GetAsync("api/OutGoingMock/Transaction2RollBack");
} }
public async Task<ActivityExecuteResult> Execute(Transaction2Dto data) public async Task<ActivityExecuteResult> Execute(Transaction2Dto data, CancellationToken cancellationToken)
{ {
var httpclient = httpClientFactory.CreateClient(); var httpclient = httpClientFactory.CreateClient();
httpclient.BaseAddress = new Uri("http://localhost:5000"); httpclient.BaseAddress = new Uri("http://localhost:5000");
......
samples/apis/SagasTest.Api/Activities/Transaction2ReturnFalseActivity.cs
...@@ -4,6 +4,7 @@ using System; ...@@ -4,6 +4,7 @@ using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Net.Http; using System.Net.Http;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace SagasTest.Api.Activities namespace SagasTest.Api.Activities
...@@ -16,14 +17,14 @@ namespace SagasTest.Api.Activities ...@@ -16,14 +17,14 @@ namespace SagasTest.Api.Activities
this.httpClientFactory = httpClientFactory; this.httpClientFactory = httpClientFactory;
} }
public async Task Compensate(Transaction2Dto data) public async Task Compensate(Transaction2Dto data, CancellationToken cancellationToken)
{ {
var httpclient = httpClientFactory.CreateClient(); var httpclient = httpClientFactory.CreateClient();
httpclient.BaseAddress = new Uri("http://localhost:5000"); httpclient.BaseAddress = new Uri("http://localhost:5000");
var result = await httpclient.GetAsync("api/OutGoingMock/Transaction2RollBack"); var result = await httpclient.GetAsync("api/OutGoingMock/Transaction2RollBack");
} }
public async Task<ActivityExecuteResult> Execute(Transaction2Dto data) public async Task<ActivityExecuteResult> Execute(Transaction2Dto data, CancellationToken cancellationToken)
{ {
var httpclient = httpClientFactory.CreateClient(); var httpclient = httpClientFactory.CreateClient();
httpclient.BaseAddress = new Uri("http://localhost:5000"); httpclient.BaseAddress = new Uri("http://localhost:5000");
......
samples/apis/SagasTest.Api/Activities/Transaction3ExceptionActivity.cs
...@@ -3,19 +3,20 @@ using Pole.Sagas.Core.Abstraction; ...@@ -3,19 +3,20 @@ using Pole.Sagas.Core.Abstraction;
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace SagasTest.Api.Activities namespace SagasTest.Api.Activities
{ {
public class Transaction3ExceptionActivity : IActivity<Transaction3Dto> public class Transaction3ExceptionActivity : IActivity<Transaction3Dto>
{ {
public Task Compensate(Transaction3Dto data) public Task Compensate(Transaction3Dto data, CancellationToken cancellationToken)
{ {
Console.WriteLine("Transaction3 Rollback"); Console.WriteLine("Transaction3 Rollback");
return Task.CompletedTask; return Task.CompletedTask;
} }
public Task<ActivityExecuteResult> Execute(Transaction3Dto data) public Task<ActivityExecuteResult> Execute(Transaction3Dto data, CancellationToken cancellationToken)
{ {
throw new NotImplementedException(); throw new NotImplementedException();
} }
......
samples/apis/SagasTest.Api/Activities/Transaction3HasResultActivity.cs
...@@ -3,19 +3,20 @@ using Pole.Sagas.Core.Abstraction; ...@@ -3,19 +3,20 @@ using Pole.Sagas.Core.Abstraction;
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace SagasTest.Api.Activities namespace SagasTest.Api.Activities
{ {
public class Transaction3HasResultActivity : IActivity<Transaction3Dto> public class Transaction3HasResultActivity : IActivity<Transaction3Dto>
{ {
public Task Compensate(Transaction3Dto data) public Task Compensate(Transaction3Dto data, CancellationToken cancellationToken)
{ {
Console.WriteLine("Transaction3 Rollback"); Console.WriteLine("Transaction3 Rollback");
return Task.CompletedTask; return Task.CompletedTask;
} }
public Task<ActivityExecuteResult> Execute(Transaction3Dto data) public Task<ActivityExecuteResult> Execute(Transaction3Dto data, CancellationToken cancellationToken)
{ {
Console.WriteLine("Transaction3 commit"); Console.WriteLine("Transaction3 commit");
var result = new ActivityExecuteResult var result = new ActivityExecuteResult
......
samples/apis/SagasTest.Api/Activities/Transaction3ReturnFalseActivity.cs
...@@ -3,19 +3,20 @@ using Pole.Sagas.Core.Abstraction; ...@@ -3,19 +3,20 @@ using Pole.Sagas.Core.Abstraction;
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace SagasTest.Api.Activities namespace SagasTest.Api.Activities
{ {
public class Transaction3ReturnFalseActivity : IActivity<Transaction3Dto> public class Transaction3ReturnFalseActivity : IActivity<Transaction3Dto>
{ {
public Task Compensate(Transaction3Dto data) public Task Compensate(Transaction3Dto data, CancellationToken cancellationToken)
{ {
Console.WriteLine("Transaction3 Rollback"); Console.WriteLine("Transaction3 Rollback");
return Task.CompletedTask; return Task.CompletedTask;
} }
public Task<ActivityExecuteResult> Execute(Transaction3Dto data) public Task<ActivityExecuteResult> Execute(Transaction3Dto data, CancellationToken cancellationToken)
{ {
Console.WriteLine("Transaction3 commit"); Console.WriteLine("Transaction3 commit");
var result = new ActivityExecuteResult var result = new ActivityExecuteResult
......
samples/apis/SagasTest.Api/Controllers/SagasTestController.cs
...@@ -117,29 +117,5 @@ namespace SagasTest.Api.Controllers ...@@ -117,29 +117,5 @@ namespace SagasTest.Api.Controllers
var result = await sagas.GetResult(); var result = await sagas.GetResult();
} }
// GET api/values/5
[HttpGet("{id}")]
public ActionResult<string> Get(int id)
{
return "value";
}
// POST api/values
[HttpPost]
public void Post([FromBody] string value)
{
}
// PUT api/values/5
[HttpPut("{id}")]
public void Put(int id, [FromBody] string value)
{
}
// DELETE api/values/5
[HttpDelete("{id}")]
public void Delete(int id)
{
}
} }
} }
src/Pole.Sagas/Core/Abstraction/IActivity.cs
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Text; using System.Text;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace Pole.Sagas.Core.Abstraction namespace Pole.Sagas.Core.Abstraction
{ {
public interface IActivity<TData> public interface IActivity<TData>
{ {
Task<ActivityExecuteResult> Execute(TData data); Task<ActivityExecuteResult> Execute(TData data ,CancellationToken cancellationToken);
Task Compensate(TData data); Task Compensate(TData data, CancellationToken cancellationToken);
} }
} }
src/Pole.Sagas/Core/Abstraction/IEventSender.cs
...@@ -6,12 +6,14 @@ namespace Pole.Sagas.Core.Abstraction ...@@ -6,12 +6,14 @@ namespace Pole.Sagas.Core.Abstraction
public interface IEventSender public interface IEventSender
{ {
Task SagaStarted(string sagaId, string serviceName); Task SagaStarted(string sagaId, string serviceName);
Task SagaEnded(string sagaId,DateTime ExpiresAt); Task SagaEnded(string sagaId, DateTime ExpiresAt);
Task ActivityExecuteStarted(string activityId, string sagaId, DateTime activityTimeoutTime, string parameterContent,int order); Task ActivityExecuteStarted(string activityId, string sagaId, int timeOutSeconds, string parameterContent, int order);
Task ActivityRetried(string activityId, string status, int retries,string resultContent); Task ActivityRetried(string activityId, string status, int retries, string resultContent);
Task ActivityExecuteAborted(string activityId,string resultContent,string errors); Task ActivityExecuteAborted(string activityId, string errors);
Task ActivityCompensateAborted(string activityId,string sagaId,string errors); Task ActivityCompensateAborted(string activityId, string sagaId, string errors);
Task ActivityEnded(string activityId, string resultContent); Task ActivityEnded(string activityId, string resultContent);
Task ActivityCompensated(string activityId); Task ActivityCompensated(string activityId);
Task ActivityExecuteOvertime(string activityId, string sagaId, string errors);
Task ActivityRevoked(string activityId);
} }
} }
src/Pole.Sagas/Core/ActivityStatus.cs
...@@ -7,9 +7,13 @@ namespace Pole.Sagas.Core ...@@ -7,9 +7,13 @@ namespace Pole.Sagas.Core
public enum ActivityStatus public enum ActivityStatus
{ {
NotStarted, NotStarted,
Executing,
Executed, Executed,
Compensating,
Compensated, Compensated,
ExecuteAborted, ExecuteAborted,
Revoked,
CompensateAborted, CompensateAborted,
Overtime
} }
} }
src/Pole.Sagas/Core/ActivityWapper.cs
...@@ -3,6 +3,7 @@ using System; ...@@ -3,6 +3,7 @@ using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq.Expressions; using System.Linq.Expressions;
using System.Text; using System.Text;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace Pole.Sagas.Core namespace Pole.Sagas.Core
...@@ -16,21 +17,23 @@ namespace Pole.Sagas.Core ...@@ -16,21 +17,23 @@ namespace Pole.Sagas.Core
public int Order { get; set; } public int Order { get; set; }
public ActivityStatus ActivityState { get; set; } public ActivityStatus ActivityState { get; set; }
public IServiceProvider ServiceProvider { get; set; } public IServiceProvider ServiceProvider { get; set; }
public DateTime TimeOut { get; set; } public int TimeOutSeconds { get; set; }
public CancellationTokenSource CancellationTokenSource { get; set; }
public Task<ActivityExecuteResult> InvokeExecute() public Task<ActivityExecuteResult> InvokeExecute()
{ {
var activityObjParams = Expression.Parameter(typeof(object), "activity"); var activityObjParams = Expression.Parameter(typeof(object), "activity");
var activityParams = Expression.Convert(activityObjParams, ActivityType); var activityParams = Expression.Convert(activityObjParams, ActivityType);
var dataObjParams = Expression.Parameter(typeof(object), "data"); var dataObjParams = Expression.Parameter(typeof(object), "data");
var dataParams = Expression.Convert(dataObjParams, ActivityDataType); var dataParams = Expression.Convert(dataObjParams, ActivityDataType);
var method = ActivityType.GetMethod("Execute", new Type[] { ActivityDataType }); var cancellationTokenParams = Expression.Parameter(typeof(CancellationToken), "ct");
var body = Expression.Call(activityParams, method, dataParams); var method = ActivityType.GetMethod("Execute", new Type[] { ActivityDataType, typeof(CancellationToken) });
var func = Expression.Lambda<Func<object, object, Task<ActivityExecuteResult>>>(body, activityObjParams, dataObjParams).Compile(); var body = Expression.Call(activityParams, method, dataParams, cancellationTokenParams);
var func = Expression.Lambda<Func<object, object, CancellationToken, Task<ActivityExecuteResult>>>(body, true, activityObjParams, dataObjParams, cancellationTokenParams).Compile();
using (var scope = ServiceProvider.CreateScope()) using (var scope = ServiceProvider.CreateScope())
{ {
var activity = scope.ServiceProvider.GetRequiredService(ActivityType); var activity = scope.ServiceProvider.GetRequiredService(ActivityType);
return func(activity, DataObj); return func(activity, DataObj, CancellationTokenSource.Token);
} }
} }
public Task InvokeCompensate() public Task InvokeCompensate()
...@@ -39,14 +42,15 @@ namespace Pole.Sagas.Core ...@@ -39,14 +42,15 @@ namespace Pole.Sagas.Core
var activityParams = Expression.Convert(activityObjParams, ActivityType); var activityParams = Expression.Convert(activityObjParams, ActivityType);
var dataObjParams = Expression.Parameter(typeof(object), "data"); var dataObjParams = Expression.Parameter(typeof(object), "data");
var dataParams = Expression.Convert(dataObjParams, ActivityDataType); var dataParams = Expression.Convert(dataObjParams, ActivityDataType);
var method = ActivityType.GetMethod("Compensate", new Type[] { ActivityDataType }); var cancellationTokenParams = Expression.Parameter(typeof(CancellationToken), "ct");
var body = Expression.Call(activityParams, method, dataParams); var method = ActivityType.GetMethod("Compensate", new Type[] { ActivityDataType, typeof(CancellationToken) });
var func = Expression.Lambda<Func<object, object, Task>>(body, activityObjParams, dataObjParams).Compile(); var body = Expression.Call(activityParams, method, dataParams, cancellationTokenParams);
var func = Expression.Lambda<Func<object, object, CancellationToken, Task>>(body, activityObjParams, dataObjParams, cancellationTokenParams).Compile();
using (var scope = ServiceProvider.CreateScope()) using (var scope = ServiceProvider.CreateScope())
{ {
var activity = scope.ServiceProvider.GetRequiredService(ActivityType); var activity = scope.ServiceProvider.GetRequiredService(ActivityType);
return func(activity, DataObj); return func(activity, DataObj, CancellationTokenSource.Token);
} }
} }
} }
......
src/Pole.Sagas/Core/EventSender.cs
...@@ -23,7 +23,7 @@ namespace Pole.Sagas.Core ...@@ -23,7 +23,7 @@ namespace Pole.Sagas.Core
return Task.CompletedTask; return Task.CompletedTask;
} }
public Task ActivityExecuteAborted(string activityId, string resultContent, string errors) public Task ActivityExecuteAborted(string activityId, string errors)
{ {
return Task.CompletedTask; return Task.CompletedTask;
} }
...@@ -33,7 +33,7 @@ namespace Pole.Sagas.Core ...@@ -33,7 +33,7 @@ namespace Pole.Sagas.Core
return Task.CompletedTask; return Task.CompletedTask;
} }
public Task ActivityExecuteStarted(string activityId, string sagaId, DateTime activityTimeoutTime, string parameterContent, int order) public Task ActivityExecuteStarted(string activityId, string sagaId, int timeoutSeconds, string parameterContent, int order)
{ {
return Task.CompletedTask; return Task.CompletedTask;
} }
...@@ -47,5 +47,15 @@ namespace Pole.Sagas.Core ...@@ -47,5 +47,15 @@ namespace Pole.Sagas.Core
{ {
return Task.CompletedTask; return Task.CompletedTask;
} }
public Task ActivityExecuteOvertime(string activityId, string sagaId, string errors)
{
return Task.CompletedTask;
}
public Task ActivityRevoked(string activityId)
{
throw new NotImplementedException();
}
} }
} }
src/Pole.Sagas/Core/ISaga.cs
...@@ -8,7 +8,7 @@ namespace Pole.Sagas.Core ...@@ -8,7 +8,7 @@ namespace Pole.Sagas.Core
public interface ISaga public interface ISaga
{ {
string Id { get; } string Id { get; }
void AddActivity(string activityName, object data); void AddActivity(string activityName, object data,int timeOutSeconds=2);
Task<SagaResult> GetResult(); Task<SagaResult> GetResult();
} }
} }
src/Pole.Sagas/Core/Saga.cs
...@@ -18,13 +18,22 @@ namespace Pole.Sagas.Core ...@@ -18,13 +18,22 @@ namespace Pole.Sagas.Core
private ISnowflakeIdGenerator snowflakeIdGenerator; private ISnowflakeIdGenerator snowflakeIdGenerator;
private IActivityFinder activityFinder; private IActivityFinder activityFinder;
private PoleSagasOption poleSagasOption; private PoleSagasOption poleSagasOption;
private int _currentMaxOrder = 0; public int CurrentMaxOrder
{
get { return activities.Count; }
}
/// <summary>
/// 如果 等于 -1 说明已经在执行补偿操作,此时这个值已经没有意义
/// </summary>
private int _currentExecuteOrder = 0; private int _currentExecuteOrder = 0;
private int _currentCompensateOrder = 0; /// <summary>
/// 如果 等于 -1 说明已经还未执行补偿操作,此时这个值没有意义
/// </summary>
private int _currentCompensateOrder = -1;
private ISerializer serializer; private ISerializer serializer;
public string Id { get; } public string Id { get; }
public Saga(ISnowflakeIdGenerator snowflakeIdGenerator, IServiceProvider serviceProvider, IEventSender eventSender, PoleSagasOption poleSagasOption, ISerializer serializer, IActivityFinder activityFinder) internal Saga(ISnowflakeIdGenerator snowflakeIdGenerator, IServiceProvider serviceProvider, IEventSender eventSender, PoleSagasOption poleSagasOption, ISerializer serializer, IActivityFinder activityFinder)
{ {
this.snowflakeIdGenerator = snowflakeIdGenerator; this.snowflakeIdGenerator = snowflakeIdGenerator;
this.serviceProvider = serviceProvider; this.serviceProvider = serviceProvider;
...@@ -35,7 +44,7 @@ namespace Pole.Sagas.Core ...@@ -35,7 +44,7 @@ namespace Pole.Sagas.Core
Id = snowflakeIdGenerator.NextId(); Id = snowflakeIdGenerator.NextId();
} }
public void AddActivity(string activityName, object data) public void AddActivity(string activityName, object data, int timeOutSeconds = 2)
{ {
var targetActivityType = activityFinder.FindType(activityName); var targetActivityType = activityFinder.FindType(activityName);
...@@ -45,15 +54,15 @@ namespace Pole.Sagas.Core ...@@ -45,15 +54,15 @@ namespace Pole.Sagas.Core
throw new ActivityImplementIrregularException(activityName); throw new ActivityImplementIrregularException(activityName);
} }
var dataType = activityInterface.GetGenericArguments()[0]; var dataType = activityInterface.GetGenericArguments()[0];
_currentMaxOrder++;
ActivityWapper activityWapper = new ActivityWapper ActivityWapper activityWapper = new ActivityWapper
{ {
ActivityDataType = dataType, ActivityDataType = dataType,
ActivityState = ActivityStatus.NotStarted, ActivityState = ActivityStatus.NotStarted,
ActivityType = targetActivityType, ActivityType = targetActivityType,
DataObj = data, DataObj = data,
Order = _currentMaxOrder, Order = CurrentMaxOrder,
ServiceProvider = serviceProvider ServiceProvider = serviceProvider,
TimeOutSeconds = 2,
}; };
activities.Add(activityWapper); activities.Add(activityWapper);
} }
...@@ -75,7 +84,7 @@ namespace Pole.Sagas.Core ...@@ -75,7 +84,7 @@ namespace Pole.Sagas.Core
private ActivityWapper GetNextExecuteActivity() private ActivityWapper GetNextExecuteActivity()
{ {
if (_currentExecuteOrder == _currentMaxOrder) if (_currentExecuteOrder == CurrentMaxOrder)
{ {
return null; return null;
} }
...@@ -115,15 +124,16 @@ namespace Pole.Sagas.Core ...@@ -115,15 +124,16 @@ namespace Pole.Sagas.Core
{ {
var activityId = snowflakeIdGenerator.NextId(); var activityId = snowflakeIdGenerator.NextId();
activityWapper.Id = activityId; activityWapper.Id = activityId;
activityWapper.CancellationTokenSource = new System.Threading.CancellationTokenSource(2 * 1000);
try try
{ {
var jsonContent = serializer.Serialize(activityWapper.DataObj, activityWapper.ActivityDataType); var jsonContent = serializer.Serialize(activityWapper.DataObj, activityWapper.ActivityDataType);
await eventSender.ActivityExecuteStarted(activityId, Id, activityWapper.TimeOut, jsonContent, activityWapper.Order); await eventSender.ActivityExecuteStarted(activityId, Id, activityWapper.TimeOutSeconds, jsonContent, activityWapper.Order);
var result = await activityWapper.InvokeExecute(); var result = await activityWapper.InvokeExecute();
if (!result.IsSuccess) if (!result.IsSuccess)
{ {
await eventSender.ActivityExecuteAborted(activityId, serializer.Serialize(result.Result), string.Empty); await eventSender.ActivityRevoked(activityId);
await CompensateActivity(result); await CompensateActivity(result,_currentExecuteOrder);
return result; return result;
} }
await eventSender.ActivityEnded(activityId, string.Empty); await eventSender.ActivityEnded(activityId, string.Empty);
...@@ -139,20 +149,37 @@ namespace Pole.Sagas.Core ...@@ -139,20 +149,37 @@ namespace Pole.Sagas.Core
} }
catch (Exception exception) catch (Exception exception)
{ {
if (activityWapper.CancellationTokenSource.Token.IsCancellationRequested)
{
var errors = exception.InnerException != null ? exception.InnerException.Message + exception.StackTrace : exception.Message + exception.StackTrace; var errors = exception.InnerException != null ? exception.InnerException.Message + exception.StackTrace : exception.Message + exception.StackTrace;
var result = new ActivityExecuteResult var result = new ActivityExecuteResult
{ {
IsSuccess = false, IsSuccess = false,
Errors = errors Errors = errors
}; };
await eventSender.ActivityExecuteAborted(activityId, string.Empty, errors); await eventSender.ActivityExecuteOvertime(activityId, Id, errors);
return await CompensateActivity(result); // 超时的时候 需要首先补偿这个超时的操作
return await CompensateActivity(result,_currentExecuteOrder+1);
}
else
{
var errors = exception.InnerException != null ? exception.InnerException.Message + exception.StackTrace : exception.Message + exception.StackTrace;
var result = new ActivityExecuteResult
{
IsSuccess = false,
Errors = errors
};
await eventSender.ActivityExecuteAborted(activityId, errors);
// 出错的时候 需要首先补偿这个出错的操作
return await CompensateActivity(result, _currentExecuteOrder + 1);
}
} }
} }
private async Task<ActivityExecuteResult> CompensateActivity(ActivityExecuteResult result) private async Task<ActivityExecuteResult> CompensateActivity(ActivityExecuteResult result,int currentCompensateOrder)
{ {
_currentCompensateOrder = _currentExecuteOrder; _currentCompensateOrder = currentCompensateOrder;
_currentExecuteOrder = -1;
var compensateActivity = GetNextCompensateActivity(); var compensateActivity = GetNextCompensateActivity();
if (compensateActivity == null) if (compensateActivity == null)
{ {
......
src/Pole.Sagas/Core/SagaFactory.cs
...@@ -30,5 +30,9 @@ namespace Pole.Sagas.Core ...@@ -30,5 +30,9 @@ namespace Pole.Sagas.Core
{ {
return new Saga(snowflakeIdGenerator, serviceProvider, eventSender, poleSagasOption, serializer, activityFinder); return new Saga(snowflakeIdGenerator, serviceProvider, eventSender, poleSagasOption, serializer, activityFinder);
} }
internal ISaga CreateSaga(string id)
{
return new Saga(snowflakeIdGenerator, serviceProvider, eventSender, poleSagasOption, serializer, activityFinder);
}
} }
} }
src/Pole.Sagas/Core/SagaStatus.cs
...@@ -8,6 +8,7 @@ namespace Pole.Sagas.Core ...@@ -8,6 +8,7 @@ namespace Pole.Sagas.Core
{ {
Started, Started,
Ended, Ended,
Error Error,
Overtime
} }
} }
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment