Enable accessing Message in TableView

[petermnhull]

Fixes #1176

Motivation

The current TableView implementation only stores the payload encoded to the SchemaValueType of each incoming message. There are many use cases where accessing the full metadata of the pulsar.Message type is needed, so this PR refactors the map storage of messages to be the pulsar.Message received from the consumer rather than the payload.

All the existing APIs are maintained by performing the payload encoding when the methods (e.g. Get) are called.

Two new public methods, Message and Messages, are introduced to access the pulsar.Message for each key so that accompanying metadata for a payload can be easily retrieved.

Modifications

• The TableView implementation handleMessage method now stores the pulsar.Message type and only encodes to the SchemaValueType for listeners.
• Get and Entries use the SchemaValueType when called to convert from pulsar.Message to the payload
• Two public methods, Message and Messages, introduced to get full metadata.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests, and is covered by existing tests, and can be verified as follows:

• Running table_view_test.go which runs new and existing tests for all functionality.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): No
• The public API: Yes, new Message and Messages methods on TableView interface. Existing API not impacted
• The schema: No
• The default values of configurations: No
• The wire protocol: No

Documentation

• Does this pull request introduce a new feature? Yes
• If yes, how is the feature documented? Code documentation, and happy to create new issue for other required documentation.

[RobertIndie]

This PR invokes new API changes. We need to draft a PIP for it: https://github.com/apache/pulsar/tree/master/pip

[merlimat]

This is returning the same underlying map, allowing for race-conditions and modifications from the application.

Note: the existing code is also wrong because it was returning tv.data instead of data