Implement Issue #389 : Support for MessagePack #471
Conversation
Codecov Report
@@ Coverage Diff @@
## master #471 +/- ##
============================================
+ Coverage 86.91% 87.09% +0.17%
- Complexity 303 356 +53
============================================
Files 20 24 +4
Lines 1292 1495 +203
Branches 134 164 +30
============================================
+ Hits 1123 1302 +179
- Misses 111 127 +16
- Partials 58 66 +8
Continue to review full report at Codecov.
|
|
@lxhoan nice work here! I'll take a look later. |
|
@lxhoan i need some time as well. |
|
@lxhoan I've nothing against reformatting the code but could you please avoid adding it to the same PR you are creating a new feature or fixing a bug? Thanks! |
There was a problem hiding this comment.
I could not find the UnitTest for the most imortant class in this PR: MessagePackTraverser responsible for creating the QueryResult and I think there are tests that should be removed (unless MessagePack support may have collateral effects on jitter/flush duration/etc.).
| public QueryResult convert(final ResponseBody value) throws IOException { | ||
| try (InputStream is = value.byteStream()) { | ||
| MessagePackTraverser traverser = new MessagePackTraverser(); | ||
| for (QueryResult queryResult : traverser.traverse(is)) { |
There was a problem hiding this comment.
I didn't understand why you are traversing the InputStream but returning when the first element is retrieved.
There was a problem hiding this comment.
In general, there should be only one QueryResult so I return the first one. However this code can look ambiguous. I refactor MessagePackTraverser to expose more method for this logic. also add javadocs to explain
| */ | ||
| @RunWith(JUnitPlatform.class) | ||
| @EnabledIfEnvironmentVariable(named = "INFLUXDB_VERSION", matches = "1\\.6|1\\.5|1\\.4") | ||
| public class MessagePackBatchOptionsTest extends BatchOptionsTest { |
There was a problem hiding this comment.
Why this class is necessary? How MessagePack support can affect jitter or how messages are flushed?
There was a problem hiding this comment.
No, it does not. The reason to extends the tests for MessagePack is due to these 2 differences in MessagePack serializer (in comparison to JSON serializer) at server side :
- Time in MessagePack is encoded as a 64 bit Long value (JSON as String)
- Empty array is encoded as it is (JSON as empty)
| this.adapter = adapter; | ||
|
|
||
| if (ResponseFormat.MSGPACK.equals(responseFormat)) { | ||
| String[] versionNumbers = version().split("\\."); |
There was a problem hiding this comment.
This should go into its own method which can be then tested more easily.
| } | ||
| if (httpUrl == null) { | ||
| throw new IllegalArgumentException("Unable to parse url: " + url); | ||
| } |
There was a problem hiding this comment.
Please revert formatting changes as they make reviewing more difficult.
| .bufferLimit(batchOptions.getBufferLimit()) | ||
| .consistencyLevel(batchOptions.getConsistency()) | ||
| .build(); | ||
| this.batchProcessor = BatchProcessor.builder(this).actions(batchOptions.getActions()) |
There was a problem hiding this comment.
same here, formatting with no changes in behavior.
| this.batchEnabled.set(true); | ||
| return this; | ||
| } | ||
|
|
||
| @Override | ||
| public InfluxDB enableBatch(final int actions, final int flushDuration, | ||
| final TimeUnit flushDurationTimeUnit) { | ||
| public InfluxDB enableBatch(final int actions, final int flushDuration, final TimeUnit flushDurationTimeUnit) { |
| enableBatch(actions, flushDuration, flushDurationTimeUnit, threadFactory, (points, throwable) -> { }); | ||
| public InfluxDB enableBatch(final int actions, final int flushDuration, final TimeUnit flushDurationTimeUnit, | ||
| final ThreadFactory threadFactory) { | ||
| enableBatch(actions, flushDuration, flushDurationTimeUnit, threadFactory, (points, throwable) -> { |
| } | ||
|
|
||
| /** | ||
| * {@inheritDoc} | ||
| */ | ||
| @Override | ||
| public List<String> describeDatabases() { | ||
| QueryResult result = execute(this.influxDBService.query(this.username, | ||
| this.password, SHOW_DATABASE_COMMAND_ENCODED)); | ||
| QueryResult result = execute( |
| call = this.influxDBService.postQuery(this.username, | ||
| this.password, query.getDatabase(), query.getCommandWithUrlEncoded(), | ||
| boundParameterQuery.getParameterJsonWithUrlEncoded()); | ||
| BoundParameterQuery boundParameterQuery = (BoundParameterQuery) query; |
| call = this.influxDBService.query(this.username, | ||
| this.password, query.getDatabase(), query.getCommandWithUrlEncoded()); | ||
| } | ||
| if (query.requiresPost()) { |
| if (datagramSocket != null && !datagramSocket.isClosed()) { | ||
| datagramSocket.close(); | ||
| } | ||
| if (datagramSocket != null && !datagramSocket.isClosed()) { |
| .append(duration) | ||
| .append(" REPLICATION ") | ||
| .append(replicationFactor); | ||
| queryBuilder.append(rpName).append("\" ON \"").append(database).append("\" DURATION ").append(duration) |
lxhoan
force-pushed
the
features/message-pack-8
branch
2 times, most recently
from
403b741
to
54b5e8d
Compare
pom.xml
Outdated
| @@ -73,6 +73,9 @@ | |||
| <exclude>docker-compose.yml</exclude> | |||
| </excludes> | |||
| </resource> | |||
| <resource> | |||
| <directory>src/test/resources</directory> | |||
There was a problem hiding this comment.
Why do we need to include test resources in the artifact ?
There was a problem hiding this comment.
that's a mistake, I fixed it, tks for this finding
refactor + add javadocs to make MessagePackTraverser more unambiguous
test for MessagePackTraverser
MessagePackBatchOptionsTest simply extends BatchOptionsTest
checking of version support at querying time
fix test/resources was specified wrongly
lxhoan
force-pushed
the
features/message-pack-8
branch
from
a1587de
to
4eaa28a
Compare
|
@lxhoan can you please add a meaningful changelog entry, thanks |
this implementation uses the messagepack-core only