Parse inner JSON and split by it

[Prolog]

Hi, I have a JSON coming from RabbitMQ structured as shown below:

{
              "MessageItemsCount": 8106,
              "Message": "[{\"Id\":\"Foo_123_456\",\"NetworkShopId\":\"333\",\"Name\":\"Some product\",\"Description\":\"Blah blah\",\"Categories\":null,\"Brand\":null,\"PhotoUrl\":\"Some url\",\"ProductUrl\":\"Some url\",\"Price\":259.0,\"RetailPrice\":259.0,\"CurrencyCode\":\"USD\"}]",
      "NetworkName": "Foo"
    }

Message field contain JSON array of multiple products up to 10000. Fields MessageItemsCount and NetworkName are irrelevant to me. All I want is to parse the inner JSON kept in Message field and then split it to single messages so I can send them to elasticsearch as separate documents. If the particular product contain ProductId field then delete action is called instead of the default create or update.

What I've got so far:

input {    
    rabbitmq {
        host => "localhost"
        queue => "create_or_update_products"
        port => 15672
        heartbeat => 30
        durable => false
        user => "guest"
        password => "guest"    
    }
    rabbitmq {
        host => "localhost"
        queue => "delete_products"
        port => 15672
        heartbeat => 30
        durable => false
        user => "guest"
        password => "guest"    
    }
}

filter {
  json {
    source => "Message"
    target => "Message"
  }
  split {     
    field => "Message"
	target => "Message"
  }
}

output {
    if([ProductId]){
        elasticsearch {
            action => delete
            hosts => "localhost:9200"
            index => "products"
            document_id => "%{[ProductId]}"
        }
    }
    else{
        elasticsearch {
            hosts => "localhost:9200"
            index => "products"
            document_id => "%{[Id]}"
        }
    }
}

Unfortunately, it doesn't work. Thanks for help!

[Prolog]

Okay, I managed to get it almost 100% working.

filter {
  json {
    source => "Message"
	target => "Message"
	remove_field => [ "MessageItemsCount", "NetworkName" ]
  }
  split {
    field => "Message"
  }
  mutate {
	add_field => { 
		"Id" => "%{[Message][Id]}"
		"NetworkShopId" => "%{[Message][NetworkShopId]}"
		"Name" => "%{[Message][Name]}"
		"Description" => "%{[Message][Description]}"
		"Categories" => "%{[Message][Categories]}"
		"Brand" => "%{[Message][Brand]}"
		"PhotoUrl" => "%{[Message][PhotoUrl]}"
		"ProductUrl" => "%{[Message][ProductUrl]}"
		"Price" => "%{[Message][Price]}"
		"RetailPrice" => "%{[Message][RetailPrice]}"
		"CurrencyCode" => "%{[Message][CurrencyCode]}"
	}
	remove_field => [ "Message" ]
  }
}

But another problem occured. If Brand or Categories is null, then it's value is not null but %{[Message][Brand]}. How to preserve the null value?

[Prolog]

For null values I used if statement like this

  if [Message][CurrencyCode] {
    mutate {
	  add_field => { 
		"CurrencyCode" => "%{[Message][CurrencyCode]}"
	  }
	}
  }
  else {
    mutate {
	  add_field => { 
		"CurrencyCode" => ""
	  }
	}
  }