bitwarden/server
1
0
Fork 0
mirror of https://github.com/bitwarden/server synced 2025-12-21 18:53:41 +00:00
Code Issues Releases Wiki Activity

Refactor event integration service collection extensions into their own extension (#6714)

Browse Source 
* Add CQRS and caching support for OrganizationIntegrationConfigurations

* Refactor event integration service collection extensions into their own extension
This commit is contained in:
Brant DeBow
2025-12-12 16:17:43 -05:00
committed by GitHub
parent 4fdc4b1b49
commit 196e555116
8 changed files with 1178 additions and 300 deletions
Download Patch File Download Diff File Expand all files Collapse all files

505
src/Core/AdminConsole/EventIntegrations/EventIntegrationsServiceCollectionExtensions.cs
View File

@@ -1,11 +1,24 @@
using Bit.Core.AdminConsole.EventIntegrations.OrganizationIntegrationConfigurations;
using Azure.Messaging.ServiceBus;
using Bit.Core.AdminConsole.EventIntegrations.OrganizationIntegrationConfigurations;
using Bit.Core.AdminConsole.EventIntegrations.OrganizationIntegrationConfigurations.Interfaces;
using Bit.Core.AdminConsole.EventIntegrations.OrganizationIntegrations;
using Bit.Core.AdminConsole.EventIntegrations.OrganizationIntegrations.Interfaces;
using Bit.Core.AdminConsole.Models.Data.EventIntegrations;
using Bit.Core.AdminConsole.Models.Teams;
using Bit.Core.AdminConsole.Repositories;
using Bit.Core.AdminConsole.Services;
using Bit.Core.AdminConsole.Services.NoopImplementations;
using Bit.Core.Repositories;
using Bit.Core.Services;
using Bit.Core.Settings;
using Bit.Core.Utilities;
using Microsoft.Bot.Builder;
using Microsoft.Bot.Builder.Integration.AspNet.Core;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using ZiggyCreatures.Caching.Fusion;
using TableStorageRepos = Bit.Core.Repositories.TableStorage;
namespace Microsoft.Extensions.DependencyInjection;
@@ -33,6 +46,460 @@ public static class EventIntegrationsServiceCollectionExtensions
return services;
}
/// <summary>
/// Registers event write services based on available configuration.
/// </summary>
/// <param name="services">The service collection to add services to.</param>
/// <param name="globalSettings">The global settings containing event logging configuration.</param>
/// <returns>The service collection for chaining.</returns>
/// <remarks>
/// <para>
/// This method registers the appropriate IEventWriteService implementation based on the available
/// configuration, checking in the following priority order:
/// </para>
/// <para>
/// 1. Azure Service Bus - If all Azure Service Bus settings are present, registers
/// EventIntegrationEventWriteService with AzureServiceBusService as the publisher
/// </para>
/// <para>
/// 2. RabbitMQ - If all RabbitMQ settings are present, registers EventIntegrationEventWriteService with
/// RabbitMqService as the publisher
/// </para>
/// <para>
/// 3. Azure Queue Storage - If Events.ConnectionString is present, registers AzureQueueEventWriteService
/// </para>
/// <para>
/// 4. Repository (Self-Hosted) - If SelfHosted is true, registers RepositoryEventWriteService
/// </para>
/// <para>
/// 5. Noop - If none of the above are configured, registers NoopEventWriteService (no-op implementation)
/// </para>
/// </remarks>
public static IServiceCollection AddEventWriteServices(this IServiceCollection services, GlobalSettings globalSettings)
{
if (IsAzureServiceBusEnabled(globalSettings))
{
services.TryAddSingleton<IEventIntegrationPublisher, AzureServiceBusService>();
services.TryAddSingleton<IEventWriteService, EventIntegrationEventWriteService>();
return services;
}
if (IsRabbitMqEnabled(globalSettings))
{
services.TryAddSingleton<IEventIntegrationPublisher, RabbitMqService>();
services.TryAddSingleton<IEventWriteService, EventIntegrationEventWriteService>();
return services;
}
if (CoreHelpers.SettingHasValue(globalSettings.Events.ConnectionString))
{
services.TryAddSingleton<IEventWriteService, AzureQueueEventWriteService>();
return services;
}
if (globalSettings.SelfHosted)
{
services.TryAddSingleton<IEventWriteService, RepositoryEventWriteService>();
return services;
}
services.TryAddSingleton<IEventWriteService, NoopEventWriteService>();
return services;
}
/// <summary>
/// Registers Azure Service Bus-based event integration listeners and supporting infrastructure.
/// </summary>
/// <param name="services">The service collection to add services to.</param>
/// <param name="globalSettings">The global settings containing Azure Service Bus configuration.</param>
/// <returns>The service collection for chaining.</returns>
/// <remarks>
/// <para>
/// If Azure Service Bus is not enabled (missing required settings), this method returns immediately
/// without registering any services.
/// </para>
/// <para>
/// When Azure Service Bus is enabled, this method registers:
/// - IAzureServiceBusService and IEventIntegrationPublisher implementations
/// - Table Storage event repository
/// - Azure Table Storage event handler
/// - All event integration services via AddEventIntegrationServices
/// </para>
/// <para>
/// PREREQUISITE: Callers must ensure AddDistributedCache has been called before this method,
/// as it is required to create the event integrations extended cache.
/// </para>
/// </remarks>
public static IServiceCollection AddAzureServiceBusListeners(this IServiceCollection services, GlobalSettings globalSettings)
{
if (!IsAzureServiceBusEnabled(globalSettings))
{
return services;
}
services.TryAddSingleton<IAzureServiceBusService, AzureServiceBusService>();
services.TryAddSingleton<IEventIntegrationPublisher, AzureServiceBusService>();
services.TryAddSingleton<IEventRepository, TableStorageRepos.EventRepository>();
services.TryAddKeyedSingleton<IEventWriteService, RepositoryEventWriteService>("persistent");
services.TryAddSingleton<AzureTableStorageEventHandler>();
services.AddEventIntegrationServices(globalSettings);
return services;
}
/// <summary>
/// Registers RabbitMQ-based event integration listeners and supporting infrastructure.
/// </summary>
/// <param name="services">The service collection to add services to.</param>
/// <param name="globalSettings">The global settings containing RabbitMQ configuration.</param>
/// <returns>The service collection for chaining.</returns>
/// <remarks>
/// <para>
/// If RabbitMQ is not enabled (missing required settings), this method returns immediately
/// without registering any services.
/// </para>
/// <para>
/// When RabbitMQ is enabled, this method registers:
/// - IRabbitMqService and IEventIntegrationPublisher implementations
/// - Event repository handler
/// - All event integration services via AddEventIntegrationServices
/// </para>
/// <para>
/// PREREQUISITE: Callers must ensure AddDistributedCache has been called before this method,
/// as it is required to create the event integrations extended cache.
/// </para>
/// </remarks>
public static IServiceCollection AddRabbitMqListeners(this IServiceCollection services, GlobalSettings globalSettings)
{
if (!IsRabbitMqEnabled(globalSettings))
{
return services;
}
services.TryAddSingleton<IRabbitMqService, RabbitMqService>();
services.TryAddSingleton<IEventIntegrationPublisher, RabbitMqService>();
services.TryAddSingleton<EventRepositoryHandler>();
services.AddEventIntegrationServices(globalSettings);
return services;
}
/// <summary>
/// Registers Slack integration services based on configuration settings.
/// </summary>
/// <param name="services">The service collection to add services to.</param>
/// <param name="globalSettings">The global settings containing Slack configuration.</param>
/// <returns>The service collection for chaining.</returns>
/// <remarks>
/// If all required Slack settings are configured (ClientId, ClientSecret, Scopes), registers the full SlackService,
/// including an HttpClient for Slack API calls. Otherwise, registers a NoopSlackService that performs no operations.
/// </remarks>
public static IServiceCollection AddSlackService(this IServiceCollection services, GlobalSettings globalSettings)
{
if (CoreHelpers.SettingHasValue(globalSettings.Slack.ClientId) &&
CoreHelpers.SettingHasValue(globalSettings.Slack.ClientSecret) &&
CoreHelpers.SettingHasValue(globalSettings.Slack.Scopes))
{
services.AddHttpClient(SlackService.HttpClientName);
services.TryAddSingleton<ISlackService, SlackService>();
}
else
{
services.TryAddSingleton<ISlackService, NoopSlackService>();
}
return services;
}
/// <summary>
/// Registers Microsoft Teams integration services based on configuration settings.
/// </summary>
/// <param name="services">The service collection to add services to.</param>
/// <param name="globalSettings">The global settings containing Teams configuration.</param>
/// <returns>The service collection for chaining.</returns>
/// <remarks>
/// If all required Teams settings are configured (ClientId, ClientSecret, Scopes), registers:
/// - TeamsService and its interfaces (IBot, ITeamsService)
/// - IBotFrameworkHttpAdapter with Teams credentials
/// - HttpClient for Teams API calls
/// Otherwise, registers a NoopTeamsService that performs no operations.
/// </remarks>
public static IServiceCollection AddTeamsService(this IServiceCollection services, GlobalSettings globalSettings)
{
if (CoreHelpers.SettingHasValue(globalSettings.Teams.ClientId) &&
CoreHelpers.SettingHasValue(globalSettings.Teams.ClientSecret) &&
CoreHelpers.SettingHasValue(globalSettings.Teams.Scopes))
{
services.AddHttpClient(TeamsService.HttpClientName);
services.TryAddSingleton<TeamsService>();
services.TryAddSingleton<IBot>(sp => sp.GetRequiredService<TeamsService>());
services.TryAddSingleton<ITeamsService>(sp => sp.GetRequiredService<TeamsService>());
services.TryAddSingleton<IBotFrameworkHttpAdapter>(_ =>
new BotFrameworkHttpAdapter(
new TeamsBotCredentialProvider(
clientId: globalSettings.Teams.ClientId,
clientSecret: globalSettings.Teams.ClientSecret
)
)
);
}
else
{
services.TryAddSingleton<ITeamsService, NoopTeamsService>();
}
return services;
}
/// <summary>
/// Registers event integration services including handlers, listeners, and supporting infrastructure.
/// </summary>
/// <param name="services">The service collection to add services to.</param>
/// <param name="globalSettings">The global settings containing integration configuration.</param>
/// <returns>The service collection for chaining.</returns>
/// <remarks>
/// <para>
/// This method orchestrates the registration of all event integration components based on the enabled
/// message broker (Azure Service Bus or RabbitMQ). It is an internal method called by the public
/// entry points AddAzureServiceBusListeners and AddRabbitMqListeners.
/// </para>
/// <para>
/// NOTE: If both Azure Service Bus and RabbitMQ are configured, Azure Service Bus takes precedence. This means that
/// Azure Service Bus listeners will be registered (and RabbitMQ listeners will NOT) even if this event is called
/// from AddRabbitMqListeners when Azure Service Bus settings are configured.
/// </para>
/// <para>
/// PREREQUISITE: Callers must ensure AddDistributedCache has been called before invoking this method.
/// This method depends on distributed cache infrastructure being available for the keyed extended
/// cache registration.
/// </para>
/// <para>
/// Registered Services:
/// - Keyed ExtendedCache for event integrations
/// - Integration filter service
/// - Integration handlers for Slack, Webhook, Hec, Datadog, and Teams
/// - Hosted services for event and integration listeners (based on enabled message broker)
/// </para>
/// </remarks>
internal static IServiceCollection AddEventIntegrationServices(this IServiceCollection services,
GlobalSettings globalSettings)
{
// Add common services
// NOTE: AddDistributedCache must be called by the caller before this method
services.AddExtendedCache(EventIntegrationsCacheConstants.CacheName, globalSettings);
services.TryAddSingleton<IIntegrationFilterService, IntegrationFilterService>();
services.TryAddKeyedSingleton<IEventWriteService, RepositoryEventWriteService>("persistent");
// Add services in support of handlers
services.AddSlackService(globalSettings);
services.AddTeamsService(globalSettings);
services.TryAddSingleton(TimeProvider.System);
services.AddHttpClient(WebhookIntegrationHandler.HttpClientName);
services.AddHttpClient(DatadogIntegrationHandler.HttpClientName);
// Add integration handlers
services.TryAddSingleton<IIntegrationHandler<SlackIntegrationConfigurationDetails>, SlackIntegrationHandler>();
services.TryAddSingleton<IIntegrationHandler<WebhookIntegrationConfigurationDetails>, WebhookIntegrationHandler>();
services.TryAddSingleton<IIntegrationHandler<DatadogIntegrationConfigurationDetails>, DatadogIntegrationHandler>();
services.TryAddSingleton<IIntegrationHandler<TeamsIntegrationConfigurationDetails>, TeamsIntegrationHandler>();
var repositoryConfiguration = new RepositoryListenerConfiguration(globalSettings);
var slackConfiguration = new SlackListenerConfiguration(globalSettings);
var webhookConfiguration = new WebhookListenerConfiguration(globalSettings);
var hecConfiguration = new HecListenerConfiguration(globalSettings);
var datadogConfiguration = new DatadogListenerConfiguration(globalSettings);
var teamsConfiguration = new TeamsListenerConfiguration(globalSettings);
if (IsAzureServiceBusEnabled(globalSettings))
{
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService,
AzureServiceBusEventListenerService<RepositoryListenerConfiguration>>(provider =>
new AzureServiceBusEventListenerService<RepositoryListenerConfiguration>(
configuration: repositoryConfiguration,
handler: provider.GetRequiredService<AzureTableStorageEventHandler>(),
serviceBusService: provider.GetRequiredService<IAzureServiceBusService>(),
serviceBusOptions: new ServiceBusProcessorOptions()
{
PrefetchCount = repositoryConfiguration.EventPrefetchCount,
MaxConcurrentCalls = repositoryConfiguration.EventMaxConcurrentCalls
},
loggerFactory: provider.GetRequiredService<ILoggerFactory>()
)
)
);
services.AddAzureServiceBusIntegration<SlackIntegrationConfigurationDetails, SlackListenerConfiguration>(slackConfiguration);
services.AddAzureServiceBusIntegration<WebhookIntegrationConfigurationDetails, WebhookListenerConfiguration>(webhookConfiguration);
services.AddAzureServiceBusIntegration<WebhookIntegrationConfigurationDetails, HecListenerConfiguration>(hecConfiguration);
services.AddAzureServiceBusIntegration<DatadogIntegrationConfigurationDetails, DatadogListenerConfiguration>(datadogConfiguration);
services.AddAzureServiceBusIntegration<TeamsIntegrationConfigurationDetails, TeamsListenerConfiguration>(teamsConfiguration);
return services;
}
if (IsRabbitMqEnabled(globalSettings))
{
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService,
RabbitMqEventListenerService<RepositoryListenerConfiguration>>(provider =>
new RabbitMqEventListenerService<RepositoryListenerConfiguration>(
handler: provider.GetRequiredService<EventRepositoryHandler>(),
configuration: repositoryConfiguration,
rabbitMqService: provider.GetRequiredService<IRabbitMqService>(),
loggerFactory: provider.GetRequiredService<ILoggerFactory>()
)
)
);
services.AddRabbitMqIntegration<SlackIntegrationConfigurationDetails, SlackListenerConfiguration>(slackConfiguration);
services.AddRabbitMqIntegration<WebhookIntegrationConfigurationDetails, WebhookListenerConfiguration>(webhookConfiguration);
services.AddRabbitMqIntegration<WebhookIntegrationConfigurationDetails, HecListenerConfiguration>(hecConfiguration);
services.AddRabbitMqIntegration<DatadogIntegrationConfigurationDetails, DatadogListenerConfiguration>(datadogConfiguration);
services.AddRabbitMqIntegration<TeamsIntegrationConfigurationDetails, TeamsListenerConfiguration>(teamsConfiguration);
}
return services;
}
/// <summary>
/// Registers Azure Service Bus-based event integration listeners for a specific integration type.
/// </summary>
/// <typeparam name="TConfig">The integration configuration details type (e.g., SlackIntegrationConfigurationDetails).</typeparam>
/// <typeparam name="TListenerConfig">The listener configuration type implementing IIntegrationListenerConfiguration.</typeparam>
/// <param name="services">The service collection to add services to.</param>
/// <param name="listenerConfiguration">The listener configuration containing routing keys and message processing settings.</param>
/// <returns>The service collection for chaining.</returns>
/// <remarks>
/// <para>
/// This method registers three key components:
/// 1. EventIntegrationHandler - Keyed singleton for processing integration events
/// 2. AzureServiceBusEventListenerService - Hosted service for listening to event messages from Azure Service Bus
/// for this integration type
/// 3. AzureServiceBusIntegrationListenerService - Hosted service for listening to integration messages from
/// Azure Service Bus for this integration type
/// </para>
/// <para>
/// The handler uses the listener configuration's routing key as its service key, allowing multiple
/// handlers to be registered for different integration types.
/// </para>
/// <para>
/// Service Bus processor options (PrefetchCount and MaxConcurrentCalls) are configured from the listener
/// configuration to optimize message throughput and concurrency.
/// </para>
/// </remarks>
internal static IServiceCollection AddAzureServiceBusIntegration<TConfig, TListenerConfig>(this IServiceCollection services,
TListenerConfig listenerConfiguration)
where TConfig : class
where TListenerConfig : IIntegrationListenerConfiguration
{
services.TryAddKeyedSingleton<IEventMessageHandler>(serviceKey: listenerConfiguration.RoutingKey, implementationFactory: (provider, _) =>
new EventIntegrationHandler<TConfig>(
integrationType: listenerConfiguration.IntegrationType,
eventIntegrationPublisher: provider.GetRequiredService<IEventIntegrationPublisher>(),
integrationFilterService: provider.GetRequiredService<IIntegrationFilterService>(),
cache: provider.GetRequiredKeyedService<IFusionCache>(EventIntegrationsCacheConstants.CacheName),
configurationRepository: provider.GetRequiredService<IOrganizationIntegrationConfigurationRepository>(),
groupRepository: provider.GetRequiredService<IGroupRepository>(),
organizationRepository: provider.GetRequiredService<IOrganizationRepository>(),
organizationUserRepository: provider.GetRequiredService<IOrganizationUserRepository>(), logger: provider.GetRequiredService<ILogger<EventIntegrationHandler<TConfig>>>())
);
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService,
AzureServiceBusEventListenerService<TListenerConfig>>(provider =>
new AzureServiceBusEventListenerService<TListenerConfig>(
configuration: listenerConfiguration,
handler: provider.GetRequiredKeyedService<IEventMessageHandler>(serviceKey: listenerConfiguration.RoutingKey),
serviceBusService: provider.GetRequiredService<IAzureServiceBusService>(),
serviceBusOptions: new ServiceBusProcessorOptions()
{
PrefetchCount = listenerConfiguration.EventPrefetchCount,
MaxConcurrentCalls = listenerConfiguration.EventMaxConcurrentCalls
},
loggerFactory: provider.GetRequiredService<ILoggerFactory>()
)
)
);
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService,
AzureServiceBusIntegrationListenerService<TListenerConfig>>(provider =>
new AzureServiceBusIntegrationListenerService<TListenerConfig>(
configuration: listenerConfiguration,
handler: provider.GetRequiredService<IIntegrationHandler<TConfig>>(),
serviceBusService: provider.GetRequiredService<IAzureServiceBusService>(),
serviceBusOptions: new ServiceBusProcessorOptions()
{
PrefetchCount = listenerConfiguration.IntegrationPrefetchCount,
MaxConcurrentCalls = listenerConfiguration.IntegrationMaxConcurrentCalls
},
loggerFactory: provider.GetRequiredService<ILoggerFactory>()
)
)
);
return services;
}
/// <summary>
/// Registers RabbitMQ-based event integration listeners for a specific integration type.
/// </summary>
/// <typeparam name="TConfig">The integration configuration details type (e.g., SlackIntegrationConfigurationDetails).</typeparam>
/// <typeparam name="TListenerConfig">The listener configuration type implementing IIntegrationListenerConfiguration.</typeparam>
/// <param name="services">The service collection to add services to.</param>
/// <param name="listenerConfiguration">The listener configuration containing routing keys and message processing settings.</param>
/// <returns>The service collection for chaining.</returns>
/// <remarks>
/// <para>
/// This method registers three key components:
/// 1. EventIntegrationHandler - Keyed singleton for processing integration events
/// 2. RabbitMqEventListenerService - Hosted service for listening to event messages from RabbitMQ for
/// this integration type
/// 3. RabbitMqIntegrationListenerService - Hosted service for listening to integration messages from RabbitMQ for
/// this integration type
/// </para>
///
/// <para>
/// The handler uses the listener configuration's routing key as its service key, allowing multiple
/// handlers to be registered for different integration types.
/// </para>
/// </remarks>
internal static IServiceCollection AddRabbitMqIntegration<TConfig, TListenerConfig>(this IServiceCollection services,
TListenerConfig listenerConfiguration)
where TConfig : class
where TListenerConfig : IIntegrationListenerConfiguration
{
services.TryAddKeyedSingleton<IEventMessageHandler>(serviceKey: listenerConfiguration.RoutingKey, implementationFactory: (provider, _) =>
new EventIntegrationHandler<TConfig>(
integrationType: listenerConfiguration.IntegrationType,
eventIntegrationPublisher: provider.GetRequiredService<IEventIntegrationPublisher>(),
integrationFilterService: provider.GetRequiredService<IIntegrationFilterService>(),
cache: provider.GetRequiredKeyedService<IFusionCache>(EventIntegrationsCacheConstants.CacheName),
configurationRepository: provider.GetRequiredService<IOrganizationIntegrationConfigurationRepository>(),
groupRepository: provider.GetRequiredService<IGroupRepository>(),
organizationRepository: provider.GetRequiredService<IOrganizationRepository>(),
organizationUserRepository: provider.GetRequiredService<IOrganizationUserRepository>(), logger: provider.GetRequiredService<ILogger<EventIntegrationHandler<TConfig>>>())
);
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService,
RabbitMqEventListenerService<TListenerConfig>>(provider =>
new RabbitMqEventListenerService<TListenerConfig>(
handler: provider.GetRequiredKeyedService<IEventMessageHandler>(serviceKey: listenerConfiguration.RoutingKey),
configuration: listenerConfiguration,
rabbitMqService: provider.GetRequiredService<IRabbitMqService>(),
loggerFactory: provider.GetRequiredService<ILoggerFactory>()
)
)
);
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService,
RabbitMqIntegrationListenerService<TListenerConfig>>(provider =>
new RabbitMqIntegrationListenerService<TListenerConfig>(
handler: provider.GetRequiredService<IIntegrationHandler<TConfig>>(),
configuration: listenerConfiguration,
rabbitMqService: provider.GetRequiredService<IRabbitMqService>(),
loggerFactory: provider.GetRequiredService<ILoggerFactory>(),
timeProvider: provider.GetRequiredService<TimeProvider>()
)
)
);
return services;
}
internal static IServiceCollection AddOrganizationIntegrationCommandsQueries(this IServiceCollection services)
{
services.TryAddScoped<ICreateOrganizationIntegrationCommand, CreateOrganizationIntegrationCommand>();
@@ -52,4 +519,40 @@ public static class EventIntegrationsServiceCollectionExtensions
return services;
}
/// <summary>
/// Determines if RabbitMQ is enabled for event integrations based on configuration settings.
/// </summary>
/// <param name="settings">The global settings containing RabbitMQ configuration.</param>
/// <returns>True if all required RabbitMQ settings are present; otherwise, false.</returns>
/// <remarks>
/// Requires all the following settings to be configured:
/// - EventLogging.RabbitMq.HostName
/// - EventLogging.RabbitMq.Username
/// - EventLogging.RabbitMq.Password
/// - EventLogging.RabbitMq.EventExchangeName
/// </remarks>
internal static bool IsRabbitMqEnabled(GlobalSettings settings)
{
return CoreHelpers.SettingHasValue(settings.EventLogging.RabbitMq.HostName) &&
CoreHelpers.SettingHasValue(settings.EventLogging.RabbitMq.Username) &&
CoreHelpers.SettingHasValue(settings.EventLogging.RabbitMq.Password) &&
CoreHelpers.SettingHasValue(settings.EventLogging.RabbitMq.EventExchangeName);
}
/// <summary>
/// Determines if Azure Service Bus is enabled for event integrations based on configuration settings.
/// </summary>
/// <param name="settings">The global settings containing Azure Service Bus configuration.</param>
/// <returns>True if all required Azure Service Bus settings are present; otherwise, false.</returns>
/// <remarks>
/// Requires both of the following settings to be configured:
/// - EventLogging.AzureServiceBus.ConnectionString
/// - EventLogging.AzureServiceBus.EventTopicName
/// </remarks>
internal static bool IsAzureServiceBusEnabled(GlobalSettings settings)
{
return CoreHelpers.SettingHasValue(settings.EventLogging.AzureServiceBus.ConnectionString) &&
CoreHelpers.SettingHasValue(settings.EventLogging.AzureServiceBus.EventTopicName);
}
}

1
src/Core/Core.csproj
View File

@@ -41,6 +41,7 @@
<PackageReference Include="Microsoft.Azure.Cosmos" Version="3.52.0" />
<PackageReference Include="Microsoft.Azure.NotificationHubs" Version="4.2.0" />
<PackageReference Include="Microsoft.Bot.Builder" Version="4.23.0" />
<PackageReference Include="Microsoft.Bot.Builder.Integration.AspNet.Core" Version="4.23.0" />
<PackageReference Include="Microsoft.Bot.Connector" Version="4.23.0" />
<PackageReference Include="Microsoft.Data.SqlClient" Version="5.2.2" />
<PackageReference Include="Microsoft.Extensions.Caching.Cosmos" Version="1.7.0" />

2
src/Core/Utilities/CACHING.md
View File

@@ -381,7 +381,7 @@ public class OrganizationAbilityService
### Example Usage: Default (Ephemeral Data)
#### 1. Registration (already done in Api, Admin, Billing, Identity, and Notifications Startup.cs files, plus Events and EventsProcessor service collection extensions):
#### 1. Registration (already done in Api, Admin, Billing, Events, EventsProcessor, Identity, and Notifications Startup.cs files):
```csharp
services.AddDistributedCache(globalSettings);

2
src/Events/Startup.cs
View File

@@ -84,6 +84,8 @@ public class Startup
services.AddHostedService<Core.HostedServices.ApplicationCacheHostedService>();
}
// Add event integration services
services.AddDistributedCache(globalSettings);
services.AddRabbitMqListeners(globalSettings);
}

3
src/EventsProcessor/Startup.cs
View File

@@ -31,7 +31,8 @@ public class Startup
// Repositories
services.AddDatabaseRepositories(globalSettings);
// Hosted Services
// Add event integration services
services.AddDistributedCache(globalSettings);
services.AddAzureServiceBusListeners(globalSettings);
services.AddHostedService<AzureQueueHostedService>();
}

297
src/SharedWeb/Utilities/ServiceCollectionExtensions.cs
View File

@@ -6,14 +6,10 @@ using System.Reflection;
using System.Security.Claims;
using System.Security.Cryptography.X509Certificates;
using AspNetCoreRateLimit;
using Azure.Messaging.ServiceBus;
using Bit.Core;
using Bit.Core.AdminConsole.AbilitiesCache;
using Bit.Core.AdminConsole.Models.Business.Tokenables;
using Bit.Core.AdminConsole.Models.Data.EventIntegrations;
using Bit.Core.AdminConsole.Models.Teams;
using Bit.Core.AdminConsole.OrganizationFeatures.Policies;
using Bit.Core.AdminConsole.Repositories;
using Bit.Core.AdminConsole.Services;
using Bit.Core.AdminConsole.Services.Implementations;
using Bit.Core.AdminConsole.Services.NoopImplementations;
@@ -74,8 +70,6 @@ using Microsoft.AspNetCore.HttpOverrides;
using Microsoft.AspNetCore.Identity;
using Microsoft.AspNetCore.Mvc.Localization;
using Microsoft.Azure.Cosmos.Fluent;
using Microsoft.Bot.Builder;
using Microsoft.Bot.Builder.Integration.AspNet.Core;
using Microsoft.Extensions.Caching.Cosmos;
using Microsoft.Extensions.Caching.Distributed;
using Microsoft.Extensions.Configuration;
@@ -87,7 +81,6 @@ using Microsoft.Extensions.Options;
using Microsoft.OpenApi.Models;
using StackExchange.Redis;
using Swashbuckle.AspNetCore.SwaggerGen;
using ZiggyCreatures.Caching.Fusion;
using NoopRepos = Bit.Core.Repositories.Noop;
using Role = Bit.Core.Entities.Role;
using TableStorageRepos = Bit.Core.Repositories.TableStorage;
@@ -525,116 +518,6 @@ public static class ServiceCollectionExtensions
return globalSettings;
}
public static IServiceCollection AddEventWriteServices(this IServiceCollection services, GlobalSettings globalSettings)
{
if (IsAzureServiceBusEnabled(globalSettings))
{
services.TryAddSingleton<IEventIntegrationPublisher, AzureServiceBusService>();
services.TryAddSingleton<IEventWriteService, EventIntegrationEventWriteService>();
return services;
}
if (IsRabbitMqEnabled(globalSettings))
{
services.TryAddSingleton<IEventIntegrationPublisher, RabbitMqService>();
services.TryAddSingleton<IEventWriteService, EventIntegrationEventWriteService>();
return services;
}
if (CoreHelpers.SettingHasValue(globalSettings.Events.ConnectionString))
{
services.TryAddSingleton<IEventWriteService, AzureQueueEventWriteService>();
return services;
}
if (globalSettings.SelfHosted)
{
services.TryAddSingleton<IEventWriteService, RepositoryEventWriteService>();
return services;
}
services.TryAddSingleton<IEventWriteService, NoopEventWriteService>();
return services;
}
public static IServiceCollection AddAzureServiceBusListeners(this IServiceCollection services, GlobalSettings globalSettings)
{
if (!IsAzureServiceBusEnabled(globalSettings))
{
return services;
}
services.TryAddSingleton<IAzureServiceBusService, AzureServiceBusService>();
services.TryAddSingleton<IEventIntegrationPublisher, AzureServiceBusService>();
services.TryAddSingleton<IEventRepository, TableStorageRepos.EventRepository>();
services.TryAddKeyedSingleton<IEventWriteService, RepositoryEventWriteService>("persistent");
services.TryAddSingleton<AzureTableStorageEventHandler>();
services.AddEventIntegrationServices(globalSettings);
return services;
}
public static IServiceCollection AddRabbitMqListeners(this IServiceCollection services, GlobalSettings globalSettings)
{
if (!IsRabbitMqEnabled(globalSettings))
{
return services;
}
services.TryAddSingleton<IRabbitMqService, RabbitMqService>();
services.TryAddSingleton<IEventIntegrationPublisher, RabbitMqService>();
services.TryAddSingleton<EventRepositoryHandler>();
services.AddEventIntegrationServices(globalSettings);
return services;
}
public static IServiceCollection AddSlackService(this IServiceCollection services, GlobalSettings globalSettings)
{
if (CoreHelpers.SettingHasValue(globalSettings.Slack.ClientId) &&
CoreHelpers.SettingHasValue(globalSettings.Slack.ClientSecret) &&
CoreHelpers.SettingHasValue(globalSettings.Slack.Scopes))
{
services.AddHttpClient(SlackService.HttpClientName);
services.TryAddSingleton<ISlackService, SlackService>();
}
else
{
services.TryAddSingleton<ISlackService, NoopSlackService>();
}
return services;
}
public static IServiceCollection AddTeamsService(this IServiceCollection services, GlobalSettings globalSettings)
{
if (CoreHelpers.SettingHasValue(globalSettings.Teams.ClientId) &&
CoreHelpers.SettingHasValue(globalSettings.Teams.ClientSecret) &&
CoreHelpers.SettingHasValue(globalSettings.Teams.Scopes))
{
services.AddHttpClient(TeamsService.HttpClientName);
services.TryAddSingleton<TeamsService>();
services.TryAddSingleton<IBot>(sp => sp.GetRequiredService<TeamsService>());
services.TryAddSingleton<ITeamsService>(sp => sp.GetRequiredService<TeamsService>());
services.TryAddSingleton<IBotFrameworkHttpAdapter>(sp =>
new BotFrameworkHttpAdapter(
new TeamsBotCredentialProvider(
clientId: globalSettings.Teams.ClientId,
clientSecret: globalSettings.Teams.ClientSecret
)
)
);
}
else
{
services.TryAddSingleton<ITeamsService, NoopTeamsService>();
}
return services;
}
public static void UseDefaultMiddleware(this IApplicationBuilder app,
IWebHostEnvironment env, GlobalSettings globalSettings)
{
@@ -881,186 +764,6 @@ public static class ServiceCollectionExtensions
return (provider, connectionString);
}
private static IServiceCollection AddAzureServiceBusIntegration<TConfig, TListenerConfig>(this IServiceCollection services,
TListenerConfig listenerConfiguration)
where TConfig : class
where TListenerConfig : IIntegrationListenerConfiguration
{
services.TryAddKeyedSingleton<IEventMessageHandler>(serviceKey: listenerConfiguration.RoutingKey, implementationFactory: (provider, _) =>
new EventIntegrationHandler<TConfig>(
integrationType: listenerConfiguration.IntegrationType,
eventIntegrationPublisher: provider.GetRequiredService<IEventIntegrationPublisher>(),
integrationFilterService: provider.GetRequiredService<IIntegrationFilterService>(),
cache: provider.GetRequiredKeyedService<IFusionCache>(EventIntegrationsCacheConstants.CacheName),
configurationRepository: provider.GetRequiredService<IOrganizationIntegrationConfigurationRepository>(),
groupRepository: provider.GetRequiredService<IGroupRepository>(),
organizationRepository: provider.GetRequiredService<IOrganizationRepository>(),
organizationUserRepository: provider.GetRequiredService<IOrganizationUserRepository>(), logger: provider.GetRequiredService<ILogger<EventIntegrationHandler<TConfig>>>())
);
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService,
AzureServiceBusEventListenerService<TListenerConfig>>(provider =>
new AzureServiceBusEventListenerService<TListenerConfig>(
configuration: listenerConfiguration,
handler: provider.GetRequiredKeyedService<IEventMessageHandler>(serviceKey: listenerConfiguration.RoutingKey),
serviceBusService: provider.GetRequiredService<IAzureServiceBusService>(),
serviceBusOptions: new ServiceBusProcessorOptions()
{
PrefetchCount = listenerConfiguration.EventPrefetchCount,
MaxConcurrentCalls = listenerConfiguration.EventMaxConcurrentCalls
},
loggerFactory: provider.GetRequiredService<ILoggerFactory>()
)
)
);
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService,
AzureServiceBusIntegrationListenerService<TListenerConfig>>(provider =>
new AzureServiceBusIntegrationListenerService<TListenerConfig>(
configuration: listenerConfiguration,
handler: provider.GetRequiredService<IIntegrationHandler<TConfig>>(),
serviceBusService: provider.GetRequiredService<IAzureServiceBusService>(),
serviceBusOptions: new ServiceBusProcessorOptions()
{
PrefetchCount = listenerConfiguration.IntegrationPrefetchCount,
MaxConcurrentCalls = listenerConfiguration.IntegrationMaxConcurrentCalls
},
loggerFactory: provider.GetRequiredService<ILoggerFactory>()
)
)
);
return services;
}
private static IServiceCollection AddEventIntegrationServices(this IServiceCollection services,
GlobalSettings globalSettings)
{
// Add common services
services.AddDistributedCache(globalSettings);
services.AddExtendedCache(EventIntegrationsCacheConstants.CacheName, globalSettings);
services.TryAddSingleton<IIntegrationFilterService, IntegrationFilterService>();
services.TryAddKeyedSingleton<IEventWriteService, RepositoryEventWriteService>("persistent");
// Add services in support of handlers
services.AddSlackService(globalSettings);
services.AddTeamsService(globalSettings);
services.TryAddSingleton(TimeProvider.System);
services.AddHttpClient(WebhookIntegrationHandler.HttpClientName);
services.AddHttpClient(DatadogIntegrationHandler.HttpClientName);
// Add integration handlers
services.TryAddSingleton<IIntegrationHandler<SlackIntegrationConfigurationDetails>, SlackIntegrationHandler>();
services.TryAddSingleton<IIntegrationHandler<WebhookIntegrationConfigurationDetails>, WebhookIntegrationHandler>();
services.TryAddSingleton<IIntegrationHandler<DatadogIntegrationConfigurationDetails>, DatadogIntegrationHandler>();
services.TryAddSingleton<IIntegrationHandler<TeamsIntegrationConfigurationDetails>, TeamsIntegrationHandler>();
var repositoryConfiguration = new RepositoryListenerConfiguration(globalSettings);
var slackConfiguration = new SlackListenerConfiguration(globalSettings);
var webhookConfiguration = new WebhookListenerConfiguration(globalSettings);
var hecConfiguration = new HecListenerConfiguration(globalSettings);
var datadogConfiguration = new DatadogListenerConfiguration(globalSettings);
var teamsConfiguration = new TeamsListenerConfiguration(globalSettings);
if (IsRabbitMqEnabled(globalSettings))
{
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService,
RabbitMqEventListenerService<RepositoryListenerConfiguration>>(provider =>
new RabbitMqEventListenerService<RepositoryListenerConfiguration>(
handler: provider.GetRequiredService<EventRepositoryHandler>(),
configuration: repositoryConfiguration,
rabbitMqService: provider.GetRequiredService<IRabbitMqService>(),
loggerFactory: provider.GetRequiredService<ILoggerFactory>()
)
)
);
services.AddRabbitMqIntegration<SlackIntegrationConfigurationDetails, SlackListenerConfiguration>(slackConfiguration);
services.AddRabbitMqIntegration<WebhookIntegrationConfigurationDetails, WebhookListenerConfiguration>(webhookConfiguration);
services.AddRabbitMqIntegration<WebhookIntegrationConfigurationDetails, HecListenerConfiguration>(hecConfiguration);
services.AddRabbitMqIntegration<DatadogIntegrationConfigurationDetails, DatadogListenerConfiguration>(datadogConfiguration);
services.AddRabbitMqIntegration<TeamsIntegrationConfigurationDetails, TeamsListenerConfiguration>(teamsConfiguration);
}
if (IsAzureServiceBusEnabled(globalSettings))
{
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService,
AzureServiceBusEventListenerService<RepositoryListenerConfiguration>>(provider =>
new AzureServiceBusEventListenerService<RepositoryListenerConfiguration>(
configuration: repositoryConfiguration,
handler: provider.GetRequiredService<AzureTableStorageEventHandler>(),
serviceBusService: provider.GetRequiredService<IAzureServiceBusService>(),
serviceBusOptions: new ServiceBusProcessorOptions()
{
PrefetchCount = repositoryConfiguration.EventPrefetchCount,
MaxConcurrentCalls = repositoryConfiguration.EventMaxConcurrentCalls
},
loggerFactory: provider.GetRequiredService<ILoggerFactory>()
)
)
);
services.AddAzureServiceBusIntegration<SlackIntegrationConfigurationDetails, SlackListenerConfiguration>(slackConfiguration);
services.AddAzureServiceBusIntegration<WebhookIntegrationConfigurationDetails, WebhookListenerConfiguration>(webhookConfiguration);
services.AddAzureServiceBusIntegration<WebhookIntegrationConfigurationDetails, HecListenerConfiguration>(hecConfiguration);
services.AddAzureServiceBusIntegration<DatadogIntegrationConfigurationDetails, DatadogListenerConfiguration>(datadogConfiguration);
services.AddAzureServiceBusIntegration<TeamsIntegrationConfigurationDetails, TeamsListenerConfiguration>(teamsConfiguration);
}
return services;
}
private static IServiceCollection AddRabbitMqIntegration<TConfig, TListenerConfig>(this IServiceCollection services,
TListenerConfig listenerConfiguration)
where TConfig : class
where TListenerConfig : IIntegrationListenerConfiguration
{
services.TryAddKeyedSingleton<IEventMessageHandler>(serviceKey: listenerConfiguration.RoutingKey, implementationFactory: (provider, _) =>
new EventIntegrationHandler<TConfig>(
integrationType: listenerConfiguration.IntegrationType,
eventIntegrationPublisher: provider.GetRequiredService<IEventIntegrationPublisher>(),
integrationFilterService: provider.GetRequiredService<IIntegrationFilterService>(),
cache: provider.GetRequiredKeyedService<IFusionCache>(EventIntegrationsCacheConstants.CacheName),
configurationRepository: provider.GetRequiredService<IOrganizationIntegrationConfigurationRepository>(),
groupRepository: provider.GetRequiredService<IGroupRepository>(),
organizationRepository: provider.GetRequiredService<IOrganizationRepository>(),
organizationUserRepository: provider.GetRequiredService<IOrganizationUserRepository>(), logger: provider.GetRequiredService<ILogger<EventIntegrationHandler<TConfig>>>())
);
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService,
RabbitMqEventListenerService<TListenerConfig>>(provider =>
new RabbitMqEventListenerService<TListenerConfig>(
handler: provider.GetRequiredKeyedService<IEventMessageHandler>(serviceKey: listenerConfiguration.RoutingKey),
configuration: listenerConfiguration,
rabbitMqService: provider.GetRequiredService<IRabbitMqService>(),
loggerFactory: provider.GetRequiredService<ILoggerFactory>()
)
)
);
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService,
RabbitMqIntegrationListenerService<TListenerConfig>>(provider =>
new RabbitMqIntegrationListenerService<TListenerConfig>(
handler: provider.GetRequiredService<IIntegrationHandler<TConfig>>(),
configuration: listenerConfiguration,
rabbitMqService: provider.GetRequiredService<IRabbitMqService>(),
loggerFactory: provider.GetRequiredService<ILoggerFactory>(),
timeProvider: provider.GetRequiredService<TimeProvider>()
)
)
);
return services;
}
private static bool IsAzureServiceBusEnabled(GlobalSettings settings)
{
return CoreHelpers.SettingHasValue(settings.EventLogging.AzureServiceBus.ConnectionString) &&
CoreHelpers.SettingHasValue(settings.EventLogging.AzureServiceBus.EventTopicName);
}
private static bool IsRabbitMqEnabled(GlobalSettings settings)
{
return CoreHelpers.SettingHasValue(settings.EventLogging.RabbitMq.HostName) &&
CoreHelpers.SettingHasValue(settings.EventLogging.RabbitMq.Username) &&
CoreHelpers.SettingHasValue(settings.EventLogging.RabbitMq.Password) &&
CoreHelpers.SettingHasValue(settings.EventLogging.RabbitMq.EventExchangeName);
}
/// <summary>
/// Adds a server with its corresponding OAuth2 client credentials security definition and requirement.
/// </summary>

667
test/Core.Test/AdminConsole/EventIntegrations/EventIntegrationServiceCollectionExtensionsTests.cs
View File

@@ -1,12 +1,19 @@
using Bit.Core.AdminConsole.EventIntegrations.OrganizationIntegrationConfigurations.Interfaces;
using Bit.Core.AdminConsole.EventIntegrations.OrganizationIntegrations.Interfaces;
using Bit.Core.AdminConsole.Models.Data.EventIntegrations;
using Bit.Core.AdminConsole.Repositories;
using Bit.Core.AdminConsole.Services;
using Bit.Core.AdminConsole.Services.NoopImplementations;
using Bit.Core.Repositories;
using Bit.Core.Services;
using Bit.Core.Settings;
using Bit.Core.Utilities;
using Microsoft.Bot.Builder;
using Microsoft.Bot.Builder.Integration.AspNet.Core;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Hosting;
using NSubstitute;
using StackExchange.Redis;
using Xunit;
@@ -185,6 +192,666 @@ public class EventIntegrationServiceCollectionExtensionsTests
Assert.Single(createCmdDescriptors);
}
[Fact]
public void IsRabbitMqEnabled_AllSettingsPresent_ReturnsTrue()
{
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:RabbitMq:HostName"] = "localhost",
["GlobalSettings:EventLogging:RabbitMq:Username"] = "user",
["GlobalSettings:EventLogging:RabbitMq:Password"] = "pass",
["GlobalSettings:EventLogging:RabbitMq:EventExchangeName"] = "exchange"
});
Assert.True(EventIntegrationsServiceCollectionExtensions.IsRabbitMqEnabled(globalSettings));
}
[Fact]
public void IsRabbitMqEnabled_MissingHostName_ReturnsFalse()
{
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:RabbitMq:HostName"] = null,
["GlobalSettings:EventLogging:RabbitMq:Username"] = "user",
["GlobalSettings:EventLogging:RabbitMq:Password"] = "pass",
["GlobalSettings:EventLogging:RabbitMq:EventExchangeName"] = "exchange"
});
Assert.False(EventIntegrationsServiceCollectionExtensions.IsRabbitMqEnabled(globalSettings));
}
[Fact]
public void IsRabbitMqEnabled_MissingUsername_ReturnsFalse()
{
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:RabbitMq:HostName"] = "localhost",
["GlobalSettings:EventLogging:RabbitMq:Username"] = null,
["GlobalSettings:EventLogging:RabbitMq:Password"] = "pass",
["GlobalSettings:EventLogging:RabbitMq:EventExchangeName"] = "exchange"
});
Assert.False(EventIntegrationsServiceCollectionExtensions.IsRabbitMqEnabled(globalSettings));
}
[Fact]
public void IsRabbitMqEnabled_MissingPassword_ReturnsFalse()
{
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:RabbitMq:HostName"] = "localhost",
["GlobalSettings:EventLogging:RabbitMq:Username"] = "user",
["GlobalSettings:EventLogging:RabbitMq:Password"] = null,
["GlobalSettings:EventLogging:RabbitMq:EventExchangeName"] = "exchange"
});
Assert.False(EventIntegrationsServiceCollectionExtensions.IsRabbitMqEnabled(globalSettings));
}
[Fact]
public void IsRabbitMqEnabled_MissingExchangeName_ReturnsFalse()
{
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:RabbitMq:HostName"] = "localhost",
["GlobalSettings:EventLogging:RabbitMq:Username"] = "user",
["GlobalSettings:EventLogging:RabbitMq:Password"] = "pass",
["GlobalSettings:EventLogging:RabbitMq:EventExchangeName"] = null
});
Assert.False(EventIntegrationsServiceCollectionExtensions.IsRabbitMqEnabled(globalSettings));
}
[Fact]
public void IsAzureServiceBusEnabled_AllSettingsPresent_ReturnsTrue()
{
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:AzureServiceBus:ConnectionString"] = "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKeyName=test;SharedAccessKey=test",
["GlobalSettings:EventLogging:AzureServiceBus:EventTopicName"] = "events"
});
Assert.True(EventIntegrationsServiceCollectionExtensions.IsAzureServiceBusEnabled(globalSettings));
}
[Fact]
public void IsAzureServiceBusEnabled_MissingConnectionString_ReturnsFalse()
{
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:AzureServiceBus:ConnectionString"] = null,
["GlobalSettings:EventLogging:AzureServiceBus:EventTopicName"] = "events"
});
Assert.False(EventIntegrationsServiceCollectionExtensions.IsAzureServiceBusEnabled(globalSettings));
}
[Fact]
public void IsAzureServiceBusEnabled_MissingTopicName_ReturnsFalse()
{
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:AzureServiceBus:ConnectionString"] = "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKeyName=test;SharedAccessKey=test",
["GlobalSettings:EventLogging:AzureServiceBus:EventTopicName"] = null
});
Assert.False(EventIntegrationsServiceCollectionExtensions.IsAzureServiceBusEnabled(globalSettings));
}
[Fact]
public void AddSlackService_AllSettingsPresent_RegistersSlackService()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:Slack:ClientId"] = "test-client-id",
["GlobalSettings:Slack:ClientSecret"] = "test-client-secret",
["GlobalSettings:Slack:Scopes"] = "test-scopes"
});
services.TryAddSingleton(globalSettings);
services.AddLogging();
services.AddSlackService(globalSettings);
var provider = services.BuildServiceProvider();
var slackService = provider.GetService<ISlackService>();
Assert.NotNull(slackService);
Assert.IsType<SlackService>(slackService);
var httpClientDescriptor = services.FirstOrDefault(s =>
s.ServiceType == typeof(IHttpClientFactory));
Assert.NotNull(httpClientDescriptor);
}
[Fact]
public void AddSlackService_SettingsMissing_RegistersNoopService()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:Slack:ClientId"] = null,
["GlobalSettings:Slack:ClientSecret"] = null,
["GlobalSettings:Slack:Scopes"] = null
});
services.AddSlackService(globalSettings);
var provider = services.BuildServiceProvider();
var slackService = provider.GetService<ISlackService>();
Assert.NotNull(slackService);
Assert.IsType<NoopSlackService>(slackService);
}
[Fact]
public void AddTeamsService_AllSettingsPresent_RegistersTeamsServices()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:Teams:ClientId"] = "test-client-id",
["GlobalSettings:Teams:ClientSecret"] = "test-client-secret",
["GlobalSettings:Teams:Scopes"] = "test-scopes"
});
services.TryAddSingleton(globalSettings);
services.AddLogging();
services.TryAddScoped(_ => Substitute.For<IOrganizationIntegrationRepository>());
services.AddTeamsService(globalSettings);
var provider = services.BuildServiceProvider();
var teamsService = provider.GetService<ITeamsService>();
Assert.NotNull(teamsService);
Assert.IsType<TeamsService>(teamsService);
var bot = provider.GetService<IBot>();
Assert.NotNull(bot);
Assert.IsType<TeamsService>(bot);
var adapter = provider.GetService<IBotFrameworkHttpAdapter>();
Assert.NotNull(adapter);
Assert.IsType<BotFrameworkHttpAdapter>(adapter);
var httpClientDescriptor = services.FirstOrDefault(s =>
s.ServiceType == typeof(IHttpClientFactory));
Assert.NotNull(httpClientDescriptor);
}
[Fact]
public void AddTeamsService_SettingsMissing_RegistersNoopService()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:Teams:ClientId"] = null,
["GlobalSettings:Teams:ClientSecret"] = null,
["GlobalSettings:Teams:Scopes"] = null
});
services.AddTeamsService(globalSettings);
var provider = services.BuildServiceProvider();
var teamsService = provider.GetService<ITeamsService>();
Assert.NotNull(teamsService);
Assert.IsType<NoopTeamsService>(teamsService);
}
[Fact]
public void AddRabbitMqIntegration_RegistersEventIntegrationHandler()
{
var services = new ServiceCollection();
var listenerConfig = new TestListenerConfiguration();
// Add required dependencies
services.TryAddSingleton(Substitute.For<IEventIntegrationPublisher>());
services.TryAddSingleton(Substitute.For<IIntegrationFilterService>());
services.TryAddKeyedSingleton(EventIntegrationsCacheConstants.CacheName, Substitute.For<IFusionCache>());
services.TryAddSingleton(Substitute.For<IOrganizationIntegrationConfigurationRepository>());
services.TryAddSingleton(Substitute.For<IGroupRepository>());
services.TryAddSingleton(Substitute.For<IOrganizationRepository>());
services.TryAddSingleton(Substitute.For<IOrganizationUserRepository>());
services.AddLogging();
services.AddRabbitMqIntegration<WebhookIntegrationConfigurationDetails, TestListenerConfiguration>(listenerConfig);
var provider = services.BuildServiceProvider();
var handler = provider.GetRequiredKeyedService<IEventMessageHandler>(listenerConfig.RoutingKey);
Assert.NotNull(handler);
}
[Fact]
public void AddRabbitMqIntegration_RegistersEventListenerService()
{
var services = new ServiceCollection();
var listenerConfig = new TestListenerConfiguration();
// Add required dependencies
services.TryAddSingleton(Substitute.For<IEventIntegrationPublisher>());
services.TryAddSingleton(Substitute.For<IIntegrationFilterService>());
services.TryAddKeyedSingleton(EventIntegrationsCacheConstants.CacheName, Substitute.For<IFusionCache>());
services.TryAddSingleton(Substitute.For<IOrganizationIntegrationConfigurationRepository>());
services.TryAddSingleton(Substitute.For<IGroupRepository>());
services.TryAddSingleton(Substitute.For<IOrganizationRepository>());
services.TryAddSingleton(Substitute.For<IOrganizationUserRepository>());
services.TryAddSingleton(Substitute.For<IRabbitMqService>());
services.AddLogging();
var beforeCount = services.Count(s => s.ServiceType == typeof(IHostedService));
services.AddRabbitMqIntegration<WebhookIntegrationConfigurationDetails, TestListenerConfiguration>(listenerConfig);
var afterCount = services.Count(s => s.ServiceType == typeof(IHostedService));
// AddRabbitMqIntegration should register 2 hosted services (Event + Integration listeners)
Assert.Equal(2, afterCount - beforeCount);
}
[Fact]
public void AddRabbitMqIntegration_RegistersIntegrationListenerService()
{
var services = new ServiceCollection();
var listenerConfig = new TestListenerConfiguration();
// Add required dependencies
services.TryAddSingleton(Substitute.For<IEventIntegrationPublisher>());
services.TryAddSingleton(Substitute.For<IIntegrationFilterService>());
services.TryAddKeyedSingleton(EventIntegrationsCacheConstants.CacheName, Substitute.For<IFusionCache>());
services.TryAddSingleton(Substitute.For<IOrganizationIntegrationConfigurationRepository>());
services.TryAddSingleton(Substitute.For<IGroupRepository>());
services.TryAddSingleton(Substitute.For<IOrganizationRepository>());
services.TryAddSingleton(Substitute.For<IOrganizationUserRepository>());
services.TryAddSingleton(Substitute.For<IRabbitMqService>());
services.TryAddSingleton(Substitute.For<IIntegrationHandler<WebhookIntegrationConfigurationDetails>>());
services.TryAddSingleton(TimeProvider.System);
services.AddLogging();
var beforeCount = services.Count(s => s.ServiceType == typeof(IHostedService));
services.AddRabbitMqIntegration<WebhookIntegrationConfigurationDetails, TestListenerConfiguration>(listenerConfig);
var afterCount = services.Count(s => s.ServiceType == typeof(IHostedService));
// AddRabbitMqIntegration should register 2 hosted services (Event + Integration listeners)
Assert.Equal(2, afterCount - beforeCount);
}
[Fact]
public void AddAzureServiceBusIntegration_RegistersEventIntegrationHandler()
{
var services = new ServiceCollection();
var listenerConfig = new TestListenerConfiguration();
// Add required dependencies
services.TryAddSingleton(Substitute.For<IEventIntegrationPublisher>());
services.TryAddSingleton(Substitute.For<IIntegrationFilterService>());
services.TryAddKeyedSingleton(EventIntegrationsCacheConstants.CacheName, Substitute.For<IFusionCache>());
services.TryAddSingleton(Substitute.For<IOrganizationIntegrationConfigurationRepository>());
services.TryAddSingleton(Substitute.For<IGroupRepository>());
services.TryAddSingleton(Substitute.For<IOrganizationRepository>());
services.TryAddSingleton(Substitute.For<IOrganizationUserRepository>());
services.AddLogging();
services.AddAzureServiceBusIntegration<WebhookIntegrationConfigurationDetails, TestListenerConfiguration>(listenerConfig);
var provider = services.BuildServiceProvider();
var handler = provider.GetRequiredKeyedService<IEventMessageHandler>(listenerConfig.RoutingKey);
Assert.NotNull(handler);
}
[Fact]
public void AddAzureServiceBusIntegration_RegistersEventListenerService()
{
var services = new ServiceCollection();
var listenerConfig = new TestListenerConfiguration();
// Add required dependencies
services.TryAddSingleton(Substitute.For<IEventIntegrationPublisher>());
services.TryAddSingleton(Substitute.For<IIntegrationFilterService>());
services.TryAddKeyedSingleton(EventIntegrationsCacheConstants.CacheName, Substitute.For<IFusionCache>());
services.TryAddSingleton(Substitute.For<IOrganizationIntegrationConfigurationRepository>());
services.TryAddSingleton(Substitute.For<IGroupRepository>());
services.TryAddSingleton(Substitute.For<IOrganizationRepository>());
services.TryAddSingleton(Substitute.For<IOrganizationUserRepository>());
services.TryAddSingleton(Substitute.For<IAzureServiceBusService>());
services.AddLogging();
var beforeCount = services.Count(s => s.ServiceType == typeof(IHostedService));
services.AddAzureServiceBusIntegration<WebhookIntegrationConfigurationDetails, TestListenerConfiguration>(listenerConfig);
var afterCount = services.Count(s => s.ServiceType == typeof(IHostedService));
// AddAzureServiceBusIntegration should register 2 hosted services (Event + Integration listeners)
Assert.Equal(2, afterCount - beforeCount);
}
[Fact]
public void AddAzureServiceBusIntegration_RegistersIntegrationListenerService()
{
var services = new ServiceCollection();
var listenerConfig = new TestListenerConfiguration();
// Add required dependencies
services.TryAddSingleton(Substitute.For<IEventIntegrationPublisher>());
services.TryAddSingleton(Substitute.For<IIntegrationFilterService>());
services.TryAddKeyedSingleton(EventIntegrationsCacheConstants.CacheName, Substitute.For<IFusionCache>());
services.TryAddSingleton(Substitute.For<IOrganizationIntegrationConfigurationRepository>());
services.TryAddSingleton(Substitute.For<IGroupRepository>());
services.TryAddSingleton(Substitute.For<IOrganizationRepository>());
services.TryAddSingleton(Substitute.For<IOrganizationUserRepository>());
services.TryAddSingleton(Substitute.For<IAzureServiceBusService>());
services.TryAddSingleton(Substitute.For<IIntegrationHandler<WebhookIntegrationConfigurationDetails>>());
services.AddLogging();
var beforeCount = services.Count(s => s.ServiceType == typeof(IHostedService));
services.AddAzureServiceBusIntegration<WebhookIntegrationConfigurationDetails, TestListenerConfiguration>(listenerConfig);
var afterCount = services.Count(s => s.ServiceType == typeof(IHostedService));
// AddAzureServiceBusIntegration should register 2 hosted services (Event + Integration listeners)
Assert.Equal(2, afterCount - beforeCount);
}
[Fact]
public void AddEventIntegrationServices_RegistersCommonServices()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings([]);
// Add prerequisites
services.TryAddSingleton(globalSettings);
services.TryAddSingleton(Substitute.For<IConnectionMultiplexer>());
services.AddLogging();
services.AddEventIntegrationServices(globalSettings);
// Verify common services are registered
Assert.Contains(services, s => s.ServiceType == typeof(IIntegrationFilterService));
Assert.Contains(services, s => s.ServiceType == typeof(TimeProvider));
// Verify HttpClients for handlers are registered
var httpClientDescriptors = services.Where(s => s.ServiceType == typeof(IHttpClientFactory)).ToList();
Assert.NotEmpty(httpClientDescriptors);
}
[Fact]
public void AddEventIntegrationServices_RegistersIntegrationHandlers()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings([]);
// Add prerequisites
services.TryAddSingleton(globalSettings);
services.TryAddSingleton(Substitute.For<IConnectionMultiplexer>());
services.AddLogging();
services.AddEventIntegrationServices(globalSettings);
// Verify integration handlers are registered
Assert.Contains(services, s => s.ServiceType == typeof(IIntegrationHandler<SlackIntegrationConfigurationDetails>));
Assert.Contains(services, s => s.ServiceType == typeof(IIntegrationHandler<WebhookIntegrationConfigurationDetails>));
Assert.Contains(services, s => s.ServiceType == typeof(IIntegrationHandler<DatadogIntegrationConfigurationDetails>));
Assert.Contains(services, s => s.ServiceType == typeof(IIntegrationHandler<TeamsIntegrationConfigurationDetails>));
}
[Fact]
public void AddEventIntegrationServices_RabbitMqEnabled_RegistersRabbitMqListeners()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:RabbitMq:HostName"] = "localhost",
["GlobalSettings:EventLogging:RabbitMq:Username"] = "user",
["GlobalSettings:EventLogging:RabbitMq:Password"] = "pass",
["GlobalSettings:EventLogging:RabbitMq:EventExchangeName"] = "exchange"
});
// Add prerequisites
services.TryAddSingleton(globalSettings);
services.TryAddSingleton(Substitute.For<IConnectionMultiplexer>());
services.AddLogging();
var beforeCount = services.Count(s => s.ServiceType == typeof(IHostedService));
services.AddEventIntegrationServices(globalSettings);
var afterCount = services.Count(s => s.ServiceType == typeof(IHostedService));
// Should register 11 hosted services for RabbitMQ: 1 repository + 5*2 integration listeners (event+integration)
Assert.Equal(11, afterCount - beforeCount);
}
[Fact]
public void AddEventIntegrationServices_AzureServiceBusEnabled_RegistersAzureListeners()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:AzureServiceBus:ConnectionString"] = "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKeyName=test;SharedAccessKey=test",
["GlobalSettings:EventLogging:AzureServiceBus:EventTopicName"] = "events"
});
// Add prerequisites
services.TryAddSingleton(globalSettings);
services.TryAddSingleton(Substitute.For<IConnectionMultiplexer>());
services.AddLogging();
var beforeCount = services.Count(s => s.ServiceType == typeof(IHostedService));
services.AddEventIntegrationServices(globalSettings);
var afterCount = services.Count(s => s.ServiceType == typeof(IHostedService));
// Should register 11 hosted services for Azure Service Bus: 1 repository + 5*2 integration listeners (event+integration)
Assert.Equal(11, afterCount - beforeCount);
}
[Fact]
public void AddEventIntegrationServices_BothEnabled_AzureServiceBusTakesPrecedence()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:RabbitMq:HostName"] = "localhost",
["GlobalSettings:EventLogging:RabbitMq:Username"] = "user",
["GlobalSettings:EventLogging:RabbitMq:Password"] = "pass",
["GlobalSettings:EventLogging:RabbitMq:EventExchangeName"] = "exchange",
["GlobalSettings:EventLogging:AzureServiceBus:ConnectionString"] = "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKeyName=test;SharedAccessKey=test",
["GlobalSettings:EventLogging:AzureServiceBus:EventTopicName"] = "events"
});
// Add prerequisites
services.TryAddSingleton(globalSettings);
services.TryAddSingleton(Substitute.For<IConnectionMultiplexer>());
services.AddLogging();
var beforeCount = services.Count(s => s.ServiceType == typeof(IHostedService));
services.AddEventIntegrationServices(globalSettings);
var afterCount = services.Count(s => s.ServiceType == typeof(IHostedService));
// Should register 11 hosted services for Azure Service Bus: 1 repository + 5*2 integration listeners (event+integration)
// NO RabbitMQ services should be enabled because ASB takes precedence
Assert.Equal(11, afterCount - beforeCount);
}
[Fact]
public void AddEventIntegrationServices_NeitherEnabled_RegistersNoListeners()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings([]);
// Add prerequisites
services.TryAddSingleton(globalSettings);
services.TryAddSingleton(Substitute.For<IConnectionMultiplexer>());
services.AddLogging();
var beforeCount = services.Count(s => s.ServiceType == typeof(IHostedService));
services.AddEventIntegrationServices(globalSettings);
var afterCount = services.Count(s => s.ServiceType == typeof(IHostedService));
// Should register no hosted services when neither RabbitMQ nor Azure Service Bus is enabled
Assert.Equal(0, afterCount - beforeCount);
}
[Fact]
public void AddEventWriteServices_AzureServiceBusEnabled_RegistersAzureServices()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:AzureServiceBus:ConnectionString"] = "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKeyName=test;SharedAccessKey=test",
["GlobalSettings:EventLogging:AzureServiceBus:EventTopicName"] = "events"
});
services.AddEventWriteServices(globalSettings);
Assert.Contains(services, s => s.ServiceType == typeof(IEventIntegrationPublisher) && s.ImplementationType == typeof(AzureServiceBusService));
Assert.Contains(services, s => s.ServiceType == typeof(IEventWriteService) && s.ImplementationType == typeof(EventIntegrationEventWriteService));
}
[Fact]
public void AddEventWriteServices_RabbitMqEnabled_RegistersRabbitMqServices()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:RabbitMq:HostName"] = "localhost",
["GlobalSettings:EventLogging:RabbitMq:Username"] = "user",
["GlobalSettings:EventLogging:RabbitMq:Password"] = "pass",
["GlobalSettings:EventLogging:RabbitMq:EventExchangeName"] = "exchange"
});
services.AddEventWriteServices(globalSettings);
Assert.Contains(services, s => s.ServiceType == typeof(IEventIntegrationPublisher) && s.ImplementationType == typeof(RabbitMqService));
Assert.Contains(services, s => s.ServiceType == typeof(IEventWriteService) && s.ImplementationType == typeof(EventIntegrationEventWriteService));
}
[Fact]
public void AddEventWriteServices_EventsConnectionStringPresent_RegistersAzureQueueService()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:Events:ConnectionString"] = "DefaultEndpointsProtocol=https;AccountName=test;AccountKey=test;EndpointSuffix=core.windows.net"
});
services.AddEventWriteServices(globalSettings);
Assert.Contains(services, s => s.ServiceType == typeof(IEventWriteService) && s.ImplementationType == typeof(AzureQueueEventWriteService));
}
[Fact]
public void AddEventWriteServices_SelfHosted_RegistersRepositoryService()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:SelfHosted"] = "true"
});
services.AddEventWriteServices(globalSettings);
Assert.Contains(services, s => s.ServiceType == typeof(IEventWriteService) && s.ImplementationType == typeof(RepositoryEventWriteService));
}
[Fact]
public void AddEventWriteServices_NothingEnabled_RegistersNoopService()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings([]);
services.AddEventWriteServices(globalSettings);
Assert.Contains(services, s => s.ServiceType == typeof(IEventWriteService) && s.ImplementationType == typeof(NoopEventWriteService));
}
[Fact]
public void AddEventWriteServices_AzureTakesPrecedenceOverRabbitMq()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:AzureServiceBus:ConnectionString"] = "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKeyName=test;SharedAccessKey=test",
["GlobalSettings:EventLogging:AzureServiceBus:EventTopicName"] = "events",
["GlobalSettings:EventLogging:RabbitMq:HostName"] = "localhost",
["GlobalSettings:EventLogging:RabbitMq:Username"] = "user",
["GlobalSettings:EventLogging:RabbitMq:Password"] = "pass",
["GlobalSettings:EventLogging:RabbitMq:EventExchangeName"] = "exchange"
});
services.AddEventWriteServices(globalSettings);
// Should use Azure Service Bus, not RabbitMQ
Assert.Contains(services, s => s.ServiceType == typeof(IEventIntegrationPublisher) && s.ImplementationType == typeof(AzureServiceBusService));
Assert.DoesNotContain(services, s => s.ServiceType == typeof(IEventIntegrationPublisher) && s.ImplementationType == typeof(RabbitMqService));
}
[Fact]
public void AddAzureServiceBusListeners_AzureServiceBusEnabled_RegistersAzureServiceBusServices()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:AzureServiceBus:ConnectionString"] = "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKeyName=test;SharedAccessKey=test",
["GlobalSettings:EventLogging:AzureServiceBus:EventTopicName"] = "events"
});
// Add prerequisites
services.TryAddSingleton(globalSettings);
services.TryAddSingleton(Substitute.For<IConnectionMultiplexer>());
services.AddLogging();
services.AddAzureServiceBusListeners(globalSettings);
Assert.Contains(services, s => s.ServiceType == typeof(IAzureServiceBusService));
Assert.Contains(services, s => s.ServiceType == typeof(IEventRepository));
Assert.Contains(services, s => s.ServiceType == typeof(AzureTableStorageEventHandler));
}
[Fact]
public void AddAzureServiceBusListeners_AzureServiceBusDisabled_ReturnsEarly()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings([]);
var initialCount = services.Count;
services.AddAzureServiceBusListeners(globalSettings);
var finalCount = services.Count;
Assert.Equal(initialCount, finalCount);
}
[Fact]
public void AddRabbitMqListeners_RabbitMqEnabled_RegistersRabbitMqServices()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings(new Dictionary<string, string?>
{
["GlobalSettings:EventLogging:RabbitMq:HostName"] = "localhost",
["GlobalSettings:EventLogging:RabbitMq:Username"] = "user",
["GlobalSettings:EventLogging:RabbitMq:Password"] = "pass",
["GlobalSettings:EventLogging:RabbitMq:EventExchangeName"] = "exchange"
});
// Add prerequisites
services.TryAddSingleton(globalSettings);
services.TryAddSingleton(Substitute.For<IConnectionMultiplexer>());
services.AddLogging();
services.AddRabbitMqListeners(globalSettings);
Assert.Contains(services, s => s.ServiceType == typeof(IRabbitMqService));
Assert.Contains(services, s => s.ServiceType == typeof(EventRepositoryHandler));
}
[Fact]
public void AddRabbitMqListeners_RabbitMqDisabled_ReturnsEarly()
{
var services = new ServiceCollection();
var globalSettings = CreateGlobalSettings([]);
var initialCount = services.Count;
services.AddRabbitMqListeners(globalSettings);
var finalCount = services.Count;
Assert.Equal(initialCount, finalCount);
}
private static GlobalSettings CreateGlobalSettings(Dictionary<string, string?> data)
{
var config = new ConfigurationBuilder()

1
test/Core.Test/AdminConsole/Models/Data/EventIntegrations/TestListenerConfiguration.cs
View File

@@ -17,4 +17,5 @@ public class TestListenerConfiguration : IIntegrationListenerConfiguration
public int EventPrefetchCount => 0;
public int IntegrationMaxConcurrentCalls => 1;
public int IntegrationPrefetchCount => 0;
public string RoutingKey => IntegrationType.ToRoutingKey();
}