Browse Source

Small bugfix when resetting.

pull/135/head
Sebastian Stehle 9 years ago
parent
commit
01f793abcc
  1. 22
      src/Squidex.Infrastructure/CQRS/Events/Actors/EventConsumerActor.cs
  2. 6
      tests/Squidex.Infrastructure.Tests/CQRS/Events/Actors/EventConsumerActorTests.cs

22
src/Squidex.Infrastructure/CQRS/Events/Actors/EventConsumerActor.cs

@ -158,7 +158,7 @@ namespace Squidex.Infrastructure.CQRS.Events.Actors
if (statusIsRunning) if (statusIsRunning)
{ {
await SubscribeAsync(); await SubscribeThisAsync(statusPosition);
} }
break; break;
@ -171,7 +171,7 @@ namespace Squidex.Infrastructure.CQRS.Events.Actors
return; return;
} }
await SubscribeAsync(); await SubscribeThisAsync(statusPosition);
statusError = null; statusError = null;
statusIsRunning = true; statusIsRunning = true;
@ -186,7 +186,7 @@ namespace Squidex.Infrastructure.CQRS.Events.Actors
return; return;
} }
await UnsubscribeAsync(); await UnsubscribeThisAsync();
statusIsRunning = false; statusIsRunning = false;
@ -195,9 +195,9 @@ namespace Squidex.Infrastructure.CQRS.Events.Actors
case ResetConsumerMessage resetConsumer: case ResetConsumerMessage resetConsumer:
{ {
await UnsubscribeAsync(); await UnsubscribeThisAsync();
await ClearAsync(); await ClearAsync();
await SubscribeAsync(); await SubscribeThisAsync(null);
statusError = null; statusError = null;
statusPosition = null; statusPosition = null;
@ -213,7 +213,7 @@ namespace Squidex.Infrastructure.CQRS.Events.Actors
return; return;
} }
await SubscribeAsync(); await SubscribeThisAsync(statusPosition);
break; break;
} }
@ -225,7 +225,7 @@ namespace Squidex.Infrastructure.CQRS.Events.Actors
return; return;
} }
await UnsubscribeAsync(); await UnsubscribeThisAsync();
if (retryWindow.CanRetryAfterFailure()) if (retryWindow.CanRetryAfterFailure())
{ {
@ -263,7 +263,7 @@ namespace Squidex.Infrastructure.CQRS.Events.Actors
{ {
try try
{ {
await UnsubscribeAsync(); await UnsubscribeThisAsync();
} }
catch (Exception unsubscribeException) catch (Exception unsubscribeException)
{ {
@ -282,7 +282,7 @@ namespace Squidex.Infrastructure.CQRS.Events.Actors
} }
} }
private async Task UnsubscribeAsync() private async Task UnsubscribeThisAsync()
{ {
if (eventSubscription != null) if (eventSubscription != null)
{ {
@ -292,11 +292,11 @@ namespace Squidex.Infrastructure.CQRS.Events.Actors
} }
} }
private Task SubscribeAsync() private Task SubscribeThisAsync(string position)
{ {
if (eventSubscription == null) if (eventSubscription == null)
{ {
eventSubscription = eventStore.CreateSubscription(this, eventConsumer.EventsFilter, statusPosition); eventSubscription = eventStore.CreateSubscription(this, eventConsumer.EventsFilter, position);
} }
return TaskHelper.Done; return TaskHelper.Done;

6
tests/Squidex.Infrastructure.Tests/CQRS/Events/Actors/EventConsumerActorTests.cs

@ -151,6 +151,12 @@ namespace Squidex.Infrastructure.CQRS.Events.Actors
A.CallTo(() => eventSubscription.StopAsync()) A.CallTo(() => eventSubscription.StopAsync())
.MustHaveHappened(Repeated.Exactly.Once); .MustHaveHappened(Repeated.Exactly.Once);
A.CallTo(() => eventStore.CreateSubscription(A<IEventSubscriber>.Ignored, A<string>.Ignored, consumerInfo.Position))
.MustHaveHappened(Repeated.Exactly.Once);
A.CallTo(() => eventStore.CreateSubscription(A<IEventSubscriber>.Ignored, A<string>.Ignored, null))
.MustHaveHappened(Repeated.Exactly.Once);
} }
[Fact] [Fact]

Loading…
Cancel
Save