You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2022/06/01 10:18:51 UTC
[GitHub] [rocketmq] RongtongJin commented on a diff in pull request #4399: [Summer of code] Fix controller bugs
RongtongJin commented on code in PR #4399:
URL: https://github.com/apache/rocketmq/pull/4399#discussion_r886613113
##########
store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAConnection.java:
##########
@@ -277,58 +282,56 @@ class HAServerReader extends AbstractHAReader {
@Override
protected boolean processReadResult(ByteBuffer byteBufferRead) {
while (true) {
+ boolean processSuccess = true;
int readSocketPos = byteBufferRead.position();
int diff = byteBufferRead.position() - ReadSocketService.this.processPosition;
- if (diff >= 4) {
+ if (diff >= Math.min(AutoSwitchHAClient.HANDSHAKE_HEADER_SIZE, AutoSwitchHAClient.TRANSFER_HEADER_SIZE)) {
int readPosition = ReadSocketService.this.processPosition;
HAConnectionState slaveState = HAConnectionState.values()[byteBufferRead.getInt(readPosition)];
switch (slaveState) {
- case HANDSHAKE:
- if (diff >= AutoSwitchHAClient.HANDSHAKE_HEADER_SIZE) {
- // AddressLength
- int addressLength = byteBufferRead.getInt(readPosition + AutoSwitchHAClient.HANDSHAKE_HEADER_SIZE - 4);
- if (diff < AutoSwitchHAClient.HANDSHAKE_HEADER_SIZE + addressLength) {
- break;
- }
- // Flag(isSyncFromLastFile)
- short syncFromLastFileFlag = byteBufferRead.getShort(readPosition + AutoSwitchHAClient.HANDSHAKE_HEADER_SIZE - 8);
- if (syncFromLastFileFlag == 1) {
- AutoSwitchHAConnection.this.isSyncFromLastFile = true;
- }
- // Flag(isAsyncLearner role)
- short isAsyncLearner = byteBufferRead.getShort(readPosition + AutoSwitchHAClient.HANDSHAKE_HEADER_SIZE - 6);
- if (isAsyncLearner == 1) {
- AutoSwitchHAConnection.this.isAsyncLearner = true;
- }
- // Address
- final byte[] addressData = new byte[addressLength];
- byteBufferRead.position(readPosition + AutoSwitchHAClient.HANDSHAKE_HEADER_SIZE);
- byteBufferRead.get(addressData);
- AutoSwitchHAConnection.this.slaveAddress = new String(addressData);
-
- isSlaveSendHandshake = true;
- byteBufferRead.position(readSocketPos);
- ReadSocketService.this.processPosition += AutoSwitchHAClient.HANDSHAKE_HEADER_SIZE + addressLength;
- LOGGER.info("Receive slave handshake, slaveId:{}, slaveAddress:{}, isSyncFromLastFile:{}, isAsyncLearner:{}",
- AutoSwitchHAConnection.this.slaveId, AutoSwitchHAConnection.this.slaveAddress,
- AutoSwitchHAConnection.this.isSyncFromLastFile, AutoSwitchHAConnection.this.isAsyncLearner);
+ case HANDSHAKE: {
Review Comment:
不需要大括号
--
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.
To unsubscribe, e-mail: dev-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org