Browse Source

Minor fixe

pull/65/head
Sebastian Stehle 9 years ago
parent
commit
cca985129d
  1. 14
      src/Squidex.Infrastructure.MongoDb/MongoEventConsumerInfoRepository.cs
  2. 11
      src/Squidex.Infrastructure/CQRS/Events/EventReceiver.cs
  3. 2
      src/Squidex.Infrastructure/CQRS/Events/IEventConsumerInfoRepository.cs

14
src/Squidex.Infrastructure.MongoDb/MongoEventConsumerInfoRepository.cs

@ -13,6 +13,9 @@ using MongoDB.Bson;
using MongoDB.Driver; using MongoDB.Driver;
using Squidex.Infrastructure.CQRS.Events; using Squidex.Infrastructure.CQRS.Events;
// ReSharper disable ConvertIfStatementToReturnStatement
// ReSharper disable RedundantIfElseBlock
namespace Squidex.Infrastructure.MongoDb namespace Squidex.Infrastructure.MongoDb
{ {
public sealed class MongoEventConsumerInfoRepository : MongoRepositoryBase<MongoEventConsumerInfo>, IEventConsumerInfoRepository public sealed class MongoEventConsumerInfoRepository : MongoRepositoryBase<MongoEventConsumerInfo>, IEventConsumerInfoRepository
@ -74,9 +77,16 @@ namespace Squidex.Infrastructure.MongoDb
return Collection.UpdateOneAsync(x => x.Name == consumerName, Update.Set(x => x.IsResetting, true)); return Collection.UpdateOneAsync(x => x.Name == consumerName, Update.Set(x => x.IsResetting, true));
} }
public Task SetLastHandledEventNumberAsync(string consumerName, string position) public Task SetPositionAsync(string consumerName, string position, bool reset)
{ {
return Collection.ReplaceOneAsync(x => x.Name == consumerName, CreateEntity(consumerName, position)); if (reset)
{
return Collection.ReplaceOneAsync(x => x.Name == consumerName, CreateEntity(consumerName, position));
}
else
{
return Collection.UpdateOneAsync(x => x.Name == consumerName, Update.Set(x => x.Position, position));
}
} }
private static MongoEventConsumerInfo CreateEntity(string consumerName, string position) private static MongoEventConsumerInfo CreateEntity(string consumerName, string position)

11
src/Squidex.Infrastructure/CQRS/Events/EventReceiver.cs

@ -11,6 +11,7 @@ using System.Threading.Tasks;
using Squidex.Infrastructure.Log; using Squidex.Infrastructure.Log;
using Squidex.Infrastructure.Timers; using Squidex.Infrastructure.Timers;
// ReSharper disable ExpressionIsAlwaysNull
// ReSharper disable ConvertToLambdaExpression // ReSharper disable ConvertToLambdaExpression
// ReSharper disable MethodSupportsCancellation // ReSharper disable MethodSupportsCancellation
// ReSharper disable ConvertIfStatementToConditionalTernaryExpression // ReSharper disable ConvertIfStatementToConditionalTernaryExpression
@ -102,9 +103,9 @@ namespace Squidex.Infrastructure.CQRS.Events
if (status.IsResetting) if (status.IsResetting)
{ {
await ResetAsync(eventConsumer, consumerName);
position = null; position = null;
await ResetAsync(eventConsumer, consumerName, position);
} }
else if (status.IsStopped) else if (status.IsStopped)
{ {
@ -138,10 +139,10 @@ namespace Squidex.Infrastructure.CQRS.Events
await DispatchConsumer(@event, eventConsumer, consumerName); await DispatchConsumer(@event, eventConsumer, consumerName);
await eventConsumerInfoRepository.SetLastHandledEventNumberAsync(consumerName, storedEvent.EventPosition); await eventConsumerInfoRepository.SetPositionAsync(consumerName, storedEvent.EventPosition, false);
} }
private async Task ResetAsync(IEventConsumer eventConsumer, string consumerName) private async Task ResetAsync(IEventConsumer eventConsumer, string consumerName, string position)
{ {
var actionId = Guid.NewGuid().ToString(); var actionId = Guid.NewGuid().ToString();
try try
@ -153,7 +154,7 @@ namespace Squidex.Infrastructure.CQRS.Events
.WriteProperty("eventConsumer", eventConsumer.GetType().Name)); .WriteProperty("eventConsumer", eventConsumer.GetType().Name));
await eventConsumer.ClearAsync(); await eventConsumer.ClearAsync();
await eventConsumerInfoRepository.SetLastHandledEventNumberAsync(consumerName, null); await eventConsumerInfoRepository.SetPositionAsync(consumerName, position, true);
log.LogInformation(w => w log.LogInformation(w => w
.WriteProperty("action", "EventConsumerReset") .WriteProperty("action", "EventConsumerReset")

2
src/Squidex.Infrastructure/CQRS/Events/IEventConsumerInfoRepository.cs

@ -25,6 +25,6 @@ namespace Squidex.Infrastructure.CQRS.Events
Task ResetAsync(string consumerName); Task ResetAsync(string consumerName);
Task SetLastHandledEventNumberAsync(string consumerName, string position); Task SetPositionAsync(string consumerName, string position, bool reset);
} }
} }

Loading…
Cancel
Save