franz1981 commented on a change in pull request #2975: ARTEMIS-2617 Improve
AMQP Journal loading
URL: https://github.com/apache/activemq-artemis/pull/2975#discussion_r377315963
##########
File path:
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
##########
@@ -1183,8 +1219,126 @@ public Object getCorrelationID() {
return this;
}
+ /**
+ *
https://en.wikipedia.org/wiki/Knuth%E2%80%93Morris%E2%80%93Pratt_algorithm jump
table
+ */
+ private static int[] createJumpTable(byte[] needle) {
+ final int[] jumpTable = new int[needle.length + 1];
+ int j = 0;
+ for (int i = 1; i < needle.length; i++) {
+ while (j > 0 && needle[j] != needle[i]) {
+ j = jumpTable[j];
+ }
+ if (needle[j] == needle[i]) {
+ j++;
+ }
+ jumpTable[i + 1] = j;
+ }
+ for (int i = 1; i < jumpTable.length; i++) {
+ if (jumpTable[i] != 0) {
+ return jumpTable;
+ }
+ }
+ // optimization over the original algorithm: it would save from
accessing any jump table
+ return null;
+ }
+
+ /**
+ *
https://en.wikipedia.org/wiki/Knuth%E2%80%93Morris%E2%80%93Pratt_algorithm
search algorithm
+ */
+ private static int indexOf(ReadableBuffer haystack, int start, int end,
byte[] needle, int[] jumpTable) {
+ final long length = end - start;
+ int j = 0;
+ final int needleLength = needle.length;
+ for (int i = 0; i < length; i++) {
+ final int index = start + i;
+ final byte value = haystack.get(index);
+ while (j > 0 && needle[j] != value) {
+ j = jumpTable == null ? 0 : jumpTable[j];
+ }
+ if (needle[j] == value) {
+ j++;
+ }
+ if (j == needleLength) {
+ final int startMatch = index - needleLength + 1;
+ return startMatch;
+ }
+ }
+ return -1;
+ }
+
+ @Override
+ public boolean hasScheduledDeliveryTime() {
+ if (scheduledTime >= 0) {
+ return true;
+ }
+ return anyMessageAnnotations(SCHEDULED_DELIVERY_SYMBOLS,
+ SCHEDULED_DELIVERY_ENCODED,
+ SCHEDULED_DELIVERY_JUMP_TABLES);
+ }
+
+ private static boolean isValidSymbolEncoding(Symbol[] symbols, byte[][]
encodedSymbols) {
+ for (int i = 0, size = symbols.length; i < size; i++) {
+ final Symbol symbol = symbols[i];
+ final byte[] encoded = encodedSymbols[i];
+ if
(!Arrays.equals(symbol.toString().getBytes(StandardCharsets.US_ASCII),
encoded)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private boolean anyMessageAnnotations(Symbol[] symbols, byte[][]
encodedSymbols, int[][] jumpTables) {
+ assert isValidSymbolEncoding(symbols, encodedSymbols);
+ final int count = symbols.length;
+ if (messageDataScanned == SCANNED) {
+ final MessageAnnotations messageAnnotations = this.messageAnnotations;
+ if (messageAnnotations == null) {
+ return false;
+ }
+ Map<Symbol, Object> map = messageAnnotations.getValue();
+ if (map == null) {
+ return false;
+ }
+ for (int i = 0; i < count; i++) {
+ if (map.containsKey(symbols[i])) {
+ return true;
+ }
+ }
+ return false;
+ }
+ // messageDataScanned != SCANNED;
+ final ReadableBuffer data = this.data;
+ DecoderImpl decoder = TLSEncode.getDecoder();
+ final int position = data.position();
+ decoder.setBuffer(data.rewind());
+ try {
+ while (data.hasRemaining()) {
+ TypeConstructor<?> constructor = decoder.readConstructor();
+ if (MessageAnnotations.class.equals(constructor.getTypeClass())) {
Review comment:
Great, tomorrow will add a commit to improve that + some tests.
Now my other concern is related to the lazely message scan: I've introduced
a whole new state in the message and I hope it won't mess with some earlier
assumptions now broken (routingType now force a message scan, while before it
was assuming to always being correctly populated)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
