nhyatt/bitwarden
1
0
Fork 0
mirror of https://github.com/bitwarden/server.git synced 2025-07-07 19:05:07 -05:00
Code Activity

Revert filescoped (#2227)

Browse Source 
* Revert "Add git blame entry (#2226)"

This reverts commit 239286737d.

* Revert "Turn on file scoped namespaces (#2225)"

This reverts commit 34fb4cca2a.
This commit is contained in:
Justin Baur
2022-08-29 15:53:48 -04:00
committed by GitHub
parent 239286737d
commit bae03feffe
1208 changed files with 74317 additions and 73126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

185
src/EventsProcessor/AzureQueueHostedService.cs
View File

@ -5,121 +5,122 @@ using Bit.Core.Models.Data;
using Bit.Core.Services;
using Bit.Core.Utilities;
namespace Bit.EventsProcessor;
public class AzureQueueHostedService : IHostedService, IDisposable
namespace Bit.EventsProcessor
{
private readonly ILogger<AzureQueueHostedService> _logger;
private readonly IConfiguration _configuration;
private Task _executingTask;
private CancellationTokenSource _cts;
private QueueClient _queueClient;
private IEventWriteService _eventWriteService;
public AzureQueueHostedService(
ILogger<AzureQueueHostedService> logger,
IConfiguration configuration)
public class AzureQueueHostedService : IHostedService, IDisposable
{
_logger = logger;
_configuration = configuration;
}
private readonly ILogger<AzureQueueHostedService> _logger;
private readonly IConfiguration _configuration;
public Task StartAsync(CancellationToken cancellationToken)
{
_logger.LogInformation(Constants.BypassFiltersEventId, "Starting service.");
_cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
_executingTask = ExecuteAsync(_cts.Token);
return _executingTask.IsCompleted ? _executingTask : Task.CompletedTask;
}
private Task _executingTask;
private CancellationTokenSource _cts;
private QueueClient _queueClient;
private IEventWriteService _eventWriteService;
public async Task StopAsync(CancellationToken cancellationToken)
{
if (_executingTask == null)
public AzureQueueHostedService(
ILogger<AzureQueueHostedService> logger,
IConfiguration configuration)
{
return;
}
_logger.LogWarning("Stopping service.");
_cts.Cancel();
await Task.WhenAny(_executingTask, Task.Delay(-1, cancellationToken));
cancellationToken.ThrowIfCancellationRequested();
}
public void Dispose()
{ }
private async Task ExecuteAsync(CancellationToken cancellationToken)
{
var storageConnectionString = _configuration["azureStorageConnectionString"];
if (string.IsNullOrWhiteSpace(storageConnectionString))
{
return;
_logger = logger;
_configuration = configuration;
}
var repo = new Core.Repositories.TableStorage.EventRepository(storageConnectionString);
_eventWriteService = new RepositoryEventWriteService(repo);
_queueClient = new QueueClient(storageConnectionString, "event");
while (!cancellationToken.IsCancellationRequested)
public Task StartAsync(CancellationToken cancellationToken)
{
try
_logger.LogInformation(Constants.BypassFiltersEventId, "Starting service.");
_cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
_executingTask = ExecuteAsync(_cts.Token);
return _executingTask.IsCompleted ? _executingTask : Task.CompletedTask;
}
public async Task StopAsync(CancellationToken cancellationToken)
{
if (_executingTask == null)
{
var messages = await _queueClient.ReceiveMessagesAsync(32);
if (messages.Value?.Any() ?? false)
return;
}
_logger.LogWarning("Stopping service.");
_cts.Cancel();
await Task.WhenAny(_executingTask, Task.Delay(-1, cancellationToken));
cancellationToken.ThrowIfCancellationRequested();
}
public void Dispose()
{ }
private async Task ExecuteAsync(CancellationToken cancellationToken)
{
var storageConnectionString = _configuration["azureStorageConnectionString"];
if (string.IsNullOrWhiteSpace(storageConnectionString))
{
return;
}
var repo = new Core.Repositories.TableStorage.EventRepository(storageConnectionString);
_eventWriteService = new RepositoryEventWriteService(repo);
_queueClient = new QueueClient(storageConnectionString, "event");
while (!cancellationToken.IsCancellationRequested)
{
try
{
foreach (var message in messages.Value)
var messages = await _queueClient.ReceiveMessagesAsync(32);
if (messages.Value?.Any() ?? false)
{
await ProcessQueueMessageAsync(message.DecodeMessageText(), cancellationToken);
await _queueClient.DeleteMessageAsync(message.MessageId, message.PopReceipt);
foreach (var message in messages.Value)
{
await ProcessQueueMessageAsync(message.DecodeMessageText(), cancellationToken);
await _queueClient.DeleteMessageAsync(message.MessageId, message.PopReceipt);
}
}
else
{
await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken);
}
}
else
catch (Exception e)
{
_logger.LogError(e, "Exception occurred: " + e.Message);
await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken);
}
}
catch (Exception e)
{
_logger.LogError(e, "Exception occurred: " + e.Message);
await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken);
}
_logger.LogWarning("Done processing.");
}
_logger.LogWarning("Done processing.");
}
public async Task ProcessQueueMessageAsync(string message, CancellationToken cancellationToken)
{
if (_eventWriteService == null || message == null || message.Length == 0)
public async Task ProcessQueueMessageAsync(string message, CancellationToken cancellationToken)
{
return;
}
try
{
_logger.LogInformation("Processing message.");
var events = new List<IEvent>();
using var jsonDocument = JsonDocument.Parse(message);
var root = jsonDocument.RootElement;
if (root.ValueKind == JsonValueKind.Array)
if (_eventWriteService == null || message == null || message.Length == 0)
{
var indexedEntities = root.ToObject<List<EventMessage>>()
.SelectMany(e => EventTableEntity.IndexEvent(e));
events.AddRange(indexedEntities);
}
else if (root.ValueKind == JsonValueKind.Object)
{
var eventMessage = root.ToObject<EventMessage>();
events.AddRange(EventTableEntity.IndexEvent(eventMessage));
return;
}
await _eventWriteService.CreateManyAsync(events);
_logger.LogInformation("Processed message.");
}
catch (JsonException)
{
_logger.LogError("JsonReaderException: Unable to parse message.");
try
{
_logger.LogInformation("Processing message.");
var events = new List<IEvent>();
using var jsonDocument = JsonDocument.Parse(message);
var root = jsonDocument.RootElement;
if (root.ValueKind == JsonValueKind.Array)
{
var indexedEntities = root.ToObject<List<EventMessage>>()
.SelectMany(e => EventTableEntity.IndexEvent(e));
events.AddRange(indexedEntities);
}
else if (root.ValueKind == JsonValueKind.Object)
{
var eventMessage = root.ToObject<EventMessage>();
events.AddRange(EventTableEntity.IndexEvent(eventMessage));
}
await _eventWriteService.CreateManyAsync(events);
_logger.LogInformation("Processed message.");
}
catch (JsonException)
{
_logger.LogError("JsonReaderException: Unable to parse message.");
}
}
}
}