Masstransit: Replacing a Deserializer on a receive endpoint should not add to other endpoints

[phatboyg]

Yes, I know.

[slaneyrw]

Migrating from 3.5.7 to 5.0.2

Not sure if this is related, but the temporary request/response receive endpoint created doesn't have the Encrypted deserializer on it... I'm getting

"No deserializer was registered for the message content type: application/vnd.masstransit+aes. Supported content types include application/vnd.masstransit+json, application/vnd.masstransit+bson, application/vnd.masstransit+xml" messages published to the "..._error" queue.

It's successfully encrypting the message outbound, received and decrypted by the consumer so I know the request message is being encrypted.

If I drill down in the debugger I can see the receive filter chain on the bus has DeserializeFilter with a SupportedMessagesDeserializers type without an "application/vnd.masstransit+aes" factory entry

Bus.Factory.CreateUsingRabbitMq(configurer => { configurer.UseEncryptedSerializer( ... ); }

[phatboyg]

Are you using host.ConnectReceiveEndpoint in this case? And if so, which ISendEndpointProvider are you using to send the request, or are you publishing it? If so, which publish endpoint? Can you share a brief snippet of code creating your request?

[slaneyrw]

The Bus is created using

        IBusControl bus = MassTransit.Bus.Factory.CreateUsingRabbitMq(configurer =>
        {

            var host = configurer.Host(new Uri("..."), options =>
            {
                options.Username("...");
                options.Password("...");

            });

            var streamProvider = new Encryption.AesCryptoStreamProviderWithRandomIV(new MyKeyProvider());

            configurer.UseEncryptedSerializer(streamProvider);
        });

The bus is started using

            var busHandle = await bus.StartAsync();

The request/response uses CreateRequestClient, the bus above is injected into the proxy

    protected async Task<TResponse> QueryAsync<TMessage, TResponse>(TMessage message, TimeSpan timeout, CancellationToken cancellationToken = default(CancellationToken))
        where TMessage : class where TResponse : class
    {
        var requestClient = this.Bus.CreateRequestClient<TMessage>();
        TResponse response = default(TResponse);

        using (var handle = requestClient.Create(message, cancellationToken, timeout))
        {
            handle.UseExecute(context =>
            {
                context.Headers.Set(MassTransit.Serialization.EncryptedMessageSerializer.EncryptionKeyHeader, this.Options.EncryptionKeyName);
            });
            var responseHandle = await handle.GetResponse<TResponse>();
            response = responseHandle.Message;
        }

        return response;
    }

The consumer receives the encrypted msg ok, decrypts it and sends an encrypted response using the same EncyptionKeyId. I can see the content of the response that ends up in the *_error queue, which I can decrypt manually using the above cyptostream provider.

If I use the exact same proxy code using a non-encrypting consumer I get the response successfully.

The above code was working in 3.5.7 using the PublishRequest extension

        TResponse response = default(TResponse);
        Fault fault = default(Fault);

        try
        {
            Request<TMessage> request = await this.Bus.PublishRequest<TMessage>(
            message,
            context =>
            {
                context.Handle<TResponse>(
                    value =>
                    {
                        response = value.Message;
                        return Task.FromResult(0);
                    });

                context.Handle<Fault>(
                    value =>
                    {
                        fault = value.Message;
                        return Task.FromResult(0);
                    });
                context.Headers.Set(EncryptedMessageSerializer.EncryptionKeyHeader, this.ProxyOptions.EncryptionKeyName)
                }

            }, cancellationToken).ConfigureAwait(false);

            await request.Task;
        }

        if (fault != null)
        {
            throw new Exception(fault.Exceptions[0].Message);
        }

        if (response != null)
        {
            return response;
        }

[phatboyg]

Excellent details on how to reproduce it. I'll dig into it tonight.

[slaneyrw]

Your comment about the receive endpoint led me to dig into the RabbitMQ factory configurer.

Adding the deserializer factory to BOTH the busConfiguration and busEndpointConfiguration has resolved the issue.

I wrapped the existing RabbitMQConfigurator and re-implemented the IBusFactoryConfigurator.AddMessageDeserializer method to added the supplied deserializerFactory to the busEndpointConfiguration as well.

What the impact of this change is is well outside of my knowledge though!

public class RabbitMqBusFactoryConfigurator : global::MassTransit.RabbitMqTransport.Configurators.RabbitMqBusFactoryConfigurator, IBusFactoryConfigurator
{
    private readonly IRabbitMqEndpointConfiguration busEndpointConfiguration;

    public RabbitMqBusFactoryConfigurator(
        IRabbitMqBusConfiguration configuration, 
        IRabbitMqEndpointConfiguration busEndpointConfiguration) : base(configuration, busEndpointConfiguration)
    {
        this.busEndpointConfiguration = busEndpointConfiguration;
    }

    void IBusFactoryConfigurator.AddMessageDeserializer(System.Net.Mime.ContentType contentType, DeserializerFactory deserializerFactory)
    {
        base.AddMessageDeserializer(contentType, deserializerFactory);
        this.busEndpointConfiguration.Serialization.AddDeserializer(contentType, deserializerFactory);
    }
}

[phatboyg]

I think I have a way to make this work without requiring it to be double-entered. Testing it now.

[phatboyg]

@slaneyrw if you can try out the develop branch, see if it works for you.

[slaneyrw]

Can confirm response msgs are decrypted successfully.

Tested using both 5.1.0 stable and 5.1.0.1518-develop ( 5.0.2.1511 isn't on nuget)

Thanks for being so responsive

[phatboyg]

Great, will close.