You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by yu...@apache.org on 2017/04/24 07:08:29 UTC
incubator-rocketmq git commit: OpenMessaging code reformat.
Repository: incubator-rocketmq
Updated Branches:
refs/heads/openmessaging-impl 6edeb8317 -> b7ec41213
OpenMessaging code reformat.
Project: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/commit/b7ec4121
Tree: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/tree/b7ec4121
Diff: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/diff/b7ec4121
Branch: refs/heads/openmessaging-impl
Commit: b7ec41213b5b7acf0109e41dc253e09cfdd4b4ba
Parents: 6edeb83
Author: yukon <yu...@apache.org>
Authored: Mon Apr 24 15:08:15 2017 +0800
Committer: yukon <yu...@apache.org>
Committed: Mon Apr 24 15:08:15 2017 +0800
----------------------------------------------------------------------
.../example/openmessaging/SimpleProducer.java | 18 ++++++++++--------
.../example/openmessaging/SimplePullConsumer.java | 6 +++---
.../example/openmessaging/SimplePushConsumer.java | 6 +++---
.../rocketmq/consumer/LocalMessageCache.java | 1 -
.../rocketmq/consumer/PushConsumerImpl.java | 6 +++---
.../rocketmq/producer/AbstractOMSProducer.java | 2 +-
.../rocketmq/producer/ProducerImpl.java | 2 +-
.../rocketmq/producer/SequenceProducerImpl.java | 2 +-
.../rocketmq/promise/DefaultPromise.java | 2 +-
.../rocketmq/promise/FutureState.java | 12 +++++++++---
.../openmessaging/rocketmq/utils/BeanUtils.java | 6 +++---
11 files changed, 35 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b7ec4121/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimpleProducer.java
----------------------------------------------------------------------
diff --git a/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimpleProducer.java b/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimpleProducer.java
index ac7f7c4..9d162ac 100644
--- a/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimpleProducer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimpleProducer.java
@@ -33,10 +33,10 @@ public class SimpleProducer {
final Producer producer = messagingAccessPoint.createProducer();
messagingAccessPoint.startup();
- System.out.println("messagingAccessPoint startup OK");
+ System.out.printf("MessagingAccessPoint startup OK%n");
producer.startup();
- System.out.println("producer startup OK");
+ System.out.printf("Producer startup OK%n");
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
@Override
@@ -50,25 +50,27 @@ public class SimpleProducer {
Message message = producer.createBytesMessageToTopic("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8")));
SendResult sendResult = producer.send(message);
//final Void aVoid = result.get(3000L);
- System.out.println("send async message OK, msgId: " + sendResult.messageId());
+ System.out.printf("Send async message OK, msgId: %s%n", sendResult.messageId());
}
{
final Promise<SendResult> result = producer.sendAsync(producer.createBytesMessageToTopic("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
result.addListener(new PromiseListener<SendResult>() {
- @Override public void operationCompleted(Promise<SendResult> promise) {
- System.out.println("Send async message OK, msgId: " + promise.get().messageId());
+ @Override
+ public void operationCompleted(Promise<SendResult> promise) {
+ System.out.printf("Send async message OK, msgId: %s%n", promise.get().messageId());
}
- @Override public void operationFailed(Promise<SendResult> promise) {
- System.out.println("send async message Failed, error: " + promise.getThrowable().getMessage());
+ @Override
+ public void operationFailed(Promise<SendResult> promise) {
+ System.out.printf("Send async message Failed, error: %s%n", promise.getThrowable().getMessage());
}
});
}
{
producer.sendOneway(producer.createBytesMessageToTopic("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
- System.out.println("Send oneway message OK");
+ System.out.printf("Send oneway message OK%n");
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b7ec4121/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePullConsumer.java
----------------------------------------------------------------------
diff --git a/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePullConsumer.java b/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePullConsumer.java
index 36b6b1d..8e06772 100644
--- a/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePullConsumer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePullConsumer.java
@@ -33,7 +33,7 @@ public class SimplePullConsumer {
OMS.newKeyValue().put(NonStandardKeys.CONSUMER_GROUP, "OMS_CONSUMER"));
messagingAccessPoint.startup();
- System.out.println("messagingAccessPoint startup OK");
+ System.out.printf("MessagingAccessPoint startup OK%n");
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
@Override
@@ -44,13 +44,13 @@ public class SimplePullConsumer {
}));
consumer.startup();
- System.out.println("consumer startup OK");
+ System.out.printf("Consumer startup OK%n");
while (true) {
Message message = consumer.poll();
if (message != null) {
String msgId = message.headers().getString(MessageHeader.MESSAGE_ID);
- System.out.println("Received one message: " + msgId);
+ System.out.printf("Received one message: %s%n", msgId);
consumer.ack(msgId);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b7ec4121/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePushConsumer.java
----------------------------------------------------------------------
diff --git a/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePushConsumer.java b/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePushConsumer.java
index 84c1b15..b0935d4 100644
--- a/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePushConsumer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePushConsumer.java
@@ -35,7 +35,7 @@ public class SimplePushConsumer {
createPushConsumer(OMS.newKeyValue().put(NonStandardKeys.CONSUMER_GROUP, "OMS_CONSUMER"));
messagingAccessPoint.startup();
- System.out.println("messagingAccessPoint startup OK");
+ System.out.printf("MessagingAccessPoint startup OK%n");
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
@Override
@@ -48,12 +48,12 @@ public class SimplePushConsumer {
consumer.attachQueue("OMS_HELLO_TOPIC", new MessageListener() {
@Override
public void onMessage(final Message message, final ReceivedMessageContext context) {
- System.out.println("Received one message: " + message.headers().getString(MessageHeader.MESSAGE_ID));
+ System.out.printf("Received one message: %s%n", message.headers().getString(MessageHeader.MESSAGE_ID));
context.ack();
}
});
consumer.startup();
- System.out.println("consumer startup OK");
+ System.out.printf("Consumer startup OK%n");
}
}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b7ec4121/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/LocalMessageCache.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/LocalMessageCache.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/LocalMessageCache.java
index 4dfdca6..90f9e03 100644
--- a/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/LocalMessageCache.java
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/LocalMessageCache.java
@@ -175,7 +175,6 @@ class LocalMessageCache implements ServiceLifecycle {
try {
if (!msgTreeMap.isEmpty()) {
msg = msgTreeMap.firstEntry().getValue();
- System.out.println(msg);
if (System.currentTimeMillis() - Long.parseLong(MessageAccessor.getConsumeStartTimeStamp(msg))
> clientConfig.getRmqMessageConsumeTimeout() * 60 * 1000) {
//Expired, ack and remove it.
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b7ec4121/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PushConsumerImpl.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PushConsumerImpl.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PushConsumerImpl.java
index a2694e4..f9b8058 100644
--- a/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PushConsumerImpl.java
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PushConsumerImpl.java
@@ -25,9 +25,9 @@ import io.openmessaging.PushConsumer;
import io.openmessaging.ReceivedMessageContext;
import io.openmessaging.exception.OMSRuntimeException;
import io.openmessaging.rocketmq.config.ClientConfig;
+import io.openmessaging.rocketmq.domain.NonStandardKeys;
import io.openmessaging.rocketmq.utils.BeanUtils;
import io.openmessaging.rocketmq.utils.OMSUtil;
-import io.openmessaging.rocketmq.domain.NonStandardKeys;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -47,7 +47,6 @@ public class PushConsumerImpl implements PushConsumer {
private final Map<String, MessageListener> subscribeTable = new ConcurrentHashMap<>();
private final ClientConfig clientConfig;
-
public PushConsumerImpl(final KeyValue properties) {
this.rocketmqPushConsumer = new DefaultMQPushConsumer();
this.properties = properties;
@@ -130,7 +129,8 @@ public class PushConsumerImpl implements PushConsumer {
class MessageListenerImpl implements MessageListenerConcurrently {
@Override
- public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> rmqMsgList, ConsumeConcurrentlyContext contextRMQ) {
+ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> rmqMsgList,
+ ConsumeConcurrentlyContext contextRMQ) {
MessageExt rmqMsg = rmqMsgList.get(0);
BytesMessage omsMsg = OMSUtil.msgConvert(rmqMsg);
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b7ec4121/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/AbstractOMSProducer.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/AbstractOMSProducer.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/AbstractOMSProducer.java
index 0b2db4d..8246bcd 100644
--- a/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/AbstractOMSProducer.java
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/AbstractOMSProducer.java
@@ -41,7 +41,7 @@ import org.slf4j.Logger;
import static io.openmessaging.rocketmq.utils.OMSUtil.buildInstanceName;
-abstract class AbstractOMSProducer implements ServiceLifecycle, MessageFactory{
+abstract class AbstractOMSProducer implements ServiceLifecycle, MessageFactory {
final static Logger log = ClientLogger.getLog();
final KeyValue properties;
final DefaultMQProducer rocketmqProducer;
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b7ec4121/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/ProducerImpl.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/ProducerImpl.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/ProducerImpl.java
index f644e7d..2c00c60 100644
--- a/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/ProducerImpl.java
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/ProducerImpl.java
@@ -25,8 +25,8 @@ import io.openmessaging.Promise;
import io.openmessaging.PropertyKeys;
import io.openmessaging.SendResult;
import io.openmessaging.exception.OMSRuntimeException;
-import io.openmessaging.rocketmq.utils.OMSUtil;
import io.openmessaging.rocketmq.promise.DefaultPromise;
+import io.openmessaging.rocketmq.utils.OMSUtil;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendStatus;
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b7ec4121/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/SequenceProducerImpl.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/SequenceProducerImpl.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/SequenceProducerImpl.java
index f03826e..05225cc 100644
--- a/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/SequenceProducerImpl.java
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/SequenceProducerImpl.java
@@ -78,7 +78,7 @@ public class SequenceProducerImpl extends AbstractOMSProducer implements Sequenc
try {
SendResult sendResult = this.rocketmqProducer.send(rmqMessages);
- String [] msgIdArray = sendResult.getMsgId().split(",");
+ String[] msgIdArray = sendResult.getMsgId().split(",");
for (int i = 0; i < messages.size(); i++) {
Message message = messages.get(i);
message.headers().put(MessageHeader.MESSAGE_ID, msgIdArray[i]);
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b7ec4121/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java
index 3e4bd26..c863ccf 100644
--- a/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java
@@ -77,7 +77,7 @@ public class DefaultPromise<V> implements Promise<V> {
} else {
long waitTime = timeout - (System.currentTimeMillis() - createTime);
if (waitTime > 0) {
- for (; ; ) {
+ for (;; ) {
try {
lock.wait(waitTime);
} catch (InterruptedException e) {
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b7ec4121/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/FutureState.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/FutureState.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/FutureState.java
index 9e2f69c..84b6c2d 100644
--- a/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/FutureState.java
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/FutureState.java
@@ -18,11 +18,17 @@
package io.openmessaging.rocketmq.promise;
public enum FutureState {
- /** the task is doing **/
+ /**
+ * the task is doing
+ **/
DOING(0),
- /** the task is done **/
+ /**
+ * the task is done
+ **/
DONE(1),
- /** ths task is cancelled **/
+ /**
+ * ths task is cancelled
+ **/
CANCELLED(2);
public final int value;
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b7ec4121/openmessaging/src/main/java/io/openmessaging/rocketmq/utils/BeanUtils.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/utils/BeanUtils.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/utils/BeanUtils.java
index d8eed84..104d3d9 100644
--- a/openmessaging/src/main/java/io/openmessaging/rocketmq/utils/BeanUtils.java
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/utils/BeanUtils.java
@@ -29,7 +29,7 @@ import org.slf4j.Logger;
public final class BeanUtils {
final static Logger log = ClientLogger.getLog();
-
+
/**
* Maps primitive {@code Class}es to their corresponding wrapper {@code Class}.
*/
@@ -136,7 +136,7 @@ public final class BeanUtils {
public static <T> T populate(final Properties properties, final T obj) {
Class<?> clazz = obj.getClass();
try {
-
+
Set<Map.Entry<Object, Object>> entries = properties.entrySet();
for (Map.Entry<Object, Object> entry : entries) {
String entryKey = entry.getKey().toString();
@@ -147,7 +147,7 @@ public final class BeanUtils {
}
String beanFieldNameWithCapitalization = StringUtils.join(keyGroup);
try {
- setProperties(clazz, obj, "set" + beanFieldNameWithCapitalization, entry.getValue());
+ setProperties(clazz, obj, "set" + beanFieldNameWithCapitalization, entry.getValue());
} catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException ignored) {
//ignored...
}