Browse Source

Potential fix in actors.

pull/131/head
Sebastian Stehle 9 years ago
parent
commit
d15ee5b26b
  1. 2
      src/Squidex.Infrastructure.GetEventStore/CQRS/Events/GetEventStoreSubscription.cs
  2. 69
      src/Squidex.Infrastructure.MongoDb/CQRS/Events/PollingSubscription.cs
  3. 9
      src/Squidex.Infrastructure/Log/Internal/ConsoleLogProcessor.cs
  4. 9
      src/Squidex.Infrastructure/Log/Internal/FileLogProcessor.cs

2
src/Squidex.Infrastructure.GetEventStore/CQRS/Events/GetEventStoreSubscription.cs

@ -197,7 +197,7 @@ namespace Squidex.Infrastructure.CQRS.Events
} }
catch (Exception ex) catch (Exception ex)
{ {
if (!(ex is ProjectionCommandConflictException)) if (!ex.Is<ProjectionCommandConflictException>())
{ {
throw; throw;
} }

69
src/Squidex.Infrastructure.MongoDb/CQRS/Events/PollingSubscription.cs

@ -69,55 +69,55 @@ namespace Squidex.Infrastructure.CQRS.Events
switch (message) switch (message)
{ {
case SubscribeMessage subscribe when parent == null: case SubscribeMessage subscribe when parent == null:
{ {
parent = subscribe.Parent; parent = subscribe.Parent;
position = subscribe.Position; position = subscribe.Position;
streamFilter = subscribe.StreamFilter; streamFilter = subscribe.StreamFilter;
streamRegex = new Regex(streamFilter); streamRegex = new Regex(streamFilter);
pollSubscription = eventNotifier.Subscribe(streamName => pollSubscription = eventNotifier.Subscribe(streamName =>
{
if (streamRegex.IsMatch(streamName))
{ {
if (streamRegex.IsMatch(streamName)) SendAsync(new StartPollMessage()).Forget();
{ }
SendAsync(new StartPollMessage()).Forget(); });
}
});
SendAsync(new StartPollMessage()).Forget(); SendAsync(new StartPollMessage()).Forget();
break; break;
} }
case StartPollMessage poll when parent != null: case StartPollMessage poll when parent != null:
{
if (!isPolling)
{ {
if (!isPolling) isPolling = true;
{
isPolling = true;
PollAsync().Forget();
}
break; PollAsync().Forget();
} }
break;
}
case StopPollMessage poll when parent != null: case StopPollMessage poll when parent != null:
{ {
isPolling = false; isPolling = false;
Task.Delay(5000).ContinueWith(t => SendAsync(new StartPollMessage())).Forget(); Task.Delay(5000).ContinueWith(t => SendAsync(new StartPollMessage())).Forget();
break; break;
} }
case ReceiveEventMessage receiveEvent when parent != null: case ReceiveEventMessage receiveEvent when parent != null:
{ {
await parent.SendAsync(receiveEvent); await parent.SendAsync(receiveEvent);
position = receiveEvent.Event.EventPosition; position = receiveEvent.Event.EventPosition;
break; break;
} }
} }
} }
@ -129,9 +129,12 @@ namespace Squidex.Infrastructure.CQRS.Events
await SendAsync(new StopPollMessage()); await SendAsync(new StopPollMessage());
} }
catch (Exception ex) when (!(ex is OperationCanceledException)) catch (Exception ex)
{ {
await SendAsync(ex); if (!ex.Is<OperationCanceledException>())
{
await SendAsync(ex);
}
} }
} }
} }

9
src/Squidex.Infrastructure/Log/Internal/ConsoleLogProcessor.cs

@ -65,11 +65,12 @@ namespace Squidex.Infrastructure.Log.Internal
{ {
outputTask.Wait(1500); outputTask.Wait(1500);
} }
catch (TaskCanceledException) catch (Exception ex)
{
}
catch (AggregateException ex) when (ex.InnerExceptions.Count == 1 && ex.InnerExceptions[0] is TaskCanceledException)
{ {
if (!ex.Is<OperationCanceledException>())
{
throw;
}
} }
} }
} }

9
src/Squidex.Infrastructure/Log/Internal/FileLogProcessor.cs

@ -39,11 +39,12 @@ namespace Squidex.Infrastructure.Log.Internal
{ {
outputTask.Wait(1500); outputTask.Wait(1500);
} }
catch (TaskCanceledException) catch (Exception ex)
{
}
catch (AggregateException ex) when (ex.InnerExceptions.Count == 1 && ex.InnerExceptions[0] is TaskCanceledException)
{ {
if (!ex.Is<OperationCanceledException>())
{
throw;
}
} }
} }
} }

Loading…
Cancel
Save