Masstransit IReceiveObserver can not get message body

Question

I am trying to get a specific info from all the messages consumed by my service, and I do not want to repeat the same code inside each consumer I have.

So I created a class that implements IReceiveObserver and inside public Task PreReceive(ReceiveContext context) I am trying to get the message body, except that I could not.

  public Task PreReceive(ReceiveContext context)
        {
            var body = context.GetBody();
            return null;
        }

I am assuming that body will contain the message body, so I am trying to deserialize it using:

   public static object DeserializeFromStream(MemoryStream stream)
        {
            IFormatter formatter = new BinaryFormatter();
            stream.Seek(0, SeekOrigin.Begin);
            object o = formatter.Deserialize(stream);
            return o;
        }
var Deserialized =(GeneralMessageType) DeserializeFromStream((MemoryStream)body);

I am getting the following error:

System.Runtime.Serialization.SerializationException: 'The input stream is not a valid binary format.

Is this the correct way of achieving what I need? I think I am following the right approach but I have an issue in extracting the message body.

Update:

Message body using Console.WriteLine("{0}", (new StreamReader(stream)).ReadToEnd());:

{\r\n \"messageId\": \"01110000-5d1b-0015-b7c9-08d5479e73b2\",\r\n \"correlationId\": \"da197981-a10e-43a9-a53b-67b7a3261511\",\r\n \"conversationId\": \"01110000-5d1b-0015-80d3-08d5479e73a6\",\r\n \"sourceAddress\": \"queue URL",\r\n \"destinationAddress\": \"receiverURL",\r\n \"messageType\": [\r\n \"urn:message:MessageType"\r\n ],\r\n \"message\": {\r\n \"correlationId\": \"da197981-a10e-43a9-a53b-67b7a3261511\",\r\n \"Id\": \"62d2bfff-7e20-4b51-b431-7d46e44abfc0\",\r\n \"Name\": \"'''yry'''\",\r\n \"Code\": \"ryryyy\",\r\n \"isActive\": true,\r\n \"username\": \"username\",\r\n \"password\": \"*****\",\r\n \"isUserActive\": true\r\n },\r\n \"headers\": {},\r\n \"host\": {\r\n \"machineName\": \"myMachine\",\r\n
\"processName\": \"w3wp\",\r\n \"processId\": 4568,\r\n
\"assembly\": \"MassTransit\",\r\n \"assemblyVersion\": \"3.5.7.1082\",\r\n \"frameworkVersion\": \"4.0.30319.42000\",\r\n \"massTransitVersion\": \"3.5.7.1082\",\r\n
\"operatingSystemVersion\": \"Microsoft Windows NT 10.0.16299.0\"\r\n }\r\n}"

Answer 1

Figured it out:

As you can see in question body context.GetBody(); gives a json containing Host info, Headers, and Message body in addition to a few other attributes.

What I did is removing "\r\n"s and replacing \" with " and then going to https://jsonutils.com/, paste the resulting Json and it will convert it to the classes you should deserialize to.

The thing is body contains much more info than the message sent, all of them added by masstransit and rabbitmq.

 public class Message
{
    //Message attributes
}

public class Headers
{
}

public class Host
{
    public string machineName { get; set; }
    public string processName { get; set; }
    public int processId { get; set; }
    public string assembly { get; set; }
    public string assemblyVersion { get; set; }
    public string frameworkVersion { get; set; }
    public string massTransitVersion { get; set; }
    public string operatingSystemVersion { get; set; }
}

public class TotalMessage
{
    public string messageId { get; set; }
    public string correlationId { get; set; }
    public string conversationId { get; set; }
    public string sourceAddress { get; set; }
    public string destinationAddress { get; set; }
    public IList<string> messageType { get; set; }
    public Message message { get; set; }
    public Headers headers { get; set; }
    public Host host { get; set; }
}

for deserializing you can use:

  public static object DeserializeFromStream(MemoryStream stream)
        {
            IFormatter formatter = new BinaryFormatter();
            stream.Seek(0, SeekOrigin.Begin);
            object o = formatter.Deserialize(stream);
            return o;
        }

 var body = context.GetBody();
 var totalMessage= JsonDeserializer<TotalMessage>((MemoryStream)body);

To get the message body: totalMessage.message

Update:

Of course you can deserialize to only the message body by something like:

public class TotalMessage
{
    public Message message { get; set; }
}

Answer 2

So you do not need to bother with the BinaryFormatter. Instead you should use System.Runtime.Serialization.DataContractJsonSerializer as the content of your stream is coming through as JSON. You can do that with the following code:

DataContractJsonSerializer serializer = new DataContractJsonSerializer(typeof(MassTransit));
MassTransit obj = (MassTransit)serializer.ReadObject(stream);

This is assuming the object you are trying to deserialize is of the type MassTransit. If it is not, then simply replace that with whatever class you are trying to deserialize an instance of.

You could make this a generic method like this:

public static T JsonDeserializer<T>(MemoryStream strm)
{
    DataContractJsonSerializer serializer = new DataContractJsonSerializer(typeof(T));
    return (T)serializer.ReadObject(strm);
}

And simply call that wherever you want to deserialize an object from a JSON stream.