opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-24 16:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #4326] Optimized some code style.

Browse Source 
[ISSUE #4326] Optimized some code style.
This commit is contained in:
Oliver
2022-05-19 14:27:41 +08:00
committed by GitHub
parent 455ac6d5ab
commit 20b3a9275c
6 changed files with 16 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java
+2
View File
@@ -447,10 +447,12 @@ public class DefaultLitePullConsumer extends ClientConfig implements LitePullCon
this.offsetStore = offsetStore;
}
@Override
public boolean isUnitMode() {
return unitMode;
}
@Override
public void setUnitMode(boolean isUnitMode) {
this.unitMode = isUnitMode;
}
client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumer.java
+2
View File
@@ -434,10 +434,12 @@ public class DefaultMQPullConsumer extends ClientConfig implements MQPullConsume
return defaultMQPullConsumerImpl;
}
@Override
public boolean isUnitMode() {
return unitMode;
}
@Override
public void setUnitMode(boolean isUnitMode) {
this.unitMode = isUnitMode;
}
client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java
+4 -2
View File
@@ -185,7 +185,7 @@ public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsume
* Consider the {@code pullBatchSize}, the instantaneous value may exceed the limit
*
* <p>
* The size of a message only measured by message body, so it's not accurate
* The size(MB) of a message only measured by message body, so it's not accurate
*/
private int pullThresholdSizeForQueue = 100;
@@ -417,7 +417,7 @@ public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsume
@Override
public void setUseTLS(boolean useTLS) {
super.setUseTLS(useTLS);
if (traceDispatcher != null && traceDispatcher instanceof AsyncTraceDispatcher) {
if (traceDispatcher instanceof AsyncTraceDispatcher) {
((AsyncTraceDispatcher) traceDispatcher).getTraceProducer().setUseTLS(useTLS);
}
}
@@ -860,10 +860,12 @@ public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsume
this.postSubscriptionWhenPull = postSubscriptionWhenPull;
}
@Override
public boolean isUnitMode() {
return unitMode;
}
@Override
public void setUnitMode(boolean isUnitMode) {
this.unitMode = isUnitMode;
}
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java
+4 -2
View File
@@ -367,8 +367,10 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
} catch (Exception e) {
log.error("parse subscription error", e);
}
ms.setSubVersion(0L);
result.add(ms);
if (ms != null) {
ms.setSubVersion(0L);
result.add(ms);
}
}
}
}
client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java
+2 -5
View File
@@ -38,9 +38,7 @@ import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageBatch;
import org.apache.rocketmq.common.message.MessageClientIDSetter;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageId;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.topic.TopicValidator;
@@ -264,7 +262,7 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
@Override
public void setUseTLS(boolean useTLS) {
super.setUseTLS(useTLS);
if (traceDispatcher != null && traceDispatcher instanceof AsyncTraceDispatcher) {
if (traceDispatcher instanceof AsyncTraceDispatcher) {
((AsyncTraceDispatcher) traceDispatcher).getTraceProducer().setUseTLS(useTLS);
}
}
@@ -896,9 +894,8 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
public MessageExt viewMessage(String topic,
String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
try {
MessageId oldMsgId = MessageDecoder.decodeMessageId(msgId);
return this.viewMessage(msgId);
} catch (Exception e) {
} catch (Exception ignored) {
}
return this.defaultMQProducerImpl.queryMessageByUniqKey(withNamespace(topic), msgId);
}
common/src/main/java/org/apache/rocketmq/common/CountDownLatch2.java
+2
View File
@@ -172,10 +172,12 @@ public class CountDownLatch2 {
return getState();
}
@Override
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
@Override
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
for (; ; ) {