diff --git a/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/DaprAspNetCore/AbpDaprEndpointRouteBuilderExtensions.cs b/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/DaprAspNetCore/AbpDaprEndpointRouteBuilderExtensions.cs deleted file mode 100644 index 24f68ef9ab..0000000000 --- a/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/DaprAspNetCore/AbpDaprEndpointRouteBuilderExtensions.cs +++ /dev/null @@ -1,296 +0,0 @@ -// ------------------------------------------------------------------------ -// Copyright 2021 The Dapr Authors -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// http://www.apache.org/licenses/LICENSE-2.0 -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. -// ------------------------------------------------------------------------ - -using System; -using System.Collections.Generic; -using System.Threading.Tasks; - -namespace Dapr -{ - /// - /// This class defines configurations for the subscribe endpoint. - /// - public class AbpSubscribeOptions - { - /// - /// Gets or Sets a value which indicates whether to enable or disable processing raw messages. - /// - public bool EnableRawPayload { get; set; } - - /// - /// An optional delegate used to configure the subscriptions. - /// - public Func, Task>? SubscriptionsCallback { get; set; } - } - - /// - /// This class defines subscribe endpoint response - /// - public class AbpSubscription - { - /// - /// Gets or sets the topic name. - /// - public string Topic { get; set; } = default!; - - /// - /// Gets or sets the pubsub name - /// - public string PubsubName { get; set; } = default!; - - /// - /// Gets or sets the route - /// - public string? Route { get; set; } - - /// - /// Gets or sets the routes - /// - public AbpRoutes? Routes { get; set; } - - /// - /// Gets or sets the metadata. - /// - public AbpMetadata? Metadata { get; set; } - - /// - /// Gets or sets the deadletter topic. - /// - public string? DeadLetterTopic { get; set; } - } - - /// - /// This class defines the metadata for subscribe endpoint. - /// - public class AbpMetadata : Dictionary - { - /// - /// Initializes a new instance of the Metadata class. - /// - public AbpMetadata() { } - - /// - /// Initializes a new instance of the Metadata class. - /// - /// - public AbpMetadata(IDictionary dictionary) : base(dictionary) { } - - /// - /// RawPayload key - /// - internal const string RawPayload = "rawPayload"; - } - - /// - /// This class defines the routes for subscribe endpoint. - /// - public class AbpRoutes - { - /// - /// Gets or sets the default route - /// - public string? Default { get; set; } - - /// - /// Gets or sets the routing rules - /// - public List? Rules { get; set; } - } - - /// - /// This class defines the rule for subscribe endpoint. - /// - public class AbpRule - { - /// - /// Gets or sets the CEL expression to match this route. - /// - public string Match { get; set; } = default!; - - /// - /// Gets or sets the path of the route. - /// - public string Path { get; set; } = default!; - } -} - -namespace Microsoft.AspNetCore.Builder -{ - using System.Collections.Generic; - using System.Linq; - using System.Text.Json; - using System.Text.Json.Serialization; - using Dapr; - using Microsoft.AspNetCore.Http; - using Microsoft.AspNetCore.Routing; - using Microsoft.AspNetCore.Routing.Patterns; - using Microsoft.Extensions.DependencyInjection; - using Microsoft.Extensions.Logging; - - /// - /// Contains extension methods for . - /// - public static class AbpDaprEndpointRouteBuilderExtensions - { - /// - /// Maps an endpoint that will respond to requests to /dapr/subscribe from the - /// Dapr runtime. - /// - /// The . - /// The . - public static IEndpointConventionBuilder MapAbpSubscribeHandler(this IEndpointRouteBuilder endpoints) - { - return CreateSubscribeEndPoint(endpoints); - } - - /// - /// Maps an endpoint that will respond to requests to /dapr/subscribe from the - /// Dapr runtime. - /// - /// The . - /// Configuration options - /// The . - /// - public static IEndpointConventionBuilder MapAbpSubscribeHandler(this IEndpointRouteBuilder endpoints, AbpSubscribeOptions options) - { - return CreateSubscribeEndPoint(endpoints, options); - } - - private static IEndpointConventionBuilder CreateSubscribeEndPoint(IEndpointRouteBuilder endpoints, AbpSubscribeOptions? options = null) - { - if (endpoints is null) - { - throw new System.ArgumentNullException(nameof(endpoints)); - } - - return endpoints.MapGet("dapr/subscribe", async context => - { - var logger = context.RequestServices.GetService()?.CreateLogger("DaprTopicSubscription"); - var dataSource = context.RequestServices.GetRequiredService(); - var subscriptions = dataSource.Endpoints - .OfType() - .Where(e => e.Metadata.GetOrderedMetadata().Any(t => t.Name != null)) // only endpoints which have TopicAttribute with not null Name. - .SelectMany(e => - { - var topicMetadata = e.Metadata.GetOrderedMetadata(); - var originalTopicMetadata = e.Metadata.GetOrderedMetadata(); - - var subs = new List<(string PubsubName, string Name, string? DeadLetterTopic, bool? EnableRawPayload, string Match, int Priority, Dictionary OriginalTopicMetadata, string? MetadataSeparator, RoutePattern RoutePattern)>(); - - for (int i = 0; i < topicMetadata.Count(); i++) - { - subs.Add((topicMetadata[i].PubsubName, - topicMetadata[i].Name, - (topicMetadata[i] as IDeadLetterTopicMetadata)?.DeadLetterTopic, - (topicMetadata[i] as IRawTopicMetadata)?.EnableRawPayload, - topicMetadata[i].Match, - topicMetadata[i].Priority, - originalTopicMetadata.Where(m => (topicMetadata[i] as IOwnedOriginalTopicMetadata)?.OwnedMetadatas?.Any(o => o.Equals(m.Id)) == true || string.IsNullOrEmpty(m.Id)) - .GroupBy(c => c.Name) - .ToDictionary(m => m.Key, m => m.Select(c => c.Value).Distinct().ToArray()), - (topicMetadata[i] as IOwnedOriginalTopicMetadata)?.MetadataSeparator, - e.RoutePattern)); - } - - return subs; - }) - .Distinct() - .GroupBy(e => new { e.PubsubName, e.Name }) - .Select(e => e.OrderBy(e => e.Priority)) - .Select(e => - { - var first = e.First(); - var rawPayload = e.Any(e => e.EnableRawPayload.GetValueOrDefault()); - var metadataSeparator = e.FirstOrDefault(e => !string.IsNullOrEmpty(e.MetadataSeparator)).MetadataSeparator?.ToString() ?? ","; - var rules = e.Where(e => !string.IsNullOrEmpty(e.Match)).ToList(); - var defaultRoutes = e.Where(e => string.IsNullOrEmpty(e.Match)).Select(e => RoutePatternToString(e.RoutePattern)).ToList(); - var defaultRoute = defaultRoutes.FirstOrDefault(); - - //multiple identical names. use comma separation. - var metadata = new AbpMetadata(e.SelectMany(c => c.OriginalTopicMetadata).GroupBy(c => c.Key).ToDictionary(c => c.Key, c => string.Join(metadataSeparator, c.SelectMany(c => c.Value).Distinct()))); - if (rawPayload || options?.EnableRawPayload is true) - { - metadata.Add(AbpMetadata.RawPayload, "true"); - } - - if (logger != null) - { - if (defaultRoutes.Count > 1) - { - logger.LogError("A default subscription to topic {name} on pubsub {pubsub} already exists.", first.Name, first.PubsubName); - } - - var duplicatePriorities = rules.GroupBy(e => e.Priority) - .Where(g => g.Count() > 1) - .ToDictionary(x => x.Key, y => y.Count()); - - foreach (var entry in duplicatePriorities) - { - logger.LogError("A subscription to topic {name} on pubsub {pubsub} has duplicate priorities for {priority}: found {count} occurrences.", first.Name, first.PubsubName, entry.Key, entry.Value); - } - } - - var subscription = new AbpSubscription - { - Topic = first.Name, - PubsubName = first.PubsubName, - Metadata = metadata.Count > 0 ? metadata : null, - }; - - if (first.DeadLetterTopic != null) - { - subscription.DeadLetterTopic = first.DeadLetterTopic; - } - - // Use the V2 routing rules structure - if (rules.Count > 0) - { - subscription.Routes = new AbpRoutes - { - Rules = rules.Select(e => new AbpRule - { - Match = e.Match, - Path = RoutePatternToString(e.RoutePattern), - }).ToList(), - Default = defaultRoute, - }; - } - // Use the V1 structure for backward compatibility. - else - { - subscription.Route = defaultRoute; - } - - return subscription; - }) - .OrderBy(e => (e.PubsubName, e.Topic)) - .ToList(); - - await options?.SubscriptionsCallback!(subscriptions)!; - await context.Response.WriteAsync(JsonSerializer.Serialize(subscriptions, - new JsonSerializerOptions - { - PropertyNamingPolicy = JsonNamingPolicy.CamelCase, - DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull - })); - }); - } - - private static string RoutePatternToString(RoutePattern routePattern) - { - return string.Join("/", routePattern.PathSegments - .Select(segment => string.Concat(segment.Parts.Cast() - .Select(part => part.Content)))); - } - } -} diff --git a/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/AbpAspNetCoreMvcDaprEventBusModule.cs b/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/AbpAspNetCoreMvcDaprEventBusModule.cs index a291c4e769..a4a4f97f5d 100644 --- a/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/AbpAspNetCoreMvcDaprEventBusModule.cs +++ b/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/AbpAspNetCoreMvcDaprEventBusModule.cs @@ -1,11 +1,16 @@ -using System.Linq; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text.Json; using System.Threading.Tasks; using Dapr; using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Http; using Microsoft.AspNetCore.Routing; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; -using Volo.Abp.DependencyInjection; +using Volo.Abp.Dapr; using Volo.Abp.EventBus; using Volo.Abp.EventBus.Dapr; using Volo.Abp.EventBus.Distributed; @@ -21,50 +26,98 @@ public class AbpAspNetCoreMvcDaprEventBusModule : AbpModule { public override void ConfigureServices(ServiceConfigurationContext context) { - var subscribeOptions = context.Services.ExecutePreConfiguredActions(); - - Configure(options => + PostConfigure(options => { options.EndpointConfigureActions.Add(endpointContext => { - var rootServiceProvider = endpointContext.ScopeServiceProvider.GetRequiredService(); - subscribeOptions.SubscriptionsCallback = subscriptions => - { - var daprEventBusOptions = rootServiceProvider.GetRequiredService>().Value; - foreach (var handler in rootServiceProvider.GetRequiredService>().Value.Handlers) + var topicMetadatas = endpointContext.Endpoints.DataSources.SelectMany(x => x.Endpoints).OfType() + .Where(e => e.Metadata.GetOrderedMetadata().Any(t => t.Name != null)) + .SelectMany(e => e.Metadata.GetOrderedMetadata()) + .ToList(); + + var endpointConventionBuilder = endpointContext.Endpoints.MapPost( + "/api/abp/dapr/event", async httpContext => { - foreach (var @interface in handler.GetInterfaces().Where(x => x.IsGenericType && x.GetGenericTypeDefinition() == typeof(IDistributedEventHandler<>))) - { - var eventType = @interface.GetGenericArguments()[0]; - var eventName = EventNameAttribute.GetNameOrDefault(eventType); - - if (subscriptions.Any(x => x.PubsubName == daprEventBusOptions.PubSubName && x.Topic == eventName)) - { - // Controllers with a [Topic] attribute can replace built-in event handlers. - continue; - } - - var subscription = new AbpSubscription - { - PubsubName = daprEventBusOptions.PubSubName, - Topic = eventName, - Route = AbpAspNetCoreMvcDaprPubSubConsts.DaprEventCallbackUrl, - Metadata = new AbpMetadata - { - { - AbpMetadata.RawPayload, "true" - } - } - }; - subscriptions.Add(subscription); - } - } - - return Task.CompletedTask; - }; - - endpointContext.Endpoints.MapAbpSubscribeHandler(subscribeOptions); + await HandleEventAsync(httpContext); + }); + + var abpEvents = GetAbpEvents(endpointContext); + foreach (var @event in abpEvents.Where(x => !topicMetadatas.Any(t => t.PubsubName == x.PubsubName && t.Name == x.Name))) + { + endpointConventionBuilder.WithMetadata(new TopicAttribute( + @event.PubsubName, + @event.Name, + true)); + } + + endpointContext.Endpoints.MapSubscribeHandler(); }); }); } + + private List GetAbpEvents(EndpointRouteBuilderContext endpointContext) + { + var subscriptions = new List(); + var daprEventBusOptions = endpointContext.Endpoints.ServiceProvider.GetRequiredService>().Value; + + foreach (var @interface in endpointContext.Endpoints.ServiceProvider.GetRequiredService>().Value.Handlers + .SelectMany(x => x.GetInterfaces().Where(i => i.IsGenericType && i.GetGenericTypeDefinition() == typeof(IDistributedEventHandler<>)))) + { + var eventType = @interface.GetGenericArguments()[0]; + var eventName = EventNameAttribute.GetNameOrDefault(eventType); + + var subscription = new TopicAttribute(daprEventBusOptions.PubSubName, eventName); + subscriptions.Add(subscription); + } + + return subscriptions; + } + + private async static Task HandleEventAsync(HttpContext httpContext) + { + var logger = httpContext.RequestServices.GetRequiredService>(); + + httpContext.ValidateDaprAppApiToken(); + + var daprSerializer = httpContext.RequestServices.GetRequiredService(); + var body = (await JsonDocument.ParseAsync(httpContext.Request.Body)); + + var pubSubName = body.RootElement.GetProperty("pubsubname").GetString(); + var topic = body.RootElement.GetProperty("topic").GetString(); + var data = body.RootElement.GetProperty("data").GetRawText(); + if (pubSubName.IsNullOrWhiteSpace() || topic.IsNullOrWhiteSpace() || data.IsNullOrWhiteSpace()) + { + logger.LogError("Invalid Dapr event request."); + httpContext.Response.StatusCode = 400; + return; + } + + var distributedEventBus = httpContext.RequestServices.GetRequiredService(); + + if (IsAbpDaprEventData(data)) + { + var daprEventData = daprSerializer.Deserialize(data, typeof(AbpDaprEventData)).As(); + var eventData = daprSerializer.Deserialize(daprEventData.JsonData, distributedEventBus.GetEventType(daprEventData.Topic)); + await distributedEventBus.TriggerHandlersAsync(distributedEventBus.GetEventType(daprEventData.Topic), eventData, daprEventData.MessageId, daprEventData.CorrelationId); + } + else + { + var eventData = daprSerializer.Deserialize(data, distributedEventBus.GetEventType(topic!)); + await distributedEventBus.TriggerHandlersAsync(distributedEventBus.GetEventType(topic!), eventData); + } + + httpContext.Response.StatusCode = 200; + } + + private static bool IsAbpDaprEventData(string data) + { + var document = JsonDocument.Parse(data); + var objects = document.RootElement.EnumerateObject().ToList(); + return objects.Count == 5 && + objects.Any(x => x.Name.Equals("PubSubName", StringComparison.CurrentCultureIgnoreCase)) && + objects.Any(x => x.Name.Equals("Topic", StringComparison.CurrentCultureIgnoreCase)) && + objects.Any(x => x.Name.Equals("MessageId", StringComparison.CurrentCultureIgnoreCase)) && + objects.Any(x => x.Name.Equals("JsonData", StringComparison.CurrentCultureIgnoreCase)) && + objects.Any(x => x.Name.Equals("CorrelationId", StringComparison.CurrentCultureIgnoreCase)); + } } diff --git a/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/AbpAspNetCoreMvcDaprPubSubConsts.cs b/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/AbpAspNetCoreMvcDaprPubSubConsts.cs deleted file mode 100644 index 5f224abc7e..0000000000 --- a/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/AbpAspNetCoreMvcDaprPubSubConsts.cs +++ /dev/null @@ -1,6 +0,0 @@ -namespace Volo.Abp.AspNetCore.Mvc.Dapr.EventBus; - -public class AbpAspNetCoreMvcDaprPubSubConsts -{ - public const string DaprEventCallbackUrl = "api/abp/dapr/event"; -} diff --git a/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/Controllers/AbpAspNetCoreMvcDaprEventsController.cs b/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/Controllers/AbpAspNetCoreMvcDaprEventsController.cs deleted file mode 100644 index 946c39f59c..0000000000 --- a/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/Controllers/AbpAspNetCoreMvcDaprEventsController.cs +++ /dev/null @@ -1,62 +0,0 @@ -using System; -using System.Linq; -using System.Text.Json; -using System.Threading.Tasks; -using Microsoft.AspNetCore.Mvc; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; -using Volo.Abp.Dapr; -using Volo.Abp.EventBus.Dapr; - -namespace Volo.Abp.AspNetCore.Mvc.Dapr.EventBus.Controllers; - -[Area("abp")] -[RemoteService(Name = "abp")] -public class AbpAspNetCoreMvcDaprEventsController : AbpController -{ - [HttpPost(AbpAspNetCoreMvcDaprPubSubConsts.DaprEventCallbackUrl)] - public virtual async Task EventAsync() - { - HttpContext.ValidateDaprAppApiToken(); - - var daprSerializer = HttpContext.RequestServices.GetRequiredService(); - var body = (await JsonDocument.ParseAsync(HttpContext.Request.Body)); - - var pubSubName = body.RootElement.GetProperty("pubsubname").GetString(); - var topic = body.RootElement.GetProperty("topic").GetString(); - var data = body.RootElement.GetProperty("data").GetRawText(); - if (pubSubName.IsNullOrWhiteSpace() || topic.IsNullOrWhiteSpace() || data.IsNullOrWhiteSpace()) - { - Logger.LogError("Invalid Dapr event request."); - return BadRequest(); - } - - var distributedEventBus = HttpContext.RequestServices.GetRequiredService(); - - if (IsAbpDaprEventData(data)) - { - var daprEventData = daprSerializer.Deserialize(data, typeof(AbpDaprEventData)).As(); - var eventData = daprSerializer.Deserialize(daprEventData.JsonData, distributedEventBus.GetEventType(daprEventData.Topic)); - await distributedEventBus.TriggerHandlersAsync(distributedEventBus.GetEventType(daprEventData.Topic), eventData, daprEventData.MessageId, daprEventData.CorrelationId); - } - else - { - var eventData = daprSerializer.Deserialize(data, distributedEventBus.GetEventType(topic!)); - await distributedEventBus.TriggerHandlersAsync(distributedEventBus.GetEventType(topic!), eventData); - } - - return Ok(); - } - - protected virtual bool IsAbpDaprEventData(string data) - { - var document = JsonDocument.Parse(data); - var objects = document.RootElement.EnumerateObject().ToList(); - return objects.Count == 5 && - objects.Any(x => x.Name.Equals("PubSubName", StringComparison.CurrentCultureIgnoreCase)) && - objects.Any(x => x.Name.Equals("Topic", StringComparison.CurrentCultureIgnoreCase)) && - objects.Any(x => x.Name.Equals("MessageId", StringComparison.CurrentCultureIgnoreCase)) && - objects.Any(x => x.Name.Equals("JsonData", StringComparison.CurrentCultureIgnoreCase)) && - objects.Any(x => x.Name.Equals("CorrelationId", StringComparison.CurrentCultureIgnoreCase)); - } -}