alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-638 Improved variable naming

Browse Source
This commit is contained in:
Jiri Pechanec 2018-03-22 13:20:49 +01:00 committed by Gunnar Morling
parent ce5196dc3e
commit 4051a4a439
1 changed files with 15 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/connection/wal2json/StreamingWal2JsonMessageDecoder.java
View File

@ -83,7 +83,12 @@ public class StreamingWal2JsonMessageDecoder implements MessageDecoder {
private final DateTimeFormat dateTime = DateTimeFormat.get();
private boolean containsMetadata = false;
private boolean messageInProgress = false;
private String bufferedContent;
/**
* To identify if the last current chunk is the last one we can send the current one
* for processing only after we read the next one or the end of message fragment.
*/
private String currentChunk;
private int txId;
@ -112,31 +117,32 @@ public void processMessage(ByteBuffer buffer, ReplicationMessageProcessor proces
timestamp = message.getString("timestamp");
commitTime = dateTime.systemTimestamp(timestamp);
messageInProgress = true;
bufferedContent = null;
currentChunk = null;
}
else {
// We are receiving changes in chunks
if (content.startsWith("{")) {
// First change, this is a valid JSON
bufferedContent = content;
currentChunk = content;
}
else if (content.startsWith(",")) {
// following changes, they have an extra comma at the start of message
doProcessMessage(processor, typeRegistry, bufferedContent, false);
bufferedContent = content.substring(1);
doProcessMessage(processor, typeRegistry, currentChunk, false);
currentChunk = content.substring(1);
}
else if (content.startsWith("]")) {
// No more changes
if (bufferedContent != null) {
doProcessMessage(processor, typeRegistry, bufferedContent, true);
if (currentChunk != null) {
doProcessMessage(processor, typeRegistry, currentChunk, true);
}
messageInProgress = false;
}
else {
throw new ConnectException("Chunk arrived in unxepected state");
throw new ConnectException("Chunk arrived in unexpected state");
}
}
} catch (final IOException e) {
}
catch (final IOException e) {
throw new ConnectException(e);
}
}