Transports RabbitMQ

RabbitMQ Transport

Use RabbitMQ as a message transport for distributed data mapping with robust routing and reliability features.

Installation

dotnet add package FxMap.RabbitMq

Or via Package Manager Console:

Install-Package FxMap.RabbitMq

Client Configuration

Configure RabbitMQ transport in your client application to fetch data from remote services:

Basic Configuration (Local Testing)

using FxMap;
using FxMap.RabbitMq;
 
var builder = WebApplication.CreateBuilder(args);
 
builder.Services.AddFxMap(cfg =>
{
    cfg.AddEntitiesFromAssemblyContaining<SomeEntityAssemblyMarker>();
    cfg.AddRabbitMq(config => config.Host("localhost", "/"));
});
 
var app = builder.Build();
app.Run();

Full Configuration with Credentials

builder.Services.AddFxMap(cfg =>
{
    cfg.AddEntitiesFromAssemblyContaining<SomeEntityAssemblyMarker>();
    cfg.AddRabbitMq(config => config.Host("localhost", "/", 5672, c =>
    {
        c.UserName("your-username");
        c.Password("your-password");
    }));
});

Production Configuration

builder.Services.AddFxMap(cfg =>
{
    cfg.AddEntitiesFromAssemblyContaining<SomeEntityAssemblyMarker>();
    cfg.AddRabbitMq(config => config.Host("rabbitmq.production.com", "/", 5672, c =>
    {
        c.UserName(builder.Configuration["RabbitMQ:Username"]);
        c.Password(builder.Configuration["RabbitMQ:Password"]);
        c.UseSsl(s =>
        {
            s.Enabled = true;
            s.ServerName = "rabbitmq.production.com";
        });
    }));
});

Service Implementation

On the service side, implement IDistributedMapper to handle incoming mapping requests:

using FxMap;
using Microsoft.EntityFrameworkCore;
 
public class UserService : IDistributedMapper
{
    private readonly ApplicationDbContext _dbContext;
 
    public UserService(ApplicationDbContext dbContext)
    {
        _dbContext = dbContext;
    }
 
    public async Task<List<Dictionary<string, object?>>> HandleMappingRequest(
        DataMappingRequest request)
    {
        // request.Ids contains the list of IDs to fetch
        // request.Expression contains the fields to return
 
        var users = await _dbContext.Users
            .Where(u => request.Ids.Contains(u.Id))
            .ToListAsync();
 
        // Map to dictionary format
        return users.Select(u => new Dictionary<string, object?>
        {
            ["Id"] = u.Id,
            ["Name"] = u.Name,
            ["Email"] = u.Email
        }).ToList();
    }
}
 
// Register in Program.cs
builder.Services.AddScoped<IDistributedMapper, UserService>();
 
builder.Services.AddFxMap(cfg =>
{
    cfg.AddEntitiesFromAssemblyContaining<SomeEntityAssemblyMarker>();
    cfg.AddRabbitMq(config => config.Host("localhost", "/"));
});

Complete Example

1. Define Distributed Key & Entity Config

// Distributed key (shared contract)
public sealed class UserDistributedKey : IDistributedKey;
 
// Entity config on the data-provider side
public class UserConfig : EntityConfigureOf<User>
{
    protected override void Configure()
    {
        Id(x => x.Id);
        DefaultProperty(x => x.Name);
        UseDistributedKey<UserDistributedKey>();
    }
}

2. Client-Side Usage

// DTO
public class OrderResponse
{
    public string Id { get; set; }
    public string UserId { get; set; }
    public decimal Amount { get; set; }
    public UserDto User { get; set; }
}
 
// Profile
public class OrderResponseProfile : ProfileOf<OrderResponse>
{
    protected override void Configure()
    {
        UseDistributedKey<UserDistributedKey>()
            .Of(x => x.UserId)
            .For(x => x.User, "{ Id, Name, Email }");
    }
}
 
public class UserDto
{
    public string Id { get; set; }
    public string Name { get; set; }
    public string Email { get; set; }
}
 
// Controller
[ApiController]
[Route("api/[controller]")]
public class OrdersController : ControllerBase
{
    private readonly IDistributedMapper _mapper;
    private readonly OrderDbContext _dbContext;
 
    public OrdersController(IDistributedMapper mapper, OrderDbContext dbContext)
    {
        _mapper = mapper;
        _dbContext = dbContext;
    }
 
    [HttpGet("{id}")]
    public async Task<ActionResult<OrderResponse>> GetOrder(string id)
    {
        var order = await _dbContext.Orders.FindAsync(id);
        if (order == null) return NotFound();
 
        var response = new OrderResponse
        {
            Id = order.Id,
            UserId = order.UserId,
            Amount = order.Amount
        };
 
        // FxMap automatically fetches User data via RabbitMQ
        await _mapper.MapDataAsync(response);
 
        return Ok(response);
    }
}

3. Service-Side Implementation

// User service that handles RabbitMQ requests
public class UserService : IDistributedMapper
{
    private readonly UserDbContext _dbContext;
    private readonly ILogger<UserService> _logger;
 
    public UserService(UserDbContext dbContext, ILogger<UserService> logger)
    {
        _dbContext = dbContext;
        _logger = logger;
    }
 
    public async Task<List<Dictionary<string, object?>>> HandleMappingRequest(
        DataMappingRequest request)
    {
        _logger.LogInformation(
            "Handling mapping request for {Count} user IDs",
            request.Ids.Count);
 
        var users = await _dbContext.Users
            .Where(u => request.Ids.Contains(u.Id))
            .ToListAsync();
 
        return users.Select(u => new Dictionary<string, object?>
        {
            ["Id"] = u.Id,
            ["Name"] = u.Name,
            ["Email"] = u.Email
        }).ToList();
    }
}
 
// Program.cs for User service
var builder = WebApplication.CreateBuilder(args);
 
builder.Services.AddDbContext<UserDbContext>(options =>
    options.UseSqlServer(builder.Configuration.GetConnectionString("DefaultConnection")));
 
builder.Services.AddScoped<IDistributedMapper, UserService>();
 
builder.Services.AddFxMap(cfg =>
{
    cfg.AddEntitiesFromAssemblyContaining<SomeEntityAssemblyMarker>();
    cfg.AddRabbitMq(config => config.Host("localhost", "/"));
});
 
var app = builder.Build();
app.Run();

Key Advantages

Reliable Message Delivery

RabbitMQ ensures messages are delivered reliably with acknowledgments and persistent queues.

Flexible Routing

Advanced routing capabilities with exchanges, queues, and bindings for complex message patterns.

Battle-Tested

Proven in production environments worldwide with extensive tooling and monitoring support.

Management UI

Built-in web-based management interface for monitoring queues, connections, and message flows.

Advanced Features

Message Timeout Configuration

builder.Services.AddFxMap(cfg =>
{
    cfg.AddEntitiesFromAssemblyContaining<SomeEntityAssemblyMarker>();
    cfg.AddRabbitMq(config =>
    {
        config.Host("localhost", "/");
        config.RequestTimeout(TimeSpan.FromSeconds(30));
    });
});

Connection Retry Policy

builder.Services.AddFxMap(cfg =>
{
    cfg.AddEntitiesFromAssemblyContaining<SomeEntityAssemblyMarker>();
    cfg.AddRabbitMq(config =>
    {
        config.Host("localhost", "/");
        config.ConnectionRetryCount(5);
        config.ConnectionRetryInterval(TimeSpan.FromSeconds(3));
    });
});

Prefetch Count

builder.Services.AddFxMap(cfg =>
{
    cfg.AddEntitiesFromAssemblyContaining<SomeEntityAssemblyMarker>();
    cfg.AddRabbitMq(config =>
    {
        config.Host("localhost", "/");
        config.PrefetchCount(100); // Number of messages to prefetch
    });
});

Important Notes

• Ensure RabbitMQ server is running and accessible from all services
• Use the same RabbitMQ connection configuration across all services
• All services must have matching FxMap package versions
• Configure appropriate message timeouts based on your data complexity

---

Source: https://fxmapper.net/docs/transports/rabbitmq/ — the website is always the most up to date version.