Build a Streaming API with .NET Core and RabbitMQ for Log/Data Streaming

By Khaled Md Tuhidul Hossain · 2025-05-30 · Technology

Streaming logs and data in real time matters for monitoring, analytics, and responsive UIs. This post walks through building a streaming API using .NET Core and RabbitMQ that can handle high-throughput data streams efficiently.

Why .NET Core and RabbitMQ

.NET Core has good performance, runs cross-platform, and has built-in support for asynchronous programming, which fits streaming scenarios well. RabbitMQ handles message queuing with flexible routing, so you can distribute streaming data to multiple consumers without writing that plumbing yourself.

Architecture overview

Our streaming API will consist of:

Data producer: Receives logs/data and publishes to RabbitMQ
RabbitMQ broker: Handles message routing and queuing
Streaming API: Consumes from RabbitMQ and streams to clients via WebSockets/Server-Sent Events
Client applications: Subscribe to real-time data streams

Prerequisites needed

.NET Core 6.0 or later
RabbitMQ server
Visual Studio or VS Code
Basic knowledge of C# and message queuing concepts

Step 1: Setting up the project

First, create a new .NET Core Web API project and install the required NuGet packages:

dotnet new webapi -n StreamingAPI
cd StreamingAPI
dotnet add package RabbitMQ.Client
dotnet add package Microsoft.AspNetCore.SignalR
dotnet add package Newtonsoft.Json

Step 2: Configure RabbitMQ connection

Create a RabbitMQ service for managing connections:

// Services/IRabbitMQService.cs
public interface IRabbitMQService
{
    void PublishMessage(string queueName, object message);
    void StartConsuming(string queueName, Action<string> onMessageReceived);
    void StopConsuming();
}

// Services/RabbitMQService.cs
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using Newtonsoft.Json;
using System.Text;

public class RabbitMQService : IRabbitMQService, IDisposable
{
    private readonly IConnection _connection;
    private readonly IModel _channel;
    private readonly string _hostname;
    private readonly string _username;
    private readonly string _password;

    public RabbitMQService(IConfiguration configuration)
    {
        _hostname = configuration["RabbitMQ:Hostname"] ?? "localhost";
        _username = configuration["RabbitMQ:Username"] ?? "guest";
        _password = configuration["RabbitMQ:Password"] ?? "guest";

        var factory = new ConnectionFactory()
        {
            HostName = _hostname,
            UserName = _username,
            Password = _password
        };

        _connection = factory.CreateConnection();
        _channel = _connection.CreateModel();
    }

    public void PublishMessage(string queueName, object message)
    {
        _channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false);

        var json = JsonConvert.SerializeObject(message);
        var body = Encoding.UTF8.GetBytes(json);

        var properties = _channel.CreateBasicProperties();
        properties.Persistent = true;

        _channel.BasicPublish(exchange: "", routingKey: queueName, basicProperties: properties, body: body);
    }

    public void StartConsuming(string queueName, Action<string> onMessageReceived)
    {
        _channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false);

        var consumer = new EventingBasicConsumer(_channel);
        consumer.Received += (model, ea) =>
        {
            var body = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);
            onMessageReceived(message);
            _channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
        };

        _channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
    }

    public void StopConsuming()
    {
        _channel?.Close();
    }

    public void Dispose()
    {
        _channel?.Dispose();
        _connection?.Dispose();
    }
}

Step 3: Create data models

Define models for your streaming data:

// Models/LogEntry.cs
public class LogEntry
{
    public string Id { get; set; } = Guid.NewGuid().ToString();
    public DateTime Timestamp { get; set; } = DateTime.UtcNow;
    public string Level { get; set; } = "INFO";
    public string Message { get; set; } = string.Empty;
    public string Source { get; set; } = string.Empty;
    public Dictionary<string, object> Properties { get; set; } = new();
}

// Models/StreamData.cs
public class StreamData
{
    public string Type { get; set; } = string.Empty;
    public object Data { get; set; } = new();
    public DateTime Timestamp { get; set; } = DateTime.UtcNow;
}

Step 4: Implement SignalR hub for real-time streaming

Create a SignalR hub to stream data to connected clients:

// Hubs/StreamingHub.cs
using Microsoft.AspNetCore.SignalR;

public class StreamingHub : Hub
{
    public async Task JoinGroup(string groupName)
    {
        await Groups.AddToGroupAsync(Context.ConnectionId, groupName);
        await Clients.Caller.SendAsync("JoinedGroup", groupName);
    }

    public async Task LeaveGroup(string groupName)
    {
        await Groups.RemoveFromGroupAsync(Context.ConnectionId, groupName);
        await Clients.Caller.SendAsync("LeftGroup", groupName);
    }

    public override async Task OnDisconnectedAsync(Exception? exception)
    {
        await base.OnDisconnectedAsync(exception);
    }
}

Step 5: Create the streaming service

Implement a background service that consumes from RabbitMQ and streams to clients:

// Services/StreamingBackgroundService.cs
using Microsoft.AspNetCore.SignalR;
using Newtonsoft.Json;

public class StreamingBackgroundService : BackgroundService
{
    private readonly IRabbitMQService _rabbitMQService;
    private readonly IHubContext<StreamingHub> _hubContext;
    private readonly ILogger<StreamingBackgroundService> _logger;

    public StreamingBackgroundService(
        IRabbitMQService rabbitMQService,
        IHubContext<StreamingHub> hubContext,
        ILogger<StreamingBackgroundService> logger)
    {
        _rabbitMQService = rabbitMQService;
        _hubContext = hubContext;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("Streaming Background Service started");

        _rabbitMQService.StartConsuming("logs", async (message) =>
        {
            try
            {
                var logEntry = JsonConvert.DeserializeObject<LogEntry>(message);
                if (logEntry != null)
                {
                    await _hubContext.Clients.Group("logs")
                        .SendAsync("ReceiveLogEntry", logEntry, stoppingToken);
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error processing log message: {Message}", message);
            }
        });

        _rabbitMQService.StartConsuming("data", async (message) =>
        {
            try
            {
                var streamData = JsonConvert.DeserializeObject<StreamData>(message);
                if (streamData != null)
                {
                    await _hubContext.Clients.Group("data")
                        .SendAsync("ReceiveStreamData", streamData, stoppingToken);
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error processing data message: {Message}", message);
            }
        });

        await Task.Delay(Timeout.Infinite, stoppingToken);
    }

    public override async Task StopAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("Streaming Background Service stopping");
        _rabbitMQService.StopConsuming();
        await base.StopAsync(cancellationToken);
    }
}

Step 6: Create API controllers

Add controllers for publishing data to the streams:

// Controllers/LogsController.cs
[ApiController]
[Route("api/[controller]")]
public class LogsController : ControllerBase
{
    private readonly IRabbitMQService _rabbitMQService;
    private readonly ILogger<LogsController> _logger;

    public LogsController(IRabbitMQService rabbitMQService, ILogger<LogsController> logger)
    {
        _rabbitMQService = rabbitMQService;
        _logger = logger;
    }

    [HttpPost]
    public IActionResult PublishLog([FromBody] LogEntry logEntry)
    {
        try
        {
            _rabbitMQService.PublishMessage("logs", logEntry);
            return Ok(new { Message = "Log published successfully", LogId = logEntry.Id });
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error publishing log entry");
            return StatusCode(500, new { Error = "Failed to publish log entry" });
        }
    }

    [HttpPost("batch")]
    public IActionResult PublishLogBatch([FromBody] List<LogEntry> logEntries)
    {
        try
        {
            foreach (var logEntry in logEntries)
            {
                _rabbitMQService.PublishMessage("logs", logEntry);
            }
            return Ok(new { Message = quot;{logEntries.Count} logs published successfully" });
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error publishing log batch");
            return StatusCode(500, new { Error = "Failed to publish log batch" });
        }
    }
}

// Controllers/DataController.cs
[ApiController]
[Route("api/[controller]")]
public class DataController : ControllerBase
{
    private readonly IRabbitMQService _rabbitMQService;
    private readonly ILogger<DataController> _logger;

    public DataController(IRabbitMQService rabbitMQService, ILogger<DataController> logger)
    {
        _rabbitMQService = rabbitMQService;
        _logger = logger;
    }

    [HttpPost("stream")]
    public IActionResult PublishStreamData([FromBody] StreamData streamData)
    {
        try
        {
            _rabbitMQService.PublishMessage("data", streamData);
            return Ok(new { Message = "Data published successfully" });
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error publishing stream data");
            return StatusCode(500, new { Error = "Failed to publish stream data" });
        }
    }
}

Step 7: Configure services in Program.cs

Wire up all the services in your Program.cs file:

// Program.cs
using StreamingAPI.Services;
using StreamingAPI.Hubs;

var builder = WebApplication.CreateBuilder(args);

// Add services
builder.Services.AddControllers();
builder.Services.AddSignalR();
builder.Services.AddSingleton<IRabbitMQService, RabbitMQService>();
builder.Services.AddHostedService<StreamingBackgroundService>();

// Add CORS
builder.Services.AddCors(options =>
{
    options.AddPolicy("AllowAll", policy =>
    {
        policy.AllowAnyOrigin()
              .AllowAnyMethod()
              .AllowAnyHeader();
    });
});

builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();

var app = builder.Build();

// Configure pipeline
if (app.Environment.IsDevelopment())
{
    app.UseSwagger();
    app.UseSwaggerUI();
}

app.UseCors("AllowAll");
app.UseHttpsRedirection();
app.UseAuthorization();
app.MapControllers();
app.MapHub<StreamingHub>("/streaminghub");

app.Run();

Step 8: Add configuration

Add RabbitMQ configuration to appsettings.json:

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft.AspNetCore": "Warning"
    }
  },
  "AllowedHosts": "*",
  "RabbitMQ": {
    "Hostname": "localhost",
    "Username": "guest",
    "Password": "guest"
  }
}

Step 9: Client implementation example

Here's a simple JavaScript client to consume the streams:

// Include SignalR client library:
// <script src="https://cdnjs.cloudflare.com/ajax/libs/microsoft-signalr/6.0.1/signalr.min.js"></script>

const connection = new signalR.HubConnectionBuilder()
    .withUrl("https://localhost:7000/streaminghub")
    .build();

connection.start().then(function () {
    // Join groups
    connection.invoke("JoinGroup", "logs");
    connection.invoke("JoinGroup", "data");
});

// Handle log entries
connection.on("ReceiveLogEntry", function (logEntry) {
    const logsDiv = document.getElementById("logs");
    const logElement = document.createElement("div");
    logElement.textContent =
        `${logEntry.timestamp} [${logEntry.level}] ${logEntry.message}`;
    logsDiv.appendChild(logElement);
});

// Handle stream data
connection.on("ReceiveStreamData", function (streamData) {
    const dataDiv = document.getElementById("data");
    const dataElement = document.createElement("div");
    dataElement.textContent =
        `${streamData.timestamp} [${streamData.type}] ${JSON.stringify(streamData.data)}`;
    dataDiv.appendChild(dataElement);
});

Testing the API locally

To test your streaming API:

Start RabbitMQ: Ensure RabbitMQ server is running
Run the API: dotnet run
Open the client: Load the HTML client in a browser
Send test data: Use Postman or curl to POST to /api/logs or /api/data/stream

Performance considerations

Connection pooling: Use connection pooling for RabbitMQ connections in high-throughput scenarios
Message batching: Batch messages when possible to reduce overhead
Memory management: Implement proper disposal patterns for resources
Error handling: Add retry logic for transient failures
Monitoring: Add health checks for both RabbitMQ and SignalR connections

Wrapping up

That covers a working streaming API built with .NET Core and RabbitMQ. The background service consumes messages from RabbitMQ and pushes them to connected clients through SignalR, so you get real-time delivery without polling.

From here you could add authentication, message filtering, persistence to a database, or more advanced RabbitMQ routing depending on what your use case needs.