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 2016/12/28 02:44:03 UTC
[07/58] [abbrv] [partial] incubator-rocketmq git commit: ROCKETMQ-18
Rename package name from com.alibaba to org.apache
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/filtersrv/RegisterFilterServerResponseHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/filtersrv/RegisterFilterServerResponseHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/filtersrv/RegisterFilterServerResponseHeader.java
new file mode 100644
index 0000000..a2a52f0
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/filtersrv/RegisterFilterServerResponseHeader.java
@@ -0,0 +1,55 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.protocol.header.filtersrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+public class RegisterFilterServerResponseHeader implements CommandCustomHeader {
+ @CFNotNull
+ private String brokerName;
+ @CFNotNull
+ private long brokerId;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ }
+
+
+ public long getBrokerId() {
+ return brokerId;
+ }
+
+
+ public void setBrokerId(long brokerId) {
+ this.brokerId = brokerId;
+ }
+
+
+ public String getBrokerName() {
+ return brokerName;
+ }
+
+
+ public void setBrokerName(String brokerName) {
+ this.brokerName = brokerName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/filtersrv/RegisterMessageFilterClassRequestHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/filtersrv/RegisterMessageFilterClassRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/filtersrv/RegisterMessageFilterClassRequestHeader.java
new file mode 100644
index 0000000..1fc94a9
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/filtersrv/RegisterMessageFilterClassRequestHeader.java
@@ -0,0 +1,79 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.protocol.header.filtersrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+public class RegisterMessageFilterClassRequestHeader implements CommandCustomHeader {
+ @CFNotNull
+ private String consumerGroup;
+ @CFNotNull
+ private String topic;
+ @CFNotNull
+ private String className;
+ @CFNotNull
+ private Integer classCRC;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ }
+
+
+ public String getConsumerGroup() {
+ return consumerGroup;
+ }
+
+
+ public void setConsumerGroup(String consumerGroup) {
+ this.consumerGroup = consumerGroup;
+ }
+
+
+ public String getTopic() {
+ return topic;
+ }
+
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+
+ public String getClassName() {
+ return className;
+ }
+
+
+ public void setClassName(String className) {
+ this.className = className;
+ }
+
+
+ public Integer getClassCRC() {
+ return classCRC;
+ }
+
+
+ public void setClassCRC(Integer classCRC) {
+ this.classCRC = classCRC;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/DeleteKVConfigRequestHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/DeleteKVConfigRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/DeleteKVConfigRequestHeader.java
new file mode 100644
index 0000000..47ec4b1
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/DeleteKVConfigRequestHeader.java
@@ -0,0 +1,59 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * @author shijia.wxr
+ *
+ */
+public class DeleteKVConfigRequestHeader implements CommandCustomHeader {
+ @CFNotNull
+ private String namespace;
+ @CFNotNull
+ private String key;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ }
+
+
+ public String getNamespace() {
+ return namespace;
+ }
+
+
+ public void setNamespace(String namespace) {
+ this.namespace = namespace;
+ }
+
+
+ public String getKey() {
+ return key;
+ }
+
+
+ public void setKey(String key) {
+ this.key = key;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/DeleteTopicInNamesrvRequestHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/DeleteTopicInNamesrvRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/DeleteTopicInNamesrvRequestHeader.java
new file mode 100644
index 0000000..5bd0632
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/DeleteTopicInNamesrvRequestHeader.java
@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * @author shijia.wxr
+ *
+ */
+public class DeleteTopicInNamesrvRequestHeader implements CommandCustomHeader {
+ @CFNotNull
+ private String topic;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ }
+
+
+ public String getTopic() {
+ return topic;
+ }
+
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetKVConfigRequestHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetKVConfigRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetKVConfigRequestHeader.java
new file mode 100644
index 0000000..29e4db4
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetKVConfigRequestHeader.java
@@ -0,0 +1,59 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * @author shijia.wxr
+ *
+ */
+public class GetKVConfigRequestHeader implements CommandCustomHeader {
+ @CFNotNull
+ private String namespace;
+ @CFNotNull
+ private String key;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ }
+
+
+ public String getNamespace() {
+ return namespace;
+ }
+
+
+ public void setNamespace(String namespace) {
+ this.namespace = namespace;
+ }
+
+
+ public String getKey() {
+ return key;
+ }
+
+
+ public void setKey(String key) {
+ this.key = key;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetKVConfigResponseHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetKVConfigResponseHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetKVConfigResponseHeader.java
new file mode 100644
index 0000000..3280ff5
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetKVConfigResponseHeader.java
@@ -0,0 +1,47 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNullable;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * @author shijia.wxr
+ *
+ */
+public class GetKVConfigResponseHeader implements CommandCustomHeader {
+ @CFNullable
+ private String value;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ }
+
+
+ public String getValue() {
+ return value;
+ }
+
+
+ public void setValue(String value) {
+ this.value = value;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetKVListByNamespaceRequestHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetKVListByNamespaceRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetKVListByNamespaceRequestHeader.java
new file mode 100644
index 0000000..bd2816e
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetKVListByNamespaceRequestHeader.java
@@ -0,0 +1,47 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * @author shijia.wxr
+ *
+ */
+public class GetKVListByNamespaceRequestHeader implements CommandCustomHeader {
+ @CFNotNull
+ private String namespace;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ }
+
+
+ public String getNamespace() {
+ return namespace;
+ }
+
+
+ public void setNamespace(String namespace) {
+ this.namespace = namespace;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetRouteInfoRequestHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetRouteInfoRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetRouteInfoRequestHeader.java
new file mode 100644
index 0000000..972cf35
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetRouteInfoRequestHeader.java
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: GetRouteInfoRequestHeader.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * @author shijia.wxr
+ */
+public class GetRouteInfoRequestHeader implements CommandCustomHeader {
+ @CFNotNull
+ private String topic;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ }
+
+
+ public String getTopic() {
+ return topic;
+ }
+
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetRouteInfoResponseHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetRouteInfoResponseHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetRouteInfoResponseHeader.java
new file mode 100644
index 0000000..8bb681e
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/GetRouteInfoResponseHeader.java
@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: GetRouteInfoResponseHeader.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * @author shijia.wxr
+ */
+public class GetRouteInfoResponseHeader implements CommandCustomHeader {
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ // TODO Auto-generated method stub
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/PutKVConfigRequestHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/PutKVConfigRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/PutKVConfigRequestHeader.java
new file mode 100644
index 0000000..01e9a5e
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/PutKVConfigRequestHeader.java
@@ -0,0 +1,67 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+public class PutKVConfigRequestHeader implements CommandCustomHeader {
+ @CFNotNull
+ private String namespace;
+ @CFNotNull
+ private String key;
+ @CFNotNull
+ private String value;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ }
+
+
+ public String getNamespace() {
+ return namespace;
+ }
+
+
+ public void setNamespace(String namespace) {
+ this.namespace = namespace;
+ }
+
+
+ public String getKey() {
+ return key;
+ }
+
+
+ public void setKey(String key) {
+ this.key = key;
+ }
+
+
+ public String getValue() {
+ return value;
+ }
+
+
+ public void setValue(String value) {
+ this.value = value;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/RegisterBrokerRequestHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/RegisterBrokerRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/RegisterBrokerRequestHeader.java
new file mode 100644
index 0000000..4c0fca5
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/RegisterBrokerRequestHeader.java
@@ -0,0 +1,97 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: RegisterBrokerRequestHeader.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * @author lansheng.zj
+ */
+public class RegisterBrokerRequestHeader implements CommandCustomHeader {
+ @CFNotNull
+ private String brokerName;
+ @CFNotNull
+ private String brokerAddr;
+ @CFNotNull
+ private String clusterName;
+ @CFNotNull
+ private String haServerAddr;
+ @CFNotNull
+ private Long brokerId;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ }
+
+
+ public String getBrokerName() {
+ return brokerName;
+ }
+
+
+ public void setBrokerName(String brokerName) {
+ this.brokerName = brokerName;
+ }
+
+
+ public String getBrokerAddr() {
+ return brokerAddr;
+ }
+
+
+ public void setBrokerAddr(String brokerAddr) {
+ this.brokerAddr = brokerAddr;
+ }
+
+
+ public String getClusterName() {
+ return clusterName;
+ }
+
+
+ public void setClusterName(String clusterName) {
+ this.clusterName = clusterName;
+ }
+
+
+ public String getHaServerAddr() {
+ return haServerAddr;
+ }
+
+
+ public void setHaServerAddr(String haServerAddr) {
+ this.haServerAddr = haServerAddr;
+ }
+
+
+ public Long getBrokerId() {
+ return brokerId;
+ }
+
+
+ public void setBrokerId(Long brokerId) {
+ this.brokerId = brokerId;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/RegisterBrokerResponseHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/RegisterBrokerResponseHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/RegisterBrokerResponseHeader.java
new file mode 100644
index 0000000..9796054
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/RegisterBrokerResponseHeader.java
@@ -0,0 +1,59 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNullable;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * @author shijia.wxr
+ *
+ */
+public class RegisterBrokerResponseHeader implements CommandCustomHeader {
+ @CFNullable
+ private String haServerAddr;
+ @CFNullable
+ private String masterAddr;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ }
+
+
+ public String getHaServerAddr() {
+ return haServerAddr;
+ }
+
+
+ public void setHaServerAddr(String haServerAddr) {
+ this.haServerAddr = haServerAddr;
+ }
+
+
+ public String getMasterAddr() {
+ return masterAddr;
+ }
+
+
+ public void setMasterAddr(String masterAddr) {
+ this.masterAddr = masterAddr;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/RegisterOrderTopicRequestHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/RegisterOrderTopicRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/RegisterOrderTopicRequestHeader.java
new file mode 100644
index 0000000..cb5b3d9
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/RegisterOrderTopicRequestHeader.java
@@ -0,0 +1,62 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: RegisterOrderTopicRequestHeader.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * @author shijia.wxr
+ */
+public class RegisterOrderTopicRequestHeader implements CommandCustomHeader {
+ @CFNotNull
+ private String topic;
+ @CFNotNull
+ private String orderTopicString;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ // TODO Auto-generated method stub
+ }
+
+
+ public String getTopic() {
+ return topic;
+ }
+
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+
+ public String getOrderTopicString() {
+ return orderTopicString;
+ }
+
+
+ public void setOrderTopicString(String orderTopicString) {
+ this.orderTopicString = orderTopicString;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/UnRegisterBrokerRequestHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/UnRegisterBrokerRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/UnRegisterBrokerRequestHeader.java
new file mode 100644
index 0000000..f2d174a
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/UnRegisterBrokerRequestHeader.java
@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: UnRegisterBrokerRequestHeader.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * @author lansheng.zj
+ */
+public class UnRegisterBrokerRequestHeader implements CommandCustomHeader {
+ @CFNotNull
+ private String brokerName;
+ @CFNotNull
+ private String brokerAddr;
+ @CFNotNull
+ private String clusterName;
+ @CFNotNull
+ private Long brokerId;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ }
+
+
+ public String getBrokerName() {
+ return brokerName;
+ }
+
+
+ public void setBrokerName(String brokerName) {
+ this.brokerName = brokerName;
+ }
+
+
+ public String getBrokerAddr() {
+ return brokerAddr;
+ }
+
+
+ public void setBrokerAddr(String brokerAddr) {
+ this.brokerAddr = brokerAddr;
+ }
+
+
+ public String getClusterName() {
+ return clusterName;
+ }
+
+
+ public void setClusterName(String clusterName) {
+ this.clusterName = clusterName;
+ }
+
+
+ public Long getBrokerId() {
+ return brokerId;
+ }
+
+
+ public void setBrokerId(Long brokerId) {
+ this.brokerId = brokerId;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/WipeWritePermOfBrokerRequestHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/WipeWritePermOfBrokerRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/WipeWritePermOfBrokerRequestHeader.java
new file mode 100644
index 0000000..f5aebb9
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/WipeWritePermOfBrokerRequestHeader.java
@@ -0,0 +1,47 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * @author shijia.wxr
+ *
+ */
+public class WipeWritePermOfBrokerRequestHeader implements CommandCustomHeader {
+ @CFNotNull
+ private String brokerName;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+
+ }
+
+
+ public String getBrokerName() {
+ return brokerName;
+ }
+
+
+ public void setBrokerName(String brokerName) {
+ this.brokerName = brokerName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/WipeWritePermOfBrokerResponseHeader.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/WipeWritePermOfBrokerResponseHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/WipeWritePermOfBrokerResponseHeader.java
new file mode 100644
index 0000000..e50641b
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/namesrv/WipeWritePermOfBrokerResponseHeader.java
@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.common.protocol.header.namesrv;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+/**
+ * @author shijia.wxr
+ *
+ */
+public class WipeWritePermOfBrokerResponseHeader implements CommandCustomHeader {
+ @CFNotNull
+ private Integer wipeTopicCount;
+
+
+ @Override
+ public void checkFields() throws RemotingCommandException {
+ }
+
+
+ public Integer getWipeTopicCount() {
+ return wipeTopicCount;
+ }
+
+
+ public void setWipeTopicCount(Integer wipeTopicCount) {
+ this.wipeTopicCount = wipeTopicCount;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ConsumeType.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ConsumeType.java b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ConsumeType.java
new file mode 100644
index 0000000..115a885
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ConsumeType.java
@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: ConsumeType.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.heartbeat;
+
+/**
+ * @author shijia.wxr
+ */
+public enum ConsumeType {
+
+ CONSUME_ACTIVELY("PULL"),
+
+ CONSUME_PASSIVELY("PUSH");
+
+ private String typeCN;
+
+ ConsumeType(String typeCN) {
+ this.typeCN = typeCN;
+ }
+
+
+ public String getTypeCN() {
+ return typeCN;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ConsumerData.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ConsumerData.java b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ConsumerData.java
new file mode 100644
index 0000000..233da6c
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ConsumerData.java
@@ -0,0 +1,107 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: ConsumerData.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.heartbeat;
+
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+
+import java.util.HashSet;
+import java.util.Set;
+
+
+/**
+ * @author shijia.wxr
+ */
+public class ConsumerData {
+ private String groupName;
+ private ConsumeType consumeType;
+ private MessageModel messageModel;
+ private ConsumeFromWhere consumeFromWhere;
+ private Set<SubscriptionData> subscriptionDataSet = new HashSet<SubscriptionData>();
+ private boolean unitMode;
+
+
+ public String getGroupName() {
+ return groupName;
+ }
+
+
+ public void setGroupName(String groupName) {
+ this.groupName = groupName;
+ }
+
+
+ public ConsumeType getConsumeType() {
+ return consumeType;
+ }
+
+
+ public void setConsumeType(ConsumeType consumeType) {
+ this.consumeType = consumeType;
+ }
+
+
+ public MessageModel getMessageModel() {
+ return messageModel;
+ }
+
+
+ public void setMessageModel(MessageModel messageModel) {
+ this.messageModel = messageModel;
+ }
+
+
+ public ConsumeFromWhere getConsumeFromWhere() {
+ return consumeFromWhere;
+ }
+
+
+ public void setConsumeFromWhere(ConsumeFromWhere consumeFromWhere) {
+ this.consumeFromWhere = consumeFromWhere;
+ }
+
+
+ public Set<SubscriptionData> getSubscriptionDataSet() {
+ return subscriptionDataSet;
+ }
+
+
+ public void setSubscriptionDataSet(Set<SubscriptionData> subscriptionDataSet) {
+ this.subscriptionDataSet = subscriptionDataSet;
+ }
+
+
+ public boolean isUnitMode() {
+ return unitMode;
+ }
+
+
+ public void setUnitMode(boolean isUnitMode) {
+ this.unitMode = isUnitMode;
+ }
+
+
+ @Override
+ public String toString() {
+ return "ConsumerData [groupName=" + groupName + ", consumeType=" + consumeType + ", messageModel="
+ + messageModel + ", consumeFromWhere=" + consumeFromWhere + ", unitMode=" + unitMode
+ + ", subscriptionDataSet=" + subscriptionDataSet + "]";
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/HeartbeatData.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/HeartbeatData.java b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/HeartbeatData.java
new file mode 100644
index 0000000..8fa5b17
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/HeartbeatData.java
@@ -0,0 +1,73 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: HeartbeatData.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.heartbeat;
+
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+
+import java.util.HashSet;
+import java.util.Set;
+
+
+/**
+ * @author shijia.wxr
+ */
+public class HeartbeatData extends RemotingSerializable {
+ private String clientID;
+ private Set<ProducerData> producerDataSet = new HashSet<ProducerData>();
+ private Set<ConsumerData> consumerDataSet = new HashSet<ConsumerData>();
+
+
+ public String getClientID() {
+ return clientID;
+ }
+
+
+ public void setClientID(String clientID) {
+ this.clientID = clientID;
+ }
+
+
+ public Set<ProducerData> getProducerDataSet() {
+ return producerDataSet;
+ }
+
+
+ public void setProducerDataSet(Set<ProducerData> producerDataSet) {
+ this.producerDataSet = producerDataSet;
+ }
+
+
+ public Set<ConsumerData> getConsumerDataSet() {
+ return consumerDataSet;
+ }
+
+
+ public void setConsumerDataSet(Set<ConsumerData> consumerDataSet) {
+ this.consumerDataSet = consumerDataSet;
+ }
+
+
+ @Override
+ public String toString() {
+ return "HeartbeatData [clientID=" + clientID + ", producerDataSet=" + producerDataSet
+ + ", consumerDataSet=" + consumerDataSet + "]";
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/MessageModel.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/MessageModel.java b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/MessageModel.java
new file mode 100644
index 0000000..4600c6f
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/MessageModel.java
@@ -0,0 +1,48 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: MessageModel.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.heartbeat;
+
+/**
+ * Message model
+ *
+ * @author shijia.wxr
+ */
+public enum MessageModel {
+ /**
+ * broadcast
+ */
+ BROADCASTING("BROADCASTING"),
+ /**
+ * clustering
+ */
+ CLUSTERING("CLUSTERING");
+
+ private String modeCN;
+
+ MessageModel(String modeCN) {
+ this.modeCN = modeCN;
+ }
+
+
+ public String getModeCN() {
+ return modeCN;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ProducerData.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ProducerData.java b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ProducerData.java
new file mode 100644
index 0000000..c83b14c
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/ProducerData.java
@@ -0,0 +1,44 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: ProducerData.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.heartbeat;
+
+/**
+ * @author shijia.wxr
+ */
+public class ProducerData {
+ private String groupName;
+
+
+ public String getGroupName() {
+ return groupName;
+ }
+
+
+ public void setGroupName(String groupName) {
+ this.groupName = groupName;
+ }
+
+
+ @Override
+ public String toString() {
+ return "ProducerData [groupName=" + groupName + "]";
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/SubscriptionData.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/SubscriptionData.java b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/SubscriptionData.java
new file mode 100644
index 0000000..28b49f1
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/heartbeat/SubscriptionData.java
@@ -0,0 +1,188 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: SubscriptionData.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.heartbeat;
+
+import com.alibaba.fastjson.annotation.JSONField;
+
+import java.util.HashSet;
+import java.util.Set;
+
+
+/**
+ * @author shijia.wxr
+ */
+public class SubscriptionData implements Comparable<SubscriptionData> {
+ public final static String SUB_ALL = "*";
+ private boolean classFilterMode = false;
+ private String topic;
+ private String subString;
+ private Set<String> tagsSet = new HashSet<String>();
+ private Set<Integer> codeSet = new HashSet<Integer>();
+ private long subVersion = System.currentTimeMillis();
+
+ @JSONField(serialize = false)
+ private String filterClassSource;
+
+
+ public SubscriptionData() {
+
+ }
+
+
+ public SubscriptionData(String topic, String subString) {
+ super();
+ this.topic = topic;
+ this.subString = subString;
+ }
+
+ public String getFilterClassSource() {
+ return filterClassSource;
+ }
+
+ public void setFilterClassSource(String filterClassSource) {
+ this.filterClassSource = filterClassSource;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+
+ public String getSubString() {
+ return subString;
+ }
+
+
+ public void setSubString(String subString) {
+ this.subString = subString;
+ }
+
+
+ public Set<String> getTagsSet() {
+ return tagsSet;
+ }
+
+
+ public void setTagsSet(Set<String> tagsSet) {
+ this.tagsSet = tagsSet;
+ }
+
+
+ public long getSubVersion() {
+ return subVersion;
+ }
+
+
+ public void setSubVersion(long subVersion) {
+ this.subVersion = subVersion;
+ }
+
+
+ public Set<Integer> getCodeSet() {
+ return codeSet;
+ }
+
+
+ public void setCodeSet(Set<Integer> codeSet) {
+ this.codeSet = codeSet;
+ }
+
+
+ public boolean isClassFilterMode() {
+ return classFilterMode;
+ }
+
+
+ public void setClassFilterMode(boolean classFilterMode) {
+ this.classFilterMode = classFilterMode;
+ }
+
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + (classFilterMode ? 1231 : 1237);
+ result = prime * result + ((codeSet == null) ? 0 : codeSet.hashCode());
+ result = prime * result + ((subString == null) ? 0 : subString.hashCode());
+ result = prime * result + ((tagsSet == null) ? 0 : tagsSet.hashCode());
+ result = prime * result + ((topic == null) ? 0 : topic.hashCode());
+ return result;
+ }
+
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ SubscriptionData other = (SubscriptionData) obj;
+ if (classFilterMode != other.classFilterMode)
+ return false;
+ if (codeSet == null) {
+ if (other.codeSet != null)
+ return false;
+ } else if (!codeSet.equals(other.codeSet))
+ return false;
+ if (subString == null) {
+ if (other.subString != null)
+ return false;
+ } else if (!subString.equals(other.subString))
+ return false;
+ if (subVersion != other.subVersion)
+ return false;
+ if (tagsSet == null) {
+ if (other.tagsSet != null)
+ return false;
+ } else if (!tagsSet.equals(other.tagsSet))
+ return false;
+ if (topic == null) {
+ if (other.topic != null)
+ return false;
+ } else if (!topic.equals(other.topic))
+ return false;
+ return true;
+ }
+
+
+ @Override
+ public String toString() {
+ return "SubscriptionData [classFilterMode=" + classFilterMode + ", topic=" + topic + ", subString="
+ + subString + ", tagsSet=" + tagsSet + ", codeSet=" + codeSet + ", subVersion=" + subVersion
+ + "]";
+ }
+
+
+ @Override
+ public int compareTo(SubscriptionData other) {
+ String thisValue = this.topic + "@" + this.subString;
+ String otherValue = other.topic + "@" + other.subString;
+ return thisValue.compareTo(otherValue);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/route/BrokerData.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/route/BrokerData.java b/common/src/main/java/org/apache/rocketmq/common/protocol/route/BrokerData.java
new file mode 100644
index 0000000..1696cd6
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/route/BrokerData.java
@@ -0,0 +1,113 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: BrokerData.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.route;
+
+import org.apache.rocketmq.common.MixAll;
+
+import java.util.HashMap;
+import java.util.Map;
+
+
+/**
+ * @author shijia.wxr
+ *
+ */
+public class BrokerData implements Comparable<BrokerData> {
+ private String cluster;
+ private String brokerName;
+ private HashMap<Long/* brokerId */, String/* broker address */> brokerAddrs;
+
+ public String selectBrokerAddr() {
+ String value = this.brokerAddrs.get(MixAll.MASTER_ID);
+ if (null == value) {
+ for (Map.Entry<Long, String> entry : this.brokerAddrs.entrySet()) {
+ return entry.getValue();
+ }
+ }
+
+ return value;
+ }
+
+ public HashMap<Long, String> getBrokerAddrs() {
+ return brokerAddrs;
+ }
+
+ public void setBrokerAddrs(HashMap<Long, String> brokerAddrs) {
+ this.brokerAddrs = brokerAddrs;
+ }
+
+ public String getCluster() {
+ return cluster;
+ }
+
+ public void setCluster(String cluster) {
+ this.cluster = cluster;
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((brokerAddrs == null) ? 0 : brokerAddrs.hashCode());
+ result = prime * result + ((brokerName == null) ? 0 : brokerName.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ BrokerData other = (BrokerData) obj;
+ if (brokerAddrs == null) {
+ if (other.brokerAddrs != null)
+ return false;
+ } else if (!brokerAddrs.equals(other.brokerAddrs))
+ return false;
+ if (brokerName == null) {
+ if (other.brokerName != null)
+ return false;
+ } else if (!brokerName.equals(other.brokerName))
+ return false;
+ return true;
+ }
+
+ @Override
+ public String toString() {
+ return "BrokerData [brokerName=" + brokerName + ", brokerAddrs=" + brokerAddrs + "]";
+ }
+
+ @Override
+ public int compareTo(BrokerData o) {
+ return this.brokerName.compareTo(o.getBrokerName());
+ }
+
+ public String getBrokerName() {
+ return brokerName;
+ }
+
+ public void setBrokerName(String brokerName) {
+ this.brokerName = brokerName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/route/QueueData.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/route/QueueData.java b/common/src/main/java/org/apache/rocketmq/common/protocol/route/QueueData.java
new file mode 100644
index 0000000..de736be
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/route/QueueData.java
@@ -0,0 +1,118 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: QueueData.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.route;
+
+public class QueueData implements Comparable<QueueData> {
+ private String brokerName;
+ private int readQueueNums;
+ private int writeQueueNums;
+ private int perm;
+ private int topicSynFlag;
+
+ public int getReadQueueNums() {
+ return readQueueNums;
+ }
+
+ public void setReadQueueNums(int readQueueNums) {
+ this.readQueueNums = readQueueNums;
+ }
+
+ public int getWriteQueueNums() {
+ return writeQueueNums;
+ }
+
+ public void setWriteQueueNums(int writeQueueNums) {
+ this.writeQueueNums = writeQueueNums;
+ }
+
+ public int getPerm() {
+ return perm;
+ }
+
+ public void setPerm(int perm) {
+ this.perm = perm;
+ }
+
+ public int getTopicSynFlag() {
+ return topicSynFlag;
+ }
+
+ public void setTopicSynFlag(int topicSynFlag) {
+ this.topicSynFlag = topicSynFlag;
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((brokerName == null) ? 0 : brokerName.hashCode());
+ result = prime * result + perm;
+ result = prime * result + readQueueNums;
+ result = prime * result + writeQueueNums;
+ result = prime * result + topicSynFlag;
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ QueueData other = (QueueData) obj;
+ if (brokerName == null) {
+ if (other.brokerName != null)
+ return false;
+ } else if (!brokerName.equals(other.brokerName))
+ return false;
+ if (perm != other.perm)
+ return false;
+ if (readQueueNums != other.readQueueNums)
+ return false;
+ if (writeQueueNums != other.writeQueueNums)
+ return false;
+ if (topicSynFlag != other.topicSynFlag)
+ return false;
+ return true;
+ }
+
+ @Override
+ public String toString() {
+ return "QueueData [brokerName=" + brokerName + ", readQueueNums=" + readQueueNums
+ + ", writeQueueNums=" + writeQueueNums + ", perm=" + perm + ", topicSynFlag=" + topicSynFlag
+ + "]";
+ }
+
+ @Override
+ public int compareTo(QueueData o) {
+ return this.brokerName.compareTo(o.getBrokerName());
+ }
+
+ public String getBrokerName() {
+ return brokerName;
+ }
+
+ public void setBrokerName(String brokerName) {
+ this.brokerName = brokerName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/route/TopicRouteData.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/route/TopicRouteData.java b/common/src/main/java/org/apache/rocketmq/common/protocol/route/TopicRouteData.java
new file mode 100644
index 0000000..13c5273
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/route/TopicRouteData.java
@@ -0,0 +1,146 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * $Id: TopicRouteData.java 1835 2013-05-16 02:00:50Z shijia.wxr $
+ */
+package org.apache.rocketmq.common.protocol.route;
+
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+
+/**
+ * @author shijia.wxr
+ */
+public class TopicRouteData extends RemotingSerializable {
+ private String orderTopicConf;
+ private List<QueueData> queueDatas;
+ private List<BrokerData> brokerDatas;
+ private HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
+
+
+ public TopicRouteData cloneTopicRouteData() {
+ TopicRouteData topicRouteData = new TopicRouteData();
+ topicRouteData.setQueueDatas(new ArrayList<QueueData>());
+ topicRouteData.setBrokerDatas(new ArrayList<BrokerData>());
+ topicRouteData.setFilterServerTable(new HashMap<String, List<String>>());
+ topicRouteData.setOrderTopicConf(this.orderTopicConf);
+
+ if (this.queueDatas != null) {
+ topicRouteData.getQueueDatas().addAll(this.queueDatas);
+ }
+
+ if (this.brokerDatas != null) {
+ topicRouteData.getBrokerDatas().addAll(this.brokerDatas);
+ }
+
+ if (this.filterServerTable != null) {
+ topicRouteData.getFilterServerTable().putAll(this.filterServerTable);
+ }
+
+ return topicRouteData;
+ }
+
+
+ public List<QueueData> getQueueDatas() {
+ return queueDatas;
+ }
+
+
+ public void setQueueDatas(List<QueueData> queueDatas) {
+ this.queueDatas = queueDatas;
+ }
+
+
+ public List<BrokerData> getBrokerDatas() {
+ return brokerDatas;
+ }
+
+
+ public void setBrokerDatas(List<BrokerData> brokerDatas) {
+ this.brokerDatas = brokerDatas;
+ }
+
+ public HashMap<String, List<String>> getFilterServerTable() {
+ return filterServerTable;
+ }
+
+ public void setFilterServerTable(HashMap<String, List<String>> filterServerTable) {
+ this.filterServerTable = filterServerTable;
+ }
+
+ public String getOrderTopicConf() {
+ return orderTopicConf;
+ }
+
+ public void setOrderTopicConf(String orderTopicConf) {
+ this.orderTopicConf = orderTopicConf;
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((brokerDatas == null) ? 0 : brokerDatas.hashCode());
+ result = prime * result + ((orderTopicConf == null) ? 0 : orderTopicConf.hashCode());
+ result = prime * result + ((queueDatas == null) ? 0 : queueDatas.hashCode());
+ result = prime * result + ((filterServerTable == null) ? 0 : filterServerTable.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ TopicRouteData other = (TopicRouteData) obj;
+ if (brokerDatas == null) {
+ if (other.brokerDatas != null)
+ return false;
+ } else if (!brokerDatas.equals(other.brokerDatas))
+ return false;
+ if (orderTopicConf == null) {
+ if (other.orderTopicConf != null)
+ return false;
+ } else if (!orderTopicConf.equals(other.orderTopicConf))
+ return false;
+ if (queueDatas == null) {
+ if (other.queueDatas != null)
+ return false;
+ } else if (!queueDatas.equals(other.queueDatas))
+ return false;
+ if (filterServerTable == null) {
+ if (other.filterServerTable != null)
+ return false;
+ } else if (!filterServerTable.equals(other.filterServerTable))
+ return false;
+ return true;
+ }
+
+ @Override
+ public String toString() {
+ return "TopicRouteData [orderTopicConf=" + orderTopicConf + ", queueDatas=" + queueDatas
+ + ", brokerDatas=" + brokerDatas + ", filterServerTable=" + filterServerTable + "]";
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/protocol/topic/OffsetMovedEvent.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/topic/OffsetMovedEvent.java b/common/src/main/java/org/apache/rocketmq/common/protocol/topic/OffsetMovedEvent.java
new file mode 100644
index 0000000..df5ec71
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/topic/OffsetMovedEvent.java
@@ -0,0 +1,76 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.protocol.topic;
+
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+
+
+public class OffsetMovedEvent extends RemotingSerializable {
+ private String consumerGroup;
+ private MessageQueue messageQueue;
+ private long offsetRequest;
+ private long offsetNew;
+
+
+ public String getConsumerGroup() {
+ return consumerGroup;
+ }
+
+
+ public void setConsumerGroup(String consumerGroup) {
+ this.consumerGroup = consumerGroup;
+ }
+
+
+ public MessageQueue getMessageQueue() {
+ return messageQueue;
+ }
+
+
+ public void setMessageQueue(MessageQueue messageQueue) {
+ this.messageQueue = messageQueue;
+ }
+
+
+ public long getOffsetRequest() {
+ return offsetRequest;
+ }
+
+
+ public void setOffsetRequest(long offsetRequest) {
+ this.offsetRequest = offsetRequest;
+ }
+
+
+ public long getOffsetNew() {
+ return offsetNew;
+ }
+
+
+ public void setOffsetNew(long offsetNew) {
+ this.offsetNew = offsetNew;
+ }
+
+
+ @Override
+ public String toString() {
+ return "OffsetMovedEvent [consumerGroup=" + consumerGroup + ", messageQueue=" + messageQueue
+ + ", offsetRequest=" + offsetRequest + ", offsetNew=" + offsetNew + "]";
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/queue/ConcurrentTreeMap.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/queue/ConcurrentTreeMap.java b/common/src/main/java/org/apache/rocketmq/common/queue/ConcurrentTreeMap.java
new file mode 100644
index 0000000..7036fdd
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/queue/ConcurrentTreeMap.java
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.queue;
+
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Comparator;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.locks.ReentrantLock;
+
+
+/**
+ * thread safe
+ *
+ * @author lansheng.zj
+ */
+public class ConcurrentTreeMap<K, V> {
+ private static final Logger log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
+ private final ReentrantLock lock;
+ private TreeMap<K, V> tree;
+ private RoundQueue<K> roundQueue;
+
+
+ public ConcurrentTreeMap(int capacity, Comparator<? super K> comparator) {
+ tree = new TreeMap<K, V>(comparator);
+ roundQueue = new RoundQueue<K>(capacity);
+ lock = new ReentrantLock(true);
+ }
+
+
+ public Map.Entry<K, V> pollFirstEntry() {
+ lock.lock();
+ try {
+ return tree.pollFirstEntry();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+
+ public V putIfAbsentAndRetExsit(K key, V value) {
+ lock.lock();
+ try {
+ if (roundQueue.put(key)) {
+ V exsit = tree.get(key);
+ if (null == exsit) {
+ tree.put(key, value);
+ exsit = value;
+ }
+ log.warn("putIfAbsentAndRetExsit success. {}", key);
+ return exsit;
+ }
+
+ else {
+ V exsit = tree.get(key);
+ return exsit;
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/queue/RoundQueue.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/queue/RoundQueue.java b/common/src/main/java/org/apache/rocketmq/common/queue/RoundQueue.java
new file mode 100644
index 0000000..f32569b
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/queue/RoundQueue.java
@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.queue;
+
+import java.util.LinkedList;
+import java.util.Queue;
+
+
+/**
+ * not thread safe
+ *
+ * @author lansheng.zj
+ */
+public class RoundQueue<E> {
+
+ private Queue<E> queue;
+ private int capacity;
+
+
+ public RoundQueue(int capacity) {
+ this.capacity = capacity;
+ queue = new LinkedList<E>();
+ }
+
+
+ public boolean put(E e) {
+ boolean ok = false;
+ if (!queue.contains(e)) {
+ if (queue.size() >= capacity) {
+ queue.poll();
+ }
+ queue.add(e);
+ ok = true;
+ }
+
+ return ok;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/running/RunningStats.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/running/RunningStats.java b/common/src/main/java/org/apache/rocketmq/common/running/RunningStats.java
new file mode 100644
index 0000000..106d111
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/running/RunningStats.java
@@ -0,0 +1,25 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.common.running;
+
+public enum RunningStats {
+ commitLogMaxOffset,
+ commitLogMinOffset,
+ commitLogDiskRatio,
+ consumeQueueDiskRatio,
+ scheduleMessageOffset,
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItem.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItem.java b/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItem.java
new file mode 100644
index 0000000..58eedb2
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItem.java
@@ -0,0 +1,82 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.stats;
+
+import org.apache.rocketmq.common.UtilAll;
+import org.slf4j.Logger;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+
+public class MomentStatsItem {
+
+ private final AtomicLong value = new AtomicLong(0);
+
+ private final String statsName;
+ private final String statsKey;
+ private final ScheduledExecutorService scheduledExecutorService;
+ private final Logger log;
+
+
+ public MomentStatsItem(String statsName, String statsKey,
+ ScheduledExecutorService scheduledExecutorService, Logger log) {
+ this.statsName = statsName;
+ this.statsKey = statsKey;
+ this.scheduledExecutorService = scheduledExecutorService;
+ this.log = log;
+ }
+
+
+ public void init() {
+ this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ printAtMinutes();
+
+ MomentStatsItem.this.value.set(0);
+ } catch (Throwable e) {
+ }
+ }
+ }, Math.abs(UtilAll.computNextMinutesTimeMillis() - System.currentTimeMillis()), 1000 * 60 * 5, TimeUnit.MILLISECONDS);
+ }
+
+
+ public void printAtMinutes() {
+ log.info(String.format("[%s] [%s] Stats Every 5 Minutes, Value: %d",
+ this.statsName,
+ this.statsKey,
+ this.value.get()));
+ }
+
+ public AtomicLong getValue() {
+ return value;
+ }
+
+
+ public String getStatsKey() {
+ return statsKey;
+ }
+
+
+ public String getStatsName() {
+ return statsName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItemSet.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItemSet.java b/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItemSet.java
new file mode 100644
index 0000000..aba61c7
--- /dev/null
+++ b/common/src/main/java/org/apache/rocketmq/common/stats/MomentStatsItemSet.java
@@ -0,0 +1,94 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.stats;
+
+import org.apache.rocketmq.common.UtilAll;
+import org.slf4j.Logger;
+
+import java.util.Iterator;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+
+public class MomentStatsItemSet {
+ private final ConcurrentHashMap<String/* key */, MomentStatsItem> statsItemTable =
+ new ConcurrentHashMap<String, MomentStatsItem>(128);
+ private final String statsName;
+ private final ScheduledExecutorService scheduledExecutorService;
+ private final Logger log;
+
+
+ public MomentStatsItemSet(String statsName, ScheduledExecutorService scheduledExecutorService, Logger log) {
+ this.statsName = statsName;
+ this.scheduledExecutorService = scheduledExecutorService;
+ this.log = log;
+ this.init();
+ }
+
+ public ConcurrentHashMap<String, MomentStatsItem> getStatsItemTable() {
+ return statsItemTable;
+ }
+
+ public String getStatsName() {
+ return statsName;
+ }
+
+ public void init() {
+
+ this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ printAtMinutes();
+ } catch (Throwable e) {
+ }
+ }
+ }, Math.abs(UtilAll.computNextMinutesTimeMillis() - System.currentTimeMillis()), 1000 * 60 * 5, TimeUnit.MILLISECONDS);
+ }
+
+ private void printAtMinutes() {
+ Iterator<Entry<String, MomentStatsItem>> it = this.statsItemTable.entrySet().iterator();
+ while (it.hasNext()) {
+ Entry<String, MomentStatsItem> next = it.next();
+ next.getValue().printAtMinutes();
+ }
+ }
+
+ public void setValue(final String statsKey, final int value) {
+ MomentStatsItem statsItem = this.getAndCreateStatsItem(statsKey);
+ statsItem.getValue().set(value);
+ }
+
+ public MomentStatsItem getAndCreateStatsItem(final String statsKey) {
+ MomentStatsItem statsItem = this.statsItemTable.get(statsKey);
+ if (null == statsItem) {
+ statsItem =
+ new MomentStatsItem(this.statsName, statsKey, this.scheduledExecutorService, this.log);
+ MomentStatsItem prev = this.statsItemTable.put(statsKey, statsItem);
+
+ if (null == prev) {
+
+ // statsItem.init();
+ }
+ }
+
+ return statsItem;
+ }
+}