CloudEventDataMapper

[slinkydeveloper]

This PR introduces an interface to perform the mapping of CloudEventData and integrates it in the read process.

Now the user, when reading an event, can provide a mapper that converts the parsed CloudEventData, in whatever shape it is, to the desired data payload. The mapper is consistently invoked both when messages are in structured mode or binary mode: in this way, users define their own CloudEventData, along with CloudEventDataMapper, and they can consistently access to the event payloads, no matter from which "medium" the event came from.
When message is transcoded from one message to another, without going through Event (look at MessageReader#visit method), the mapper is not invoked.

Because we reuse CloudEventData, the user can eventually develop optimizations in this mapper (e.g. if data is instance of JsonCloudEventData, then faster mapping to my pojo, etc).

Also introduces CloudEventBuilder#from static method to simplify copying events.

Among the other usability issues, this fixes #238, in the sense that now users can finally provide a mapper in the kafka CloudEventDeserializer to map data to their business object (e.g. CloudEventDataMapper can implement the conversion data in avro format -> business object)

Signed-off-by: Francesco Guardiani francescoguard@gmail.com

[slinkydeveloper]

A sample usage:

    CloudEventDataMapper mapper = data -> {
        // If it's a json, then optimize the mapping
        if (data instanceof JsonCloudEventData && ((JsonCloudEventData) data).getNode().isInt()) {
            return new MyCloudEventData(((JsonCloudEventData) data).getNode().asInt());
        }
        return MyCloudEventData.fromStringBytes(data.toBytes());
    };
    HttpResponse<Buffer> vertxResponse = ...
    CloudEvent event = VertxMessageFactory.createReader(vertxResponse).toEvent(mapper);
    int myInt = ((MyCloudEventData)event.getData()).getValue()

I think then we can add, in another PR, a mapper that integrates with jackson, so the user can just do something like:

CloudEvent event = VertxMessageFactory.createReader(vertxResponse).toEvent(JacksonDataMapper.map(typeRef))
MyPojo myInt = (MyPojo)(((JacksonPojoCloudEventData)event.getData()).getValue())

[johanhaleby]

LGTM! 😃

From what I can tell this solves the problems I ran into when implementing Occurrent (issue #196) and also the other problems that have been discussed in various issues.

Thanks for your efforts @slinkydeveloper :)