Question: How can I proxy the raw message without serializing it?

[nspyke]

Firstly, thank you for creating this excellent package!
For the most part, I've found it easy to use and have got it running without too much hassle.

One question I have though is;
How can I proxy a raw message from an inbound channel directly onto an outbound channel?

I would like to write the raw string message data from a MQTT inbound channel directly on to a Kafka outbound channel. No serialization required - publish it out exactly as it comes in.
(I will be saving this message into a S3 bucket to use as part of a cold path analysis process.)

[BEagle1984]

There are multiple ways to achieve your goal.

1. Using the BinaryFileMessage

As the name suggests, this was meant to transfer binary files, but you can leverage it to publish any binary content including your UTF8 encoded string.

public class MyBinaryMessage : IBinaryFileMessage
{
    public MyBinaryMessage(byte[] content)
    {
        Content = new MemoryStream(content);
    }

    public Stream Content { get; }
}

public class MyService
{
    private readonly IPublisher _publisher;

    public MyService(IPublisher publisher)
    {
        _publisher = publisher;
    }

    public Task ProduceTestMessageAsync() => _publisher.PublishAsync(new MyBinaryMessage(Encoding.UTF8.GetBytes("your string message")));
}

You must of course configure the endpoints to route the IBinaryFileMessage/BinaryFileMessage (or your concrete implementation, MyBinaryMessage in the example) to the intended endpoint.
See also: https://silverback-messaging.net/concepts/broker/binary-files.html

2. Using a custom IMessageSerializer

You can write your own "serializer" and use a DTO just to transport your byte array.

public record MyBinaryMessage(byte[] Content);

public class CustomMessageSerializer : IMessageSerializer
{
    public bool RequireHeaders => false;

    public ValueTask<(object? Message, Type MessageType)> DeserializeAsync(Stream? messageStream, MessageHeaderCollection messageHeaders, MessageSerializationContext context) =>
        throw new NotSupportedException();

    public ValueTask<Stream?> SerializeAsync(object? message, MessageHeaderCollection messageHeaders, MessageSerializationContext context) =>
        return ValueTask.FromResult(new MemoryStream(((MyBinaryMessage)message).Content));
}

public class MyService
{
    private readonly IPublisher _publisher;
    
    public MyService(IPublisher publisher)
    {
        _publisher= publisher;
    }
    
    public Task ProduceTestMessageAsync() => _publisher.PublishAsync(new MyBinaryMessage(Encoding.UTF8.GetBytes("your string message")));
}

In the endpoint configuration, you use the SerializeUsing method (or the Serializer property if not using the fluent API) to inject your CustomMessageSerializer.

3. Using the IProducer directly

You could skip the mediator (IPublisher) and directly use a producer. In that case, you can just produce a byte array.

public class MyService
{
    private readonly IProducer _producer;
    
    public MyService(KafkaBroker kafkaBroker)
    {
        _producer = kafkaBroker.GetProducer("your-topic");
    }
    
    public Task ProduceTestMessageAsync() => _producer.ProduceAsync(Encoding.UTF8.GetBytes("your string message"));
}

You could probably also use the RawProduceAsync method to skip the entire Silverback pipeline and get a small performance boost if you aren't using any other feature and you really just want to produce your raw byte array. See also https://silverback-messaging.net/concepts/broker/producer.html.

[BEagle1984]

Since you are processing a message and you are therefore in a subscriber, you can also take advantage of the auto-publishing feature, when you chose either solution 1 or 2.

public class MySubscriber
{
    public MyBinaryMessage ProcessMqttMessage(YourMqttMessage message) => new MyBinaryMessage(Encoding.UTF8.GetBytes(...));
}

See also https://silverback-messaging.net/concepts/bus/subscribe.html#return-new-messages-republishing.

[nspyke]

Thanks @BEagle1984
Using a subclass of BinaryFileMessage, I have been able to get this working rather nicely. I didn't even require a subscriber, although I made one anyway in order to log the output.

For reference for anyone wondering how to do the same things

Model

using System.IO;
using Silverback.Messaging.Messages;

public class PlainMessage : BinaryFileMessage, IEvent
{
    public override string ToString()
    {
        return Content is {CanRead: true} 
            ? new StreamReader(Content).ReadToEnd() 
            : string.Empty;
    }
}

Inbound Endpoints Configurator

using MQTTnet.Formatter;
using MQTTnet.Protocol;
using Silverback.Messaging.Configuration;

public class MqttConfigurator : IEndpointsConfigurator
{
    public void Configure(IEndpointsConfigurationBuilder builder)
    {
        builder.AddMqttEndpoints(endpoints =>
            endpoints
                .Configure(config => config
                    .ConnectViaTcp("localhost", 8883)
                    .EnableTls()
                )
                .AddInbound(eb => eb
                    .ConsumeFrom("message/#")
                    .Configure(c => { c.ClientId = $"{_options.Value.ClientId}.{nameof(PlainMessage)}"; })
                    .ConsumeBinaryFiles()
                )
        );
    }
}

Subscriber

public class BinaryFileMessageSubscriber
{
    private readonly ILogger<BinaryFileMessageSubscriber> _logger;

    public BinaryFileMessageSubscriber(ILogger<BinaryFileMessageSubscriber> logger)
    {
        _logger = logger;
    }
    
    public PlainMessage? OnMessageReceived(IBinaryFileMessage message)
    {
            var m = new PlainMessage(message.Content, "application/json");
            _logger.LogInformation("BinaryMessage: {Message}", m.ToString());
            return m;
    }
}

Outbound Endpoints Configurator

public class KafkaConfigurator : IEndpointsConfigurator
{
    public void Configure(IEndpointsConfigurationBuilder builder)
    {
        builder
            .AddKafkaEndpoints(endpoints => endpoints
                .Configure(kcc => {  })
                .AddOutbound<PlainMessage>(obe =>
                    obe.ProduceTo("raw-messages")
                )
            );
    }
}