itfx/src/InterfaceForward.Application/Services/InterfaceForwardService.cs

using System.ComponentModel.DataAnnotations;
using Microsoft.AspNetCore.Authorization;
using Newtonsoft.Json.Linq;
using InterfaceForward.Application.Contracts.ForwardCore;
using InterfaceForward.Application.Filters;
using InterfaceForward.Application.Rabbit;
using InterfaceForward.Application.Services.App;
using InterfaceForward.Domain.Shared;
using InterfaceForward.Repositories.Log.Services;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;
using RabbitMQ.Client;

namespace InterfaceForward.Application.Services;

/// <summary>
/// 接口转发服务
/// </summary>
[ApiExplorerSettings(GroupName = "接口转发服务")]
[TypeFilter(typeof(ForwardAuthorizeFilter))]   
[TypeFilter(typeof(ForwardExceptionFilter))]
[AllowAnonymous]
public class InterfaceForwardService : ApplicationService
{
    private readonly InterfaceForwardCommon _forwardCommon;
    private readonly ILogRepository _logRepository;
    private readonly RabbitClient _rabbitClient;
    private readonly IHttpContextAccessor _httpContextAccessor;

    /// <inheritdoc />
    public InterfaceForwardService(ILogRepository logRepository,
        InterfaceForwardCommon forwardCommon, RabbitClient rabbitClient,IHttpContextAccessor httpContextAccessor)
    {
        _logRepository = logRepository;
        _forwardCommon = forwardCommon;
        _rabbitClient = rabbitClient;
        _httpContextAccessor = httpContextAccessor;
    }

    [HttpPost("Itfx/BatchRequest")]
    public async Task<InterfaceRelayUnifyResultDto> BatchForwardThenPushToQueueAsync(
        [FromQuery] [Required] string standardCode,
        [FromQuery] [Required] string supplierCode,
        [FromBody] [Required] List<object> data,
        [FromHeader] [Required] string followName,
        [FromHeader] string? other = null,
        [FromHeader] bool sourceData = false)
    {
        var contextList = await _forwardCommon.VerifyBusinessAndBuildForwardContexts(standardCode,
            supplierCode);

        var @event = new BatchForwardEvent
        {
            RequestId = _logRepository.AppRequestLog.RequestId,
            AppId = _logRepository.AppId,
            UpStreamCode = standardCode,
            ServiceProviderCode = supplierCode,
            Attach = other,
            IsReturnMessage = sourceData,
            ResSubscribeName = followName
        };

        var queueName = $"{BatchForwardEventHandler.BatchForwardEventSubscribe}.{_httpContextAccessor.HttpContext!.Request
            .Headers["appKey"]}.{standardCode}.{supplierCode}";

        // tips：订阅需要持久化
        _rabbitClient.Subscribe(new ConsumeOptions
        {
            Queue = queueName,
            FetchCount = AppSubscribeService.DefaultFetchCount, // 并发控制
            FailedRequeue = AppSubscribeService.DefaultFailedRequeue,
            Declaration = o =>
            {
                o.ExchangeDeclare(GlobalConst.InterfaceRelayExchange, ExchangeType.Direct, true);
                o.QueueDeclare(queueName, true);
                o.QueueBind(queueName, GlobalConst.InterfaceRelayExchange, queueName);
            }
        });

        // 手工切割任务数量
        var chunk = data.Count;
        var minLimit = contextList.Min(x => x.TargetInterface.Limit);
        var limit = minLimit > 0 ? minLimit : 1;
        if (limit > 1)
        {
            chunk /= limit;
            if (data.Count % limit != 0) chunk++;
        }

        for (int x = 0; x < chunk; x++)
        {
            var arr = new JArray();
            for (int i = 0; i < limit && x * limit + i < data.Count; i++)
            {
                arr.Add(data[x * limit + i]);
            }

            var subEvent = @event.DeepCopy();

            subEvent.Data = arr.ToJson(); // always arr
            var eventBytes = System.Text.Json.JsonSerializer.SerializeToUtf8Bytes(subEvent);
            _rabbitClient.Publish(GlobalConst.InterfaceRelayExchange, queueName, eventBytes);
        }

        return new InterfaceRelayUnifyResultDto(true, _logRepository.AppRequestLog.RequestId)
        {
            RqId = _logRepository.AppRequestLog.RequestId,
            SourceData = null,
            Other = other
        };
    }

    [HttpPost("Itfx/ForwardWithSubscribe")]
    public async Task<InterfaceRelayUnifyResultDto> ForwardWithSubscribeAsync(
        [FromQuery] [Required] string standardCode,
        [FromQuery] [Required] string supplierCode,
        [FromBody] [Required] object data,
        [FromHeader] string? other = null,
        [FromHeader] bool sourceData = false,
        [FromHeader] string? failedSubscribeName = null)
    {
        if (JToken.FromObject(data) is not JObject jobject)
        {
            throw new BusinessException(message: "非法结构，请传对象");
        }

        var contextList = await _forwardCommon.VerifyBusinessAndBuildForwardContexts(standardCode, supplierCode);

        var context = contextList.First();
        context.OriginalInterfaceInput = contextList.Count == 1
            ? jobject
            : new JObject
            {
                [standardCode] = jobject
            };

        var res = contextList.Count == 1
            ? await _forwardCommon.InternalForwardAsync(context)
            : await _forwardCommon.InternalForward2Async(contextList);

        return new InterfaceRelayUnifyResultDto(true, res)
        {
            RqId = context.ServiceProviderRequestLog.RequestId,
            SourceData = sourceData ? context.ServiceProviderRequestLog.Response : null,
            Other = other
        };
    }

    [HttpPost("InterfaceForward/ReForward")]
    public async Task<InterfaceRelayUnifyResultDto> ReForwardAsync(
        [FromQuery] [Required] string standardCode,
        [FromQuery] [Required] string supplierCode,
        [FromHeader] [Required] Guid requestId,
        [FromBody] [Required] object data,
        [FromHeader] string? other = null,
        [FromHeader] bool sourceData = false,
        [FromHeader] string? failedSubscribeName = null)
    {
        var logs = await _logRepository.GetRequestLogsAsync(requestId);

        var failedLog = logs.FirstOrDefault(x => !x.IsSuccess);
        if (failedLog == default) throw new BusinessException(message: "找不到失败日志，无法重试");

        var contextList = await _forwardCommon.VerifyBusinessAndBuildForwardContexts(standardCode, supplierCode);

        var context = contextList.First();
        context.OriginalInterfaceInput = contextList.Count == 1
            ? JObject.FromObject(data)
            : JArray.FromObject(data);

        // 从第一个失败的开始重试
        var requestLogs = logs.Where(x => !x.IsApp).ToList();
        var res = contextList.Count == 1
            ? await _forwardCommon.InternalForwardAsync(context)
            : await _forwardCommon.InternalForward3Async(contextList, requestLogs);

        return new InterfaceRelayUnifyResultDto(true, res)
        {
            RqId = context.ServiceProviderRequestLog.RequestId,
            SourceData = sourceData ? _logRepository.ServiceProviderRequestLog.Response : null,
            Other = other
        };
    }
}