Having difficulty getting RabbitMQ request response to work across process.

[jlenk]

I have been experimenting with Pub/Sub and that is working nicely - thanks. But I cannot seem to get a handler in another process to pick up the call to send. I have tried the sample code for webapi image handler modified for rabbitmq but I am still not successful.

[zarusz]

hey @jlenk, I have not tried that sample with RabbitMq transport for request-response.
Is the handler process not picking up the request message?
Is the .Send() not resolving in the requesting process?

I assume you've looked at this.

Perhaps we can refine the sample to include for Rabbit as an alternative transport. Happy to help here.

[jlenk]

Thanks for the reply. I have since gotten it to work :). I have attached my sample here for others to see. Thanks again.

// Producer App.cs
using Microsoft.Extensions.DependencyInjection;
using SharedModels;
using SlimMessageBus;
using SlimMessageBus.Host;
using SlimMessageBus.Host.RabbitMQ;
using SlimMessageBus.Host.Serialization.SystemTextJson;
using System.Reflection;

namespace ProducerApp;
class Program
{
    static async Task Main(string[] args)
    {
        // Configure DI  
        var services = new ServiceCollection();

        services.AddSlimMessageBus(mbb =>
        {
            mbb.WithProviderRabbitMQ(cfg =>
            {
                cfg.ConnectionString = "amqp://guest:[email protected]:5672/";
            });
            mbb.Produce<WeatherForecastRequest>(x =>
            {
                x.DefaultTopic("weather-request");
            })
            .ExpectRequestResponses(x =>
            {
                // Tell the handler to which exchange send the responses to
                x.ReplyToExchange("weather-response-exchange", ExchangeType.Fanout);
                // Which queue to use to read responses from
                x.Queue("weather-response-queue");
                // Timeout if the response doesn't arrive within 60 seconds
                x.DefaultTimeout(TimeSpan.FromSeconds(60));
            });

            mbb.AddJsonSerializer();
            mbb.AddServicesFromAssembly(Assembly.GetEntryAssembly());
            mbb.AddServicesFromAssemblyContaining<WeatherForecastRequest>();
        });

        services.AddLogging();

        var provider = services.BuildServiceProvider();
        var bus = provider.GetRequiredService<IMessageBus>();
        // Send a request  
        Console.WriteLine("Sending weather forecast request for London...");
        var response = await bus.Send(new WeatherForecastRequest("London"));
        Console.WriteLine($"Received response: {response.City}, {response.TemperatureC}°C, {response.Summary}");
    }
}

// ConsumerApp.cs
using Microsoft.Extensions.DependencyInjection;
using SharedModels;
using SlimMessageBus;
using SlimMessageBus.Host;
using SlimMessageBus.Host.RabbitMQ;
using SlimMessageBus.Host.Serialization.SystemTextJson;
using System.Reflection;
namespace ConsumerApp;
class Program
{
    static async Task Main(string[] args)
    {
        // Configure DI
        var services = new ServiceCollection();
        services.AddSlimMessageBus(mbb =>
        {
            mbb.WithProviderRabbitMQ(cfg =>
            {
                cfg.ConnectionString = "amqp://guest:[email protected]:5672/";
            })
            .Handle<WeatherForecastRequest, WeatherForecastResponse>(x => x
                .Queue("weather-request-queue")
                .Topic("weather-request")
                .ExchangeBinding("weather-request")
                .WithHandler<WeatherForecastHandler>()
            );
            mbb.AddJsonSerializer();
            mbb.AddServicesFromAssembly(Assembly.GetEntryAssembly());
        });

        services.AddLogging();
        var provider = services.BuildServiceProvider();
        var bus = provider.GetRequiredService<IMessageBus>();
        Console.WriteLine("Consumer started. Waiting for requests...");
        await Task.Delay(-1); // Keep running
    }
}
// Handler for the request
public class WeatherForecastHandler : IRequestHandler<WeatherForecastRequest, WeatherForecastResponse>
{
    public async Task<WeatherForecastResponse> OnHandle(WeatherForecastRequest request, CancellationToken cancellationToken)
    {
        Console.WriteLine($"Processing request for {request.City}");
        // Simulate some work
        await Task.Delay(1000);
        // Generate a sample response
        return new WeatherForecastResponse(request.City, Random.Shared.Next(-20, 55), Summaries[Random.Shared.Next(Summaries.Length)]);
    }
    private static readonly string[] Summaries = new[] { "Freezing", "Bracing", "Chilly", "Cool", "Mild", "Warm", "Balmy", "Hot", "Swelterin", "Scorching" };
}

// WeatherForcast.cs
using SlimMessageBus;

namespace SharedModels;
public record WeatherForecastRequest(string City) : IRequest<WeatherForecastResponse>;

// WeatherForcastResponse.cs
namespace SharedModels;
public record WeatherForecastResponse(string City, int TemperatureC, string Summary);

[zarusz]

Awesome! Could you do a contribution and update the sample to work for RabbitMq too? :)
The simple console app has an enum at the application start that allows users to set the messaging transport. Would be cool to have something similar like that to alternate between Kafka and Rabbit.

[jlenk]

I will try to get to this soon. Might be another week or so.

[zarusz]

Appreciate it, thanks!