Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,12 @@ public void streamIn(int size, ByteBufferInputStream bbis) throws IOException {
while (size > 0) {
OptionHeader header = new OptionHeader();
header.streamIn(bbis);

int totalSize =
header.packed()
? OptionHeader.HEADER_SIZE
: header.words() * Protocol.WORD_SIZE;

switch (header.type()) {
case SUB_QUEUE_INFOS:
if (subQueueInfosOption == null) {
Expand All @@ -51,32 +57,29 @@ public void streamIn(int size, ByteBufferInputStream bbis) throws IOException {
logger.debug("New options: {}", subQueueInfosOption);
} else {
logger.warn("Multiple SubQueueInfos option: {}", header.type());

int numSkip =
header.packed()
? 0
: header.words() * Protocol.WORD_SIZE
- OptionHeader.HEADER_SIZE;

bbis.skip(numSkip);

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

numSkip is equivalent to totalSize - OptionHeader.HEADER_SIZE (in this PR).
I wanted to have exactly 1 expression so I refactored it to use totalSize.

bbis.skip(totalSize - OptionHeader.HEADER_SIZE);
}
break;
case SUB_QUEUE_IDS_OLD:
// Only SUB_QUEUE_INFOS options may be packed.
assert !header.packed();
if (subQueueIdsOption == null) {
subQueueIdsOption = new SubQueueIdsOption(header);
subQueueIdsOption.streamIn(bbis);
logger.debug("Old options: {}", subQueueIdsOption);
} else {
logger.warn("Multiple SubQueueIds option: {}", header.type());
bbis.skip(header.words() * Protocol.WORD_SIZE - OptionHeader.HEADER_SIZE);
bbis.skip(totalSize - OptionHeader.HEADER_SIZE);
}
break;
default:
// Only SUB_QUEUE_INFOS options may be packed.
assert !header.packed();
logger.warn("Unsupported option type: {}", header.type());
bbis.skip(header.words() * Protocol.WORD_SIZE - OptionHeader.HEADER_SIZE);
bbis.skip(totalSize - OptionHeader.HEADER_SIZE);
break;
}
size -= header.words() * Protocol.WORD_SIZE;

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This size decrement was fully wrong for packed options header.

size -= totalSize;
}

if (subQueueIdsOption != null && subQueueInfosOption != null) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,44 @@ void testStreamInInfos() throws IOException {
assertEquals(0, bbis.available());
}

@Test
void testStreamInPackedInfosThenIds() throws IOException {
// A *packed* SUB_QUEUE_INFOS option occupies exactly one word (4 bytes): the
// header only, with no option content. Its 'words' field is reinterpreted as
// the RDA counter, so 'size' must be decremented by HEADER_SIZE, not by
// words()*WORD_SIZE. If a second option follows, an incorrect decrement causes
// the loop to exit early and the trailing option to be misread/dropped.
ByteBuffer bb =
ByteBuffer.wrap(
new byte[] {
// Packed Infos: type=SUB_QUEUE_INFOS(3), packed=1,
// words field reused as RDA counter (5)
0b00001110, 0b00000000, 0b00000000, 0b00000101, // header only

// Ids
0b00000100, 0b00000000, 0b00000000, 0b00000010, // header
0b00000000, 0b00000000, 0b00000000, 0b00000111, // subqueue id 7
});
ByteBufferInputStream bbis = new ByteBufferInputStream(bb);

Options options = new Options();
options.streamIn(12, bbis);

logger.info("Options: {}", options);

assertNotNull(options.subQueueInfosOption());
assertArrayEquals(
new Integer[] {com.bloomberg.bmq.impl.QueueId.k_DEFAULT_SUBQUEUE_ID},
options.subQueueInfosOption().subQueueIds());

// The option after the packed one must still be parsed.
assertNotNull(options.subQueueIdsOption());
assertArrayEquals(new Integer[] {7}, options.subQueueIdsOption().subQueueIds());

// Check that input stream is empty
assertEquals(0, bbis.available());
}

@Test
void testStreamInInfosIds() throws IOException {
ByteBuffer bb =
Expand Down
Loading