cap-publisher-extensions.cs

public interface IEventBusPublisher : ICapPublisher
{
    /// <summary>
    /// Asynchronous publish an object message.
    /// </summary>
    /// <param name="name">the topic name or exchange router key.</param>
    /// <param name="tenantId">tenant id</param>
    /// <param name="contentObj">message body content, that will be serialized. (can be null)</param>
    /// <param name="callbackName">callback subscriber name</param>
    /// <param name="cancellationToken"></param>
    Task PublishAsync<T>(string name, Guid tenantId, [CanBeNull] T contentObj, string callbackName = null,
        CancellationToken cancellationToken = default);

    /// <summary>
    /// Asynchronous publish an object message with custom headers
    /// </summary>
    /// <typeparam name="T">content object</typeparam>
    /// <param name="name">the topic name or exchange router key.</param>
    /// <param name="tenantId">tenant id</param>
    /// <param name="contentObj">message body content, that will be serialized. (can be null)</param>
    /// <param name="headers">message additional headers.</param>
    /// <param name="cancellationToken"></param>
    Task PublishAsync<T>(string name, Guid tenantId, [CanBeNull] T contentObj, IDictionary<string, string> headers,
        CancellationToken cancellationToken = default);

    /// <summary>
    /// Publish an object message.
    /// </summary>
    /// <param name="name">the topic name or exchange router key.</param>
    /// <param name="tenantId">tenant id</param>
    /// <param name="contentObj">message body content, that will be serialized. (can be null)</param>
    /// <param name="callbackName">callback subscriber name</param>
    void Publish<T>(string name, Guid tenantId, [CanBeNull] T contentObj, string callbackName = null);

    /// <summary>
    /// Publish an object message.
    /// </summary>
    /// <param name="name">the topic name or exchange router key.</param>
    /// <param name="tenantId">tenant id</param>
    /// <param name="contentObj">message body content, that will be serialized. (can be null)</param>
    /// <param name="headers">message additional headers.</param>
    void Publish<T>(string name, Guid tenantId, [CanBeNull] T contentObj, IDictionary<string, string> headers);
}


public class DefaultEventBusPublisher : IEventBusPublisher
{
    private readonly ICapPublisher _capPublisher;

    public DefaultEventBusPublisher(ICapPublisher capPublisher)
    {
        _capPublisher = capPublisher;
    }

    public IServiceProvider ServiceProvider => _capPublisher.ServiceProvider;

    public AsyncLocal<ICapTransaction> Transaction => _capPublisher.Transaction;

    public Task PublishAsync<T>(string name, T contentObj, string callbackName = null,
        CancellationToken cancellationToken = new CancellationToken())
    {
        return _capPublisher.PublishAsync(name, contentObj, callbackName, cancellationToken);
    }

    public Task PublishAsync<T>(string name, T contentObj, IDictionary<string, string> headers,
        CancellationToken cancellationToken = new CancellationToken())
    {
        return _capPublisher.PublishAsync(name, contentObj, headers, cancellationToken);
    }

    public void Publish<T>(string name, T contentObj, string callbackName = null)
    {
        _capPublisher.Publish(name, contentObj, callbackName);
    }

    public void Publish<T>(string name, T contentObj, IDictionary<string, string> headers)
    {
        _capPublisher.Publish(name, contentObj, headers);
    }

    public Task PublishAsync<T>(string name, Guid tenantId, T contentObj, string callbackName = null,
        CancellationToken cancellationToken = default)
    {
        var header = new Dictionary<string, string>
        {
            {Headers.CallbackName, callbackName},
            {EventBusHeaders.TenantId, tenantId.ToString()}
        };

        return _capPublisher.PublishAsync(name, contentObj, header, cancellationToken);
    }

    public Task PublishAsync<T>(string name, Guid tenantId, T contentObj, IDictionary<string, string> headers,
        CancellationToken cancellationToken = default)
    {
        if (headers == null) headers = new Dictionary<string, string>();

        headers.Add(EventBusHeaders.TenantId, tenantId.ToString());

        return _capPublisher.PublishAsync(name, contentObj, headers, cancellationToken);
    }

    public void Publish<T>(string name, Guid tenantId, T contentObj, string callbackName = null)
    {
        var header = new Dictionary<string, string>
        {
            {Headers.CallbackName, callbackName},
            {EventBusHeaders.TenantId, tenantId.ToString()}
        };

        _capPublisher.Publish(name, contentObj, header);
    }

    public void Publish<T>(string name, Guid tenantId, T contentObj, IDictionary<string, string> headers)
    {
        if (headers == null) headers = new Dictionary<string, string>();

        headers.Add(EventBusHeaders.TenantId, tenantId.ToString());

        _capPublisher.Publish(name, contentObj, headers);
    }
}