在对Service Fabric传输中的可靠服务的调用中传递用户和审核信息

在这里,我们不能只注入一个特定的处理程序或类似的服务,我们必须提供我们自己的自定义工厂.为了不必重新实现FabricTransportServiceRemotingClientFactory我的细节,我只需将其封装在我自己的实现中IServiceRemotingClientFactory:

public class AuditedFabricTransportServiceRemotingClientFactory : IServiceRemotingClientFactory, ICommunicationClientFactory
{
    private readonly ICommunicationClientFactory _innerClientFactory;

    public AuditedFabricTransportServiceRemotingClientFactory(ICommunicationClientFactory innerClientFactory)
    {
        _innerClientFactory = innerClientFactory;
        _innerClientFactory.ClientConnected += OnClientConnected;
        _innerClientFactory.ClientDisconnected += OnClientDisconnected;
    }

    private void OnClientConnected(object sender, CommunicationClientEventArgs e)
    {
        EventHandler> clientConnected = this.ClientConnected;
        if (clientConnected == null) return;
        clientConnected((object)this, new CommunicationClientEventArgs()
        {
            Client = e.Client
        });
    }

    private void OnClientDisconnected(object sender, CommunicationClientEventArgs e)
    {
        EventHandler> clientDisconnected = this.ClientDisconnected;
        if (clientDisconnected == null) return;
        clientDisconnected((object)this, new CommunicationClientEventArgs()
        {
            Client = e.Client
        });
    }

    public async Task GetClientAsync(
        Uri serviceUri,
        ServicePartitionKey partitionKey, 
        TargetReplicaSelector targetReplicaSelector, 
        string listenerName,
        OperationRetrySettings retrySettings, 
        CancellationToken cancellationToken)
    {
        var client = await _innerClientFactory.GetClientAsync(
            serviceUri, 
            partitionKey, 
            targetReplicaSelector, 
            listenerName, 
            retrySettings, 
            cancellationToken);
        return new AuditedFabricTransportServiceRemotingClient(client);
    }

    public async Task GetClientAsync(
        ResolvedServicePartition previousRsp, 
        TargetReplicaSelector targetReplicaSelector, 
        string listenerName, 
        OperationRetrySettings retrySettings,
        CancellationToken cancellationToken)
    {
        var client = await _innerClientFactory.GetClientAsync(
            previousRsp, 
            targetReplicaSelector, 
            listenerName, 
            retrySettings, 
            cancellationToken);
        return new AuditedFabricTransportServiceRemotingClient(client);
    }

    public Task ReportOperationExceptionAsync(
        IServiceRemotingClient client, 
        ExceptionInformation exceptionInformation, 
        OperationRetrySettings retrySettings,
        CancellationToken cancellationToken)
    {
        return _innerClientFactory.ReportOperationExceptionAsync(
            client, 
            exceptionInformation, 
            retrySettings, 
            cancellationToken);
    }

    public event EventHandler> ClientConnected;
    public event EventHandler> ClientDisconnected;
}

 public class AuditedFabricTransportServiceRemotingClient : IServiceRemotingClient, ICommunicationClient
{
    private readonly IServiceRemotingClient _innerClient;

    public AuditedFabricTransportServiceRemotingClient(IServiceRemotingClient innerClient)
    {
        _innerClient = innerClient;
    }

    ~AuditedFabricTransportServiceRemotingClient()
    {
        if (this._innerClient == null) return;
        var disposable = this._innerClient as IDisposable;
        disposable?.Dispose();
    }

    Task IServiceRemotingClient.RequestResponseAsync(ServiceRemotingMessageHeaders messageHeaders, byte[] requestBody)
    {            
        messageHeaders.SetUser(ServiceRequestContext.Current.User);
        messageHeaders.SetCorrelationId(ServiceRequestContext.Current.CorrelationId);
        return this._innerClient.RequestResponseAsync(messageHeaders, requestBody);
    }

    void IServiceRemotingClient.SendOneWay(ServiceRemotingMessageHeaders messageHeaders, byte[] requestBody)
    {
        messageHeaders.SetUser(ServiceRequestContext.Current.User);
        messageHeaders.SetCorrelationId(ServiceRequestContext.Current.CorrelationId);
        this._innerClient.SendOneWay(messageHeaders, requestBody);
    }

    public ResolvedServicePartition ResolvedServicePartition
    {
        get { return this._innerClient.ResolvedServicePartition; }
        set { this._innerClient.ResolvedServicePartition = value; }
    }

    public string ListenerName
    {
        get { return this._innerClient.ListenerName; }
        set { this._innerClient.ListenerName = value; }
    }
    public ResolvedServiceEndpoint Endpoint
    {
        get { return this._innerClient.Endpoint; }
        set { this._innerClient.Endpoint = value; }
    }
}

public sealed class ServiceRequestContext
{
    private static readonly string ContextKey = Guid.NewGuid().ToString();

    public ServiceRequestContext(Guid correlationId, string user)
    {
        this.CorrelationId = correlationId;
        this.User = user;
    }

    public Guid CorrelationId { get; private set; }

    public string User { get; private set; }

    public static ServiceRequestContext Current
    {
        get { return (ServiceRequestContext)CallContext.LogicalGetData(ContextKey); }
        internal set
        {
            if (value == null)
            {
                CallContext.FreeNamedDataSlot(ContextKey);
            }
            else
            {
                CallContext.LogicalSetData(ContextKey, value);
            }
        }
    }

    public static Task RunInRequestContext(Func action, Guid correlationId, string user)
    {
        Task task = null;
        task = new Task(async () =>
        {
            Debug.Assert(ServiceRequestContext.Current == null);
            ServiceRequestContext.Current = new ServiceRequestContext(correlationId, user);
            try
            {
                await action();
            }
            finally
            {
                ServiceRequestContext.Current = null;
            }
        });
        task.Start();
        return task.Unwrap();
    }

    public static Task RunInRequestContext(Func> action, Guid correlationId, string user)
    {
        Task> task = null;
        task = new Task>(async () =>
        {
            Debug.Assert(ServiceRequestContext.Current == null);
            ServiceRequestContext.Current = new ServiceRequestContext(correlationId, user);
            try
            {
                return await action();
            }
            finally
            {
                ServiceRequestContext.Current = null;
            }
        });
        task.Start();
        return task.Unwrap();
    }
}

    public override Task RequestResponseAsync(
        IServiceRemotingRequestContext requestContext, 
        ServiceRemotingMessageHeaders messageHeaders, 
        byte[] requestBody)
    {
        var user = messageHeaders.GetUser();
        var correlationId = messageHeaders.GetCorrelationId();

        return ServiceRequestContext.RunInRequestContext(async () => 
            await base.RequestResponseAsync(
                requestContext, 
                messageHeaders, 
                requestBody), 
            correlationId, user);
    }

var task = ServiceRequestContext.RunInRequestContext(async () =>
{
    var serviceA = ServiceProxyFactory.CreateServiceProxy(new Uri($"{FabricRuntime.GetActivationContext().ApplicationName}/ServiceA"));
    await serviceA.DoStuffAsync(CancellationToken.None);
}, Guid.NewGuid(), user);