Receive data from Masstransit Consumer

If you've worked with the Masstransit library, you’ve likely used its Send or Publish methods. In this approach, you send data and register a Consumer so that when data is sent, the Consumer receives it and performs the necessary processing. However, with this method, you cannot receive a response from Masstransit. Essentially, this was suitable for operations that didn’t require a response. But if you want to receive a response from the Consumer , you need to use Requests. The Masstransit library provides a feature called Requests that allows you to fetch the required data from the Consumer.

In this article, we’ll demonstrate how to implement a Consumer that returns a user’s posts. To do this, you first need to install the Masstransit library and register its services:

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<GetUserPostsConsumer>()
        .Endpoint(e => e.Name = "user-posts");

    x.AddRequestClient<GetUserPostsData>(new Uri("exchange:user-posts"));

    x.UsingInMemory((context, cfg) =>
    {
        cfg.ConfigureEndpoints(context);
    });
});

Note: The Consumer's Endpoint name (e.g., user-posts) must also be specified in AddRequestClient after the exchange.

The GetUserPostsData class serves as the input data for the Consumer.

 The GetUserPostsConsumer class handles the request and returns the user’s posts (response types are not explicitly declared here).

public class GetUserPostsConsumer : IConsumer<GetUserPostsData>
{
    private readonly List<UserPosts> _userPosts = new()
    {
        new UserPosts { Id = 1, UserId = 1, Name = "John Doe", Title = "First Post" },
        new UserPosts { Id = 2, UserId = 1,  Name = "John Doe", Title = "Second Post" },
        new UserPosts { Id = 3, UserId = 2,  Name = "Jane Smith", Title = "Hello World" }
    };

    public async Task Consume(ConsumeContext<GetUserPostsData> context)
    {
        var userPosts = _userPosts.Where(p => p.UserId == context.Message.UserId).ToList();

        if (userPosts != null && userPosts.Any())
        {
            await context.RespondAsync(new GetUserPostsResult
            {
                UserPosts = userPosts
            });
        }
        else
        {
            await context.RespondAsync(new GetUserPostNotFoundResult { });
        }
    }
}

For simplicity, we've used an in-memory list. The Consumer's input is the GetUserPostsData class, which only contains a single property called UserId.

public class GetUserPostsData
{
    public int UserId { get; set; }
}

Then, based on the  UserId, the posts are retrieved from the usersPosts list. If the user has posts, we use the RespondAsync method to return a GetUserPostsResult response containing the user's post list. However, if the user has no posts, we return a GetUserPostNotFoundResult response instead. This demonstrates how we can support multiple response types.

Next, we'll implement the client class that will receive and process the user's post data:

public class MassTransitPostClient : IPostClient
{
    private readonly IRequestClient<GetUserPostsData> _requestClient;

    public MassTransitPostClient(IRequestClient<GetUserPostsData> requestClient)
    {
        _requestClient = requestClient;
    }

    public async Task<List<UserPosts>> GetUserPostsAsync(int userId)
    {
        var response = await _requestClient.GetResponse<GetUserPostsResult, GetUserPostNotFoundResult>(new GetUserPostsData
        {
            UserId = userId
        });
        if(response.Is(out Response<GetUserPostsResult> userPost))
        {
            return userPost.Message.UserPosts;
        }
        else if (response.Is(out Response<GetUserPostNotFoundResult> userPostNotFound))
        {
            return null;
        }
        return null;
    }
}

First, you need to inject the registered IRequestClient (which was registered during Masstransit setup) to be able to use it. Then, in your GetUserPostsAsync method, you should call the GetResponse method while passing the GetUserPostsData class containing the  UserId.

The GetResponse method can handle multiple response classes. Just as we could return multiple response types from the Consumer, we need to specify the possible response types when making the call.

We can then use the Is method to determine which response type was received from the Consumer:

response.Is(out Response<GetUserPostsResult> userPost)

If the Consumer has populated the GetUserPostsResult, you can retrieve the user's post list data. Otherwise, you'll receive a GetUserPostNotFoundResult response.

You can implement your internal project services this way, using Masstransit's Request/Response pattern instead of traditional API calls.

Case Study:
In one of our projects where we implemented this feature, we approached it as follows. For instance, when Service A needed to retrieve data from Service B, we first implemented the consumer in Service B:

x.AddConsumer<ExampleConsumer>().Endpoint(e => e.Name = "b-service");

Then in Project A, we registered the request client to fetch data from Service B:

x.AddRequestClient<GetExampleData>(new Uri("exchange:b-service"));

Later, we discovered another service (Service C) needed the same data. The only change required was registering the same AddRequestClient in Service C, allowing it to retrieve the required data directly from Service B.

Powered by Froala Editor