Cfrgtkky

I am trying to integrate Spring Cloud streams and publishing a custom Java Object across services with RabbitMQ as broker. The object I am publishing looks like:

public class AppMessageEnvelope implements Serializable {

  ...

  private Object messageBody;

  private Date sentAt = new Date();

  ...



  // setters and getters

}

This is just a wrapper object and the original object is put in messageBody. The object I am putting in messageBody looks like:

public class Job {

    ...

    private String message;

    private Map<MyEnum, String> myMap;

    ...

}

Note that both AppMessageEnvelope and Job are in a different model project which is imported as a Maven dependency in the publisher and subscriber Spring Boot projects, so models are exactly the same.

In producer, I publish the object as:

@EnableBinding(Source.class)

public class JobDistributor {



    private final Source jobQueue;



    @Autowired

    public JobDistributor(Source jobQueue) {

        this.jobQueue = jobQueue;

    }



    public AppMessageEnvelope publishJob(AppMessageEnvelope message) {

        LOG.info("Sending message: {}.", message);

        jobQueue.output().send(MessageBuilder.withPayload(message).build());

        return message;

    }

}

In consumer, I get the message as:

@Component

@EnableBinding(Sink.class)

public class JobConsumer {



    private final JobManager jobManager;

    private final ObjectMapper objectMapper;



    @Autowired

    public JobConsumer(

            JobManager jobManager, ObjectMapper objectMapper) {

        this.jobManager = jobManager;

        this.objectMapper = objectMapper;

    }



    @StreamListener(target = Sink.INPUT)

    public void processData(AppMessageEnvelope messageEnvelope) {

        LOG.info("Envelope received: {}.", messageEnvelope);



        try {

            TypeReference<Job> mapType = new TypeReference<Job>() {};

            Job job = objectMapper.readValue(messageEnvelope.getMessageBody().toString(), mapType);

            jobManager.processRequest(job);

        } catch (Exception ex) {

            LOG.error("Couldn't convert to correct object for processing: {}.", ex);

        }

    }

}

I try to use TypeReference to convert the internal object into a correct object but I get an error as:

JobConsumer - Couldn't convert to correct object for processing: {}.

    com.fasterxml.jackson.core.JsonParseException: Unexpected character ('i' (code 105)): was expecting double-quote to start field name

    at [Source: (StringReader); line: 1, column: 3]

Before the message is converted, I log it:

JobConsumer - Envelope received: AppMessageEnvelope{..., messageBody={id=5bf3a7302dbe9c7cf9927c60, jobId=8c0bfcb0b21248e694b5cd52337a1f9e, submittedAt=2018-11-20T06:18:24+0000, lastUpdatedOn=null, message=null, ..., fileContentMap={FILE_BYTES=JVBERi0xLjUKJb/3ov}}, sentAt=Tue Nov 20 11:48:24 IST 2018}

I tried configuring ObjectMapper as:

@Autowired

private ObjectMapper objectMapper() {

    JsonFactory factory = new JsonFactory();

    factory.enable(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES);

    return new ObjectMapper(factory);

}

I tried enabling unquoting fields with this too:

objectMapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true);

I tried solutions provided by this blog and some similar SO problem but nothing solved. What am I missing?