mirror of
https://github.com/bitwarden/server.git
synced 2025-07-02 00:22:50 -05:00
Turn on file scoped namespaces (#2225)
This commit is contained in:
Justin Baur
@ -5,122 +5,121 @@ using Bit.Core.Models.Data;
|
||||
using Bit.Core.Services;
|
||||
using Bit.Core.Utilities;
|
||||
|
||||
namespace Bit.EventsProcessor
|
||||
namespace Bit.EventsProcessor;
|
||||
|
||||
public class AzureQueueHostedService : IHostedService, IDisposable
|
||||
{
|
||||
public class AzureQueueHostedService : IHostedService, IDisposable
|
||||
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)
|
||||
{
|
||||
private readonly ILogger<AzureQueueHostedService> _logger;
|
||||
private readonly IConfiguration _configuration;
|
||||
_logger = logger;
|
||||
_configuration = configuration;
|
||||
}
|
||||
|
||||
private Task _executingTask;
|
||||
private CancellationTokenSource _cts;
|
||||
private QueueClient _queueClient;
|
||||
private IEventWriteService _eventWriteService;
|
||||
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;
|
||||
}
|
||||
|
||||
public AzureQueueHostedService(
|
||||
ILogger<AzureQueueHostedService> logger,
|
||||
IConfiguration configuration)
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
if (_executingTask == null)
|
||||
{
|
||||
_logger = logger;
|
||||
_configuration = 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;
|
||||
}
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
var repo = new Core.Repositories.TableStorage.EventRepository(storageConnectionString);
|
||||
_eventWriteService = new RepositoryEventWriteService(repo);
|
||||
_queueClient = new QueueClient(storageConnectionString, "event");
|
||||
|
||||
while (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
_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)
|
||||
try
|
||||
{
|
||||
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
|
||||
var messages = await _queueClient.ReceiveMessagesAsync(32);
|
||||
if (messages.Value?.Any() ?? false)
|
||||
{
|
||||
var messages = await _queueClient.ReceiveMessagesAsync(32);
|
||||
if (messages.Value?.Any() ?? false)
|
||||
foreach (var message in messages.Value)
|
||||
{
|
||||
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);
|
||||
await ProcessQueueMessageAsync(message.DecodeMessageText(), cancellationToken);
|
||||
await _queueClient.DeleteMessageAsync(message.MessageId, message.PopReceipt);
|
||||
}
|
||||
}
|
||||
catch (Exception e)
|
||||
else
|
||||
{
|
||||
_logger.LogError(e, "Exception occurred: " + e.Message);
|
||||
await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
_logger.LogWarning("Done processing.");
|
||||
catch (Exception e)
|
||||
{
|
||||
_logger.LogError(e, "Exception occurred: " + e.Message);
|
||||
await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task ProcessQueueMessageAsync(string message, CancellationToken cancellationToken)
|
||||
_logger.LogWarning("Done processing.");
|
||||
}
|
||||
|
||||
public async Task ProcessQueueMessageAsync(string message, CancellationToken cancellationToken)
|
||||
{
|
||||
if (_eventWriteService == null || message == null || message.Length == 0)
|
||||
{
|
||||
if (_eventWriteService == null || message == null || message.Length == 0)
|
||||
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)
|
||||
{
|
||||
return;
|
||||
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));
|
||||
}
|
||||
|
||||
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.");
|
||||
}
|
||||
await _eventWriteService.CreateManyAsync(events);
|
||||
_logger.LogInformation("Processed message.");
|
||||
}
|
||||
catch (JsonException)
|
||||
{
|
||||
_logger.LogError("JsonReaderException: Unable to parse message.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,22 +1,21 @@
|
||||
using Bit.Core.Utilities;
|
||||
using Serilog.Events;
|
||||
|
||||
namespace Bit.EventsProcessor
|
||||
namespace Bit.EventsProcessor;
|
||||
|
||||
public class Program
|
||||
{
|
||||
public class Program
|
||||
public static void Main(string[] args)
|
||||
{
|
||||
public static void Main(string[] args)
|
||||
{
|
||||
Host
|
||||
.CreateDefaultBuilder(args)
|
||||
.ConfigureWebHostDefaults(webBuilder =>
|
||||
{
|
||||
webBuilder.UseStartup<Startup>();
|
||||
webBuilder.ConfigureLogging((hostingContext, logging) =>
|
||||
logging.AddSerilog(hostingContext, e => e.Level >= LogEventLevel.Warning));
|
||||
})
|
||||
.Build()
|
||||
.Run();
|
||||
}
|
||||
Host
|
||||
.CreateDefaultBuilder(args)
|
||||
.ConfigureWebHostDefaults(webBuilder =>
|
||||
{
|
||||
webBuilder.UseStartup<Startup>();
|
||||
webBuilder.ConfigureLogging((hostingContext, logging) =>
|
||||
logging.AddSerilog(hostingContext, e => e.Level >= LogEventLevel.Warning));
|
||||
})
|
||||
.Build()
|
||||
.Run();
|
||||
}
|
||||
}
|
||||
|
||||
@ -4,53 +4,52 @@ using Bit.Core.Utilities;
|
||||
using Bit.SharedWeb.Utilities;
|
||||
using Microsoft.IdentityModel.Logging;
|
||||
|
||||
namespace Bit.EventsProcessor
|
||||
namespace Bit.EventsProcessor;
|
||||
|
||||
public class Startup
|
||||
{
|
||||
public class Startup
|
||||
public Startup(IWebHostEnvironment env, IConfiguration configuration)
|
||||
{
|
||||
public Startup(IWebHostEnvironment env, IConfiguration configuration)
|
||||
CultureInfo.DefaultThreadCurrentCulture = new CultureInfo("en-US");
|
||||
Configuration = configuration;
|
||||
Environment = env;
|
||||
}
|
||||
|
||||
public IConfiguration Configuration { get; }
|
||||
public IWebHostEnvironment Environment { get; set; }
|
||||
|
||||
public void ConfigureServices(IServiceCollection services)
|
||||
{
|
||||
// Options
|
||||
services.AddOptions();
|
||||
|
||||
// Settings
|
||||
services.AddGlobalSettingsServices(Configuration, Environment);
|
||||
|
||||
// Hosted Services
|
||||
services.AddHostedService<AzureQueueHostedService>();
|
||||
}
|
||||
|
||||
public void Configure(
|
||||
IApplicationBuilder app,
|
||||
IWebHostEnvironment env,
|
||||
IHostApplicationLifetime appLifetime,
|
||||
GlobalSettings globalSettings)
|
||||
{
|
||||
IdentityModelEventSource.ShowPII = true;
|
||||
app.UseSerilog(env, appLifetime, globalSettings);
|
||||
// Add general security headers
|
||||
app.UseMiddleware<SecurityHeadersMiddleware>();
|
||||
app.UseRouting();
|
||||
app.UseEndpoints(endpoints =>
|
||||
{
|
||||
CultureInfo.DefaultThreadCurrentCulture = new CultureInfo("en-US");
|
||||
Configuration = configuration;
|
||||
Environment = env;
|
||||
}
|
||||
endpoints.MapGet("/alive",
|
||||
async context => await context.Response.WriteAsJsonAsync(System.DateTime.UtcNow));
|
||||
endpoints.MapGet("/now",
|
||||
async context => await context.Response.WriteAsJsonAsync(System.DateTime.UtcNow));
|
||||
endpoints.MapGet("/version",
|
||||
async context => await context.Response.WriteAsJsonAsync(CoreHelpers.GetVersion()));
|
||||
|
||||
public IConfiguration Configuration { get; }
|
||||
public IWebHostEnvironment Environment { get; set; }
|
||||
|
||||
public void ConfigureServices(IServiceCollection services)
|
||||
{
|
||||
// Options
|
||||
services.AddOptions();
|
||||
|
||||
// Settings
|
||||
services.AddGlobalSettingsServices(Configuration, Environment);
|
||||
|
||||
// Hosted Services
|
||||
services.AddHostedService<AzureQueueHostedService>();
|
||||
}
|
||||
|
||||
public void Configure(
|
||||
IApplicationBuilder app,
|
||||
IWebHostEnvironment env,
|
||||
IHostApplicationLifetime appLifetime,
|
||||
GlobalSettings globalSettings)
|
||||
{
|
||||
IdentityModelEventSource.ShowPII = true;
|
||||
app.UseSerilog(env, appLifetime, globalSettings);
|
||||
// Add general security headers
|
||||
app.UseMiddleware<SecurityHeadersMiddleware>();
|
||||
app.UseRouting();
|
||||
app.UseEndpoints(endpoints =>
|
||||
{
|
||||
endpoints.MapGet("/alive",
|
||||
async context => await context.Response.WriteAsJsonAsync(System.DateTime.UtcNow));
|
||||
endpoints.MapGet("/now",
|
||||
async context => await context.Response.WriteAsJsonAsync(System.DateTime.UtcNow));
|
||||
endpoints.MapGet("/version",
|
||||
async context => await context.Response.WriteAsJsonAsync(CoreHelpers.GetVersion()));
|
||||
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user