Receive Event Grid from a Service Bus Handler on Azure Services

Question

I am sending the event on the event grid like. i can see they arriving at the azure dashboard

            NomeEmailChange yay = new NomeEmailChange
            {
                Nome = "cesco",
                Email = "cesco"
            };

            var primaryTopicKey = _config["EventGridConfig:AcessKey"];
            var primaryTopic = _config["EventGridConfig:Endpoint"];

            var primaryTopicHostname = new Uri(primaryTopic).Host;

            var topicCredentials = new TopicCredentials(primaryTopicKey);
            var client = new EventGridClient(topicCredentials);

            var id = Guid.NewGuid().ToString();
            var hey = new List<EventGridEvent>
            {
                new EventGridEvent()
                {
                    Id = id,
                    EventType = "cesco-cesco",
                    Data = (yay),
                    EventTime = DateTime.Now,
                    Subject = "MS_Clientes",
                    DataVersion = "1.0",
                }
            };
            ;
            client.PublishEventsAsync(primaryTopicHostname, hey);

then I created a event grid subscrition. i can confirm on the event grid messages arriving to the event grid subscription.

On another project I am subscribing to the service bus like the bellow. it works fine for consuming messages sent directly to the bus.

        public static IServiceCollection AddBus(this IServiceCollection services, IConfiguration configuration,
            IHostingEnvironment env)
        {
            services.AddMassTransit(x => { x.AddConsumer<NomeEmailChangeConsumer>(); });
            services.AddSingleton(provider => Bus.Factory.CreateUsingAzureServiceBus(cfg =>
            {
                var keyName = "RootManageSharedAccessKey";
                var busName = configuration["ServiceBus:Name"];
                var secret = configuration["ServiceBus:Secret"];
                var host = cfg.Host(
                    "Endpoint=sb://" + busName + ".servicebus.windows.net/;" +
                    "SharedAccessKeyName=" + keyName + ";" +
                    "SharedAccessKey=" + secret,
                    z =>
                    {
                        TokenProvider
                            .CreateSharedAccessSignatureTokenProvider(keyName, secret);
                    });
                cfg.UseExtensionsLogging(provider.GetService<ILoggerFactory>());
                cfg.ReceiveEndpoint(host, configuration["ServiceBus:Topic"],
                    e => { e.Consumer<NomeEmailChangeConsumer>(provider); });
            }));
            services.AddSingleton<IPublishEndpoint>(provider => provider.GetRequiredService<IBusControl>());
            services.AddSingleton<ISendEndpointProvider>(provider => provider.GetRequiredService<IBusControl>());
            services.AddSingleton<IBus>(provider => provider.GetRequiredService<IBusControl>());
            services.AddScoped(provider => provider.GetRequiredService<IBus>().CreateRequestClient<NomeEmailChange>());
            services.AddSingleton<IHostedService, BusService>();
            return services;
        }

all supposed to work by now, but on this other project when the message arrives I get the following error

fail: MassTransit.Messages[0]
      R-FAULT sb://sbacompanharreldev.servicebus.windows.net/bff-queue 9ade19ec-238c-4c08-8e03-28bac695ea7b No deserializer was registered for the message content type: application/json; charset=utf-8. Supported content types include application/vnd.masstransit+json, application/vnd.masstransit+bson, application/vnd.masstransit+xml
System.Runtime.Serialization.SerializationException: No deserializer was registered for the message content type: application/json; charset=utf-8. Supported content types include application/vnd.masstransit+json, application/vnd.masstransit+bson, application/vnd.masstransit+xml
   at MassTransit.Serialization.SupportedMessageDeserializers.Deserialize(ReceiveContext receiveContext)
   at MassTransit.Pipeline.Filters.DeserializeFilter.Send(ReceiveContext context, IPipe`1 next)
   at GreenPipes.Filters.RescueFilter`2.GreenPipes.IFilter<TContext>.Send(TContext context, IPipe`1 next)

https://stackoverflow.com/a/59195982/11228967 - feedback would be nice, to know if it works or not for you, as decent amout of time was invested to explore this problem. Thanks :) — kgalic, Dec 06 '19 at 22:07
@kgalic the deserializer you send me does not compile on this line eturn new JsonConsumeContext(JsonSerializer.CreateDefault(), receiveContext, messageEnv); — CESCO, Dec 07 '19 at 01:35
Did you try to download the sample from the repository I provided and try to run it? — kgalic, Dec 07 '19 at 01:36
I can try to run it once more but if you downloaded the repository it should work as it is based on the dotnet core. Can you please verify if you have the same package versions installed as in my example and repo? — kgalic, Dec 07 '19 at 02:04
MassTransit is an open-source project, please have a look here, and you can see that JsonConsumeContext has constructor only with 3 params: https://github.com/MassTransit/MassTransit/blob/6e75ca3ed77af800d216dec0edc8124f1c2b4589/src/MassTransit/Serialization/JsonConsumeContext.cs — kgalic, Dec 07 '19 at 03:35
Here is another example of how to use it(also 3 params), similar like in my deserializer: https://github.com/MassTransit/MassTransit/blob/3b98c2819fdddc519d4ad88e3873255013aa137d/src/MassTransit/Serialization/BsonMessageDeserializer.cs I also tried to run the code, and as in my screenshot(from the answer), I am receiving the messages through my deserializer. Could you please check if you can build and run my example, with your queue/topic connection string and queue/topic name? — kgalic, Dec 07 '19 at 03:38

kgalic · Answer 1 · 2019-12-05T13:49:41.413

As suggested you have to write the custom deserializer. In my implementation, I changed a couple of things on how to deal with the MassTransit package.

First, you have to register deserializers for different Content Types:

For example, if your message come from MassTransit publisher it will be the default message ContentType(in C#): JsonMessageSerializer.JsonContentType
If your message comes from the ServiceBus queue/topic, it is usually "application/json"

The following code shows how to register deserializers, and how to set up the receiving endpoint. One difference compared to your approach is that I used connection string, so no need for SAS token part.

class Program
    {
        static string ContentTypeJson = "application/json";
        static async Task Main(string[] args)
        {
            var bus = Bus.Factory.CreateUsingAzureServiceBus(cfg =>
            {
                var queueName = "Your SB Queue Name";
                var connectionString = "Connection String with RooTManage policy";
                var host = cfg.Host(connectionString, h =>
                {
                    h.OperationTimeout = TimeSpan.FromSeconds(60);
                });

                cfg.ReceiveEndpoint(queueName,
                    e =>
                    {
                        e.AddMessageDeserializer(contentType: new ContentType(ContentTypeJson), () =>
                        {
                            return new EventGridMessgeDeserializer(ContentTypeJson);
                        });
                        e.Consumer(() => new EventGridMessageConsumer());

                        // Uncomment if required deserializer for local messages - mass transit as publisher or direct messages from SB
                        //e.AddMessageDeserializer(contentType: JsonMessageSerializer.JsonContentType, () =>
                        //{
                        //    return new CustomMessageDeserializer(JsonMessageSerializer.JsonContentType.ToString());
                        //});
                        //e.Consumer(() => new MessageConsumer());
                    });

            });
            bus.Start();

            Console.WriteLine("Press any key to exit");
            // for testing purposes of local messages - mass transit as publisher
            // await bus.Publish<CustomMessage>(new {  Hello = "Hello, World." });
            await Task.Run(() => Console.ReadKey());

            await bus.StopAsync();

        }

I used my message simulator to pipe the messages to the EventGrid, which forwards the messages to the SB queue, and below you can see the outcome of running this code:

Deserializer for event grid messages, as well as full code, you can find on Github: https://github.com/kgalic/MassTransitSample and also in my answer on your other question.

score 0 · Answer 2 · answered Nov 29 '19 at 03:20

0

MassTransit is using message envelope that is used to deserialize the content. You will need to create and register a custom serializer to allow MassTransit ingest and process your messages. For more information, see Interoperability documentation. Once a custom serializer is created, it can be registered using configuration API.

answered Nov 29 '19 at 03:20

Sean Feldman

23,443
7
55
80

any example on how to do this ? – CESCO Nov 29 '19 at 03:57
I couldn't locate any samples within the documentation. Perhaps that's worth a separate question – Sean Feldman Dec 02 '19 at 03:22
And looks like you've [raised it](https://stackoverflow.com/questions/59107284/how-to-write-a-masstransit-json-deserializer-for-azure-services) already. – Sean Feldman Dec 02 '19 at 03:23
and I started a bounty there too! – CESCO Dec 02 '19 at 14:00
@CESCO I provided an example of how to write custom deserializer for EventGrid messages, as well as the full working solution as an example. please check your other question, as well as the answer on this one which explains how to incorporate deserializer with MassTransit setup. – kgalic Dec 05 '19 at 13:28
@CESCO did you have a chance to check out my sample? Any luck? – kgalic Dec 06 '19 at 15:08
Sorry have not. – Sean Feldman Dec 06 '19 at 15:19

Receive Event Grid from a Service Bus Handler on Azure Services

2 Answers2