ayan.poddar/eShopOnContainers
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Update EventBusServiceBus.cs

Browse Source 
Do not complete messages until actually processed.
This commit is contained in:
i02coroj 2018-04-13 12:03:34 +02:00 committed by GitHub
parent 1b8806e43d
commit 8b846968e2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 9 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/BuildingBlocks/EventBus/EventBusServiceBus/EventBusServiceBus.cs
View File

@ -129,10 +129,13 @@
{ {
var eventName = $"{message.Label}{INTEGRATION_EVENT_SUFIX}"; var eventName = $"{message.Label}{INTEGRATION_EVENT_SUFIX}";
var messageData = Encoding.UTF8.GetString(message.Body); var messageData = Encoding.UTF8.GetString(message.Body);
await ProcessEvent(eventName, messageData); var processed = await ProcessEvent(eventName, messageData);
// Complete the message so that it is not received again. // Complete the message so that it is not received again.
await _subscriptionClient.CompleteAsync(message.SystemProperties.LockToken); if (processed)
{
await _subscriptionClient.CompleteAsync(message.SystemProperties.LockToken);
}
}, },
new MessageHandlerOptions(ExceptionReceivedHandler) { MaxConcurrentCalls = 10, AutoComplete = false }); new MessageHandlerOptions(ExceptionReceivedHandler) { MaxConcurrentCalls = 10, AutoComplete = false });
} }
@ -148,8 +151,9 @@
return Task.CompletedTask; return Task.CompletedTask;
} }
private async Task ProcessEvent(string eventName, string message) private async Task<bool> ProcessEvent(string eventName, string message)
{ {
var processed = false;
if (_subsManager.HasSubscriptionsForEvent(eventName)) if (_subsManager.HasSubscriptionsForEvent(eventName))
{ {
using (var scope = _autofac.BeginLifetimeScope(AUTOFAC_SCOPE_NAME)) using (var scope = _autofac.BeginLifetimeScope(AUTOFAC_SCOPE_NAME))
@ -173,7 +177,9 @@
} }
} }
} }
processed = true;
} }
return processed
} }
private void RemoveDefaultRule() private void RemoveDefaultRule()