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 09:13:54 UTC
[03/99] [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/example/src/main/java/org/apache/rocketmq/example/simple/RandomAsyncCommit.java
----------------------------------------------------------------------
diff --git a/example/src/main/java/org/apache/rocketmq/example/simple/RandomAsyncCommit.java b/example/src/main/java/org/apache/rocketmq/example/simple/RandomAsyncCommit.java
new file mode 100644
index 0000000..cda7952
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/simple/RandomAsyncCommit.java
@@ -0,0 +1,60 @@
+/**
+ * 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.example.simple;
+
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+
+
+public class RandomAsyncCommit {
+ private final ConcurrentHashMap<MessageQueue, CachedQueue> mqCachedTable =
+ new ConcurrentHashMap<MessageQueue, CachedQueue>();
+
+
+ public void putMessages(final MessageQueue mq, final List<MessageExt> msgs) {
+ CachedQueue cachedQueue = this.mqCachedTable.get(mq);
+ if (null == cachedQueue) {
+ cachedQueue = new CachedQueue();
+ this.mqCachedTable.put(mq, cachedQueue);
+ }
+ for (MessageExt msg : msgs) {
+ cachedQueue.getMsgCachedTable().put(msg.getQueueOffset(), msg);
+ }
+ }
+
+
+ public void removeMessage(final MessageQueue mq, long offset) {
+ CachedQueue cachedQueue = this.mqCachedTable.get(mq);
+ if (null != cachedQueue) {
+ cachedQueue.getMsgCachedTable().remove(offset);
+ }
+ }
+
+
+ public long commitableOffset(final MessageQueue mq) {
+ CachedQueue cachedQueue = this.mqCachedTable.get(mq);
+ if (null != cachedQueue) {
+ return cachedQueue.getMsgCachedTable().firstKey();
+ }
+
+ return -1;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/example/src/main/java/org/apache/rocketmq/example/simple/TestProducer.java
----------------------------------------------------------------------
diff --git a/example/src/main/java/org/apache/rocketmq/example/simple/TestProducer.java b/example/src/main/java/org/apache/rocketmq/example/simple/TestProducer.java
new file mode 100644
index 0000000..0304a63
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/simple/TestProducer.java
@@ -0,0 +1,54 @@
+/**
+ * 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.example.simple;
+
+import org.apache.rocketmq.client.QueryResult;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
+
+public class TestProducer {
+ public static void main(String[] args) throws MQClientException, InterruptedException {
+ DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
+ producer.start();
+
+ for (int i = 0; i < 1; i++)
+ try {
+ {
+ Message msg = new Message("TopicTest1",
+ "TagA",
+ "key113",
+ "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
+ SendResult sendResult = producer.send(msg);
+ System.out.printf("%s%n", sendResult);
+
+ QueryResult queryMessage =
+ producer.queryMessage("TopicTest1", "key113", 10, 0, System.currentTimeMillis());
+ for (MessageExt m : queryMessage.getMessageList()) {
+ System.out.printf("%s%n", m);
+ }
+ }
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ producer.shutdown();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/example/src/main/java/org/apache/rocketmq/example/transaction/TransactionCheckListenerImpl.java
----------------------------------------------------------------------
diff --git a/example/src/main/java/org/apache/rocketmq/example/transaction/TransactionCheckListenerImpl.java b/example/src/main/java/org/apache/rocketmq/example/transaction/TransactionCheckListenerImpl.java
new file mode 100644
index 0000000..fea93a8
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/transaction/TransactionCheckListenerImpl.java
@@ -0,0 +1,45 @@
+/**
+ * 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.example.transaction;
+
+import org.apache.rocketmq.client.producer.LocalTransactionState;
+import org.apache.rocketmq.client.producer.TransactionCheckListener;
+import org.apache.rocketmq.common.message.MessageExt;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+
+public class TransactionCheckListenerImpl implements TransactionCheckListener {
+ private AtomicInteger transactionIndex = new AtomicInteger(0);
+
+
+ @Override
+ public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
+ System.out.printf("server checking TrMsg " + msg.toString() + "%n");
+
+ int value = transactionIndex.getAndIncrement();
+ if ((value % 6) == 0) {
+ throw new RuntimeException("Could not find db");
+ } else if ((value % 5) == 0) {
+ return LocalTransactionState.ROLLBACK_MESSAGE;
+ } else if ((value % 4) == 0) {
+ return LocalTransactionState.COMMIT_MESSAGE;
+ }
+
+ return LocalTransactionState.UNKNOW;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/example/src/main/java/org/apache/rocketmq/example/transaction/TransactionExecuterImpl.java
----------------------------------------------------------------------
diff --git a/example/src/main/java/org/apache/rocketmq/example/transaction/TransactionExecuterImpl.java b/example/src/main/java/org/apache/rocketmq/example/transaction/TransactionExecuterImpl.java
new file mode 100644
index 0000000..eb787fd
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/transaction/TransactionExecuterImpl.java
@@ -0,0 +1,43 @@
+/**
+ * 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.example.transaction;
+
+import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
+import org.apache.rocketmq.client.producer.LocalTransactionState;
+import org.apache.rocketmq.common.message.Message;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class TransactionExecuterImpl implements LocalTransactionExecuter {
+ private AtomicInteger transactionIndex = new AtomicInteger(1);
+
+
+ @Override
+ public LocalTransactionState executeLocalTransactionBranch(final Message msg, final Object arg) {
+ int value = transactionIndex.getAndIncrement();
+
+ if (value == 0) {
+ throw new RuntimeException("Could not find db");
+ } else if ((value % 5) == 0) {
+ return LocalTransactionState.ROLLBACK_MESSAGE;
+ } else if ((value % 4) == 0) {
+ return LocalTransactionState.COMMIT_MESSAGE;
+ }
+
+ return LocalTransactionState.UNKNOW;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/example/src/main/java/org/apache/rocketmq/example/transaction/TransactionProducer.java
----------------------------------------------------------------------
diff --git a/example/src/main/java/org/apache/rocketmq/example/transaction/TransactionProducer.java b/example/src/main/java/org/apache/rocketmq/example/transaction/TransactionProducer.java
new file mode 100644
index 0000000..5a868c6
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/transaction/TransactionProducer.java
@@ -0,0 +1,61 @@
+/**
+ * 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.example.transaction;
+
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.TransactionCheckListener;
+import org.apache.rocketmq.client.producer.TransactionMQProducer;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
+
+import java.io.UnsupportedEncodingException;
+
+public class TransactionProducer {
+ public static void main(String[] args) throws MQClientException, InterruptedException {
+ TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl();
+ TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
+ producer.setCheckThreadPoolMinSize(2);
+ producer.setCheckThreadPoolMaxSize(2);
+ producer.setCheckRequestHoldMax(2000);
+ producer.setTransactionCheckListener(transactionCheckListener);
+ producer.start();
+
+ String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
+ TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl();
+ for (int i = 0; i < 100; i++) {
+ try {
+ Message msg =
+ new Message("TopicTest", tags[i % tags.length], "KEY" + i,
+ ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
+ SendResult sendResult = producer.sendMessageInTransaction(msg, tranExecuter, null);
+ System.out.printf("%s%n", sendResult);
+
+ Thread.sleep(10);
+ } catch (MQClientException e) {
+ e.printStackTrace();
+ } catch (UnsupportedEncodingException e) {
+ e.printStackTrace();
+ }
+ }
+
+ for (int i = 0; i < 100000; i++) {
+ Thread.sleep(1000);
+ }
+ producer.shutdown();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/example/src/main/resources/MessageFilterImpl.java
----------------------------------------------------------------------
diff --git a/example/src/main/resources/MessageFilterImpl.java b/example/src/main/resources/MessageFilterImpl.java
index 2894a55..3ff3f48 100644
--- a/example/src/main/resources/MessageFilterImpl.java
+++ b/example/src/main/resources/MessageFilterImpl.java
@@ -15,10 +15,10 @@
* limitations under the License.
*/
-package com.alibaba.rocketmq.example.filter;
+package org.apache.rocketmq.example.filter;
-import com.alibaba.rocketmq.common.filter.MessageFilter;
-import com.alibaba.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.filter.MessageFilter;
+import org.apache.rocketmq.common.message.MessageExt;
public class MessageFilterImpl implements MessageFilter {
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/filtersrv/pom.xml
----------------------------------------------------------------------
diff --git a/filtersrv/pom.xml b/filtersrv/pom.xml
index 28c360b..bebd10a 100644
--- a/filtersrv/pom.xml
+++ b/filtersrv/pom.xml
@@ -18,7 +18,7 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
- <groupId>com.alibaba.rocketmq</groupId>
+ <groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
<version>4.0.0-SNAPSHOT</version>
</parent>
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FilterServerOuterAPI.java
----------------------------------------------------------------------
diff --git a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FilterServerOuterAPI.java b/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FilterServerOuterAPI.java
deleted file mode 100644
index b469b3f..0000000
--- a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FilterServerOuterAPI.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/**
- * 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 com.alibaba.rocketmq.filtersrv;
-
-import com.alibaba.rocketmq.client.exception.MQBrokerException;
-import com.alibaba.rocketmq.common.protocol.RequestCode;
-import com.alibaba.rocketmq.common.protocol.ResponseCode;
-import com.alibaba.rocketmq.common.protocol.header.filtersrv.RegisterFilterServerRequestHeader;
-import com.alibaba.rocketmq.common.protocol.header.filtersrv.RegisterFilterServerResponseHeader;
-import com.alibaba.rocketmq.remoting.RemotingClient;
-import com.alibaba.rocketmq.remoting.exception.RemotingCommandException;
-import com.alibaba.rocketmq.remoting.exception.RemotingConnectException;
-import com.alibaba.rocketmq.remoting.exception.RemotingSendRequestException;
-import com.alibaba.rocketmq.remoting.exception.RemotingTimeoutException;
-import com.alibaba.rocketmq.remoting.netty.NettyClientConfig;
-import com.alibaba.rocketmq.remoting.netty.NettyRemotingClient;
-import com.alibaba.rocketmq.remoting.protocol.RemotingCommand;
-
-
-/**
- * @author shijia.wxr
- */
-public class FilterServerOuterAPI {
- private final RemotingClient remotingClient;
-
-
- public FilterServerOuterAPI() {
- this.remotingClient = new NettyRemotingClient(new NettyClientConfig());
- }
-
-
- public void start() {
- this.remotingClient.start();
- }
-
-
- public void shutdown() {
- this.remotingClient.shutdown();
- }
-
-
- public RegisterFilterServerResponseHeader registerFilterServerToBroker(
- final String brokerAddr,
- final String filterServerAddr
- ) throws RemotingCommandException, RemotingConnectException, RemotingSendRequestException,
- RemotingTimeoutException, InterruptedException, MQBrokerException {
- RegisterFilterServerRequestHeader requestHeader = new RegisterFilterServerRequestHeader();
- requestHeader.setFilterServerAddr(filterServerAddr);
- RemotingCommand request =
- RemotingCommand.createRequestCommand(RequestCode.REGISTER_FILTER_SERVER, requestHeader);
-
- RemotingCommand response = this.remotingClient.invokeSync(brokerAddr, request, 3000);
- assert response != null;
- switch (response.getCode()) {
- case ResponseCode.SUCCESS: {
- RegisterFilterServerResponseHeader responseHeader =
- (RegisterFilterServerResponseHeader) response
- .decodeCommandCustomHeader(RegisterFilterServerResponseHeader.class);
-
- return responseHeader;
- }
- default:
- break;
- }
-
- throw new MQBrokerException(response.getCode(), response.getRemark());
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FiltersrvConfig.java
----------------------------------------------------------------------
diff --git a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FiltersrvConfig.java b/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FiltersrvConfig.java
deleted file mode 100644
index fac620f..0000000
--- a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FiltersrvConfig.java
+++ /dev/null
@@ -1,158 +0,0 @@
-/**
- * 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 com.alibaba.rocketmq.filtersrv;
-
-import com.alibaba.rocketmq.common.MixAll;
-import com.alibaba.rocketmq.common.annotation.ImportantField;
-import com.alibaba.rocketmq.remoting.common.RemotingUtil;
-
-
-public class FiltersrvConfig {
- private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY,
- System.getenv(MixAll.ROCKETMQ_HOME_ENV));
-
- @ImportantField
- private String namesrvAddr = System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY,
- System.getenv(MixAll.NAMESRV_ADDR_ENV));
-
- private String connectWhichBroker = "127.0.0.1:10911";
- private String filterServerIP = RemotingUtil.getLocalAddress();
-
- private int compressMsgBodyOverHowmuch = 1024 * 8;
- private int zipCompressLevel = 5;
-
-
- private boolean clientUploadFilterClassEnable = true;
-
-
- private String filterClassRepertoryUrl = "http://fsrep.tbsite.net/filterclass";
-
- private int fsServerAsyncSemaphoreValue = 2048;
- private int fsServerCallbackExecutorThreads = 64;
- private int fsServerWorkerThreads = 64;
-
-
- public String getRocketmqHome() {
- return rocketmqHome;
- }
-
-
- public void setRocketmqHome(String rocketmqHome) {
- this.rocketmqHome = rocketmqHome;
- }
-
-
- public String getNamesrvAddr() {
- return namesrvAddr;
- }
-
-
- public void setNamesrvAddr(String namesrvAddr) {
- this.namesrvAddr = namesrvAddr;
- }
-
-
- public String getConnectWhichBroker() {
- return connectWhichBroker;
- }
-
-
- public void setConnectWhichBroker(String connectWhichBroker) {
- this.connectWhichBroker = connectWhichBroker;
- }
-
-
- public String getFilterServerIP() {
- return filterServerIP;
- }
-
-
- public void setFilterServerIP(String filterServerIP) {
- this.filterServerIP = filterServerIP;
- }
-
-
- public int getCompressMsgBodyOverHowmuch() {
- return compressMsgBodyOverHowmuch;
- }
-
-
- public void setCompressMsgBodyOverHowmuch(int compressMsgBodyOverHowmuch) {
- this.compressMsgBodyOverHowmuch = compressMsgBodyOverHowmuch;
- }
-
-
- public int getZipCompressLevel() {
- return zipCompressLevel;
- }
-
-
- public void setZipCompressLevel(int zipCompressLevel) {
- this.zipCompressLevel = zipCompressLevel;
- }
-
-
- public boolean isClientUploadFilterClassEnable() {
- return clientUploadFilterClassEnable;
- }
-
-
- public void setClientUploadFilterClassEnable(boolean clientUploadFilterClassEnable) {
- this.clientUploadFilterClassEnable = clientUploadFilterClassEnable;
- }
-
-
- public String getFilterClassRepertoryUrl() {
- return filterClassRepertoryUrl;
- }
-
-
- public void setFilterClassRepertoryUrl(String filterClassRepertoryUrl) {
- this.filterClassRepertoryUrl = filterClassRepertoryUrl;
- }
-
-
- public int getFsServerAsyncSemaphoreValue() {
- return fsServerAsyncSemaphoreValue;
- }
-
-
- public void setFsServerAsyncSemaphoreValue(int fsServerAsyncSemaphoreValue) {
- this.fsServerAsyncSemaphoreValue = fsServerAsyncSemaphoreValue;
- }
-
-
- public int getFsServerCallbackExecutorThreads() {
- return fsServerCallbackExecutorThreads;
- }
-
-
- public void setFsServerCallbackExecutorThreads(int fsServerCallbackExecutorThreads) {
- this.fsServerCallbackExecutorThreads = fsServerCallbackExecutorThreads;
- }
-
-
- public int getFsServerWorkerThreads() {
- return fsServerWorkerThreads;
- }
-
-
- public void setFsServerWorkerThreads(int fsServerWorkerThreads) {
- this.fsServerWorkerThreads = fsServerWorkerThreads;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FiltersrvController.java
----------------------------------------------------------------------
diff --git a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FiltersrvController.java b/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FiltersrvController.java
deleted file mode 100644
index 0e3f696..0000000
--- a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FiltersrvController.java
+++ /dev/null
@@ -1,226 +0,0 @@
-/**
- * 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 com.alibaba.rocketmq.filtersrv;
-
-import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
-import com.alibaba.rocketmq.common.MixAll;
-import com.alibaba.rocketmq.common.ThreadFactoryImpl;
-import com.alibaba.rocketmq.common.UtilAll;
-import com.alibaba.rocketmq.common.constant.LoggerName;
-import com.alibaba.rocketmq.common.protocol.header.filtersrv.RegisterFilterServerResponseHeader;
-import com.alibaba.rocketmq.filtersrv.filter.FilterClassManager;
-import com.alibaba.rocketmq.filtersrv.processor.DefaultRequestProcessor;
-import com.alibaba.rocketmq.filtersrv.stats.FilterServerStatsManager;
-import com.alibaba.rocketmq.remoting.RemotingServer;
-import com.alibaba.rocketmq.remoting.netty.NettyRemotingServer;
-import com.alibaba.rocketmq.remoting.netty.NettyServerConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-
-/**
- * @author shijia.wxr
- */
-public class FiltersrvController {
- private static final Logger log = LoggerFactory.getLogger(LoggerName.FILTERSRV_LOGGER_NAME);
-
- private final FiltersrvConfig filtersrvConfig;
-
- private final NettyServerConfig nettyServerConfig;
- private final FilterClassManager filterClassManager;
-
- private final FilterServerOuterAPI filterServerOuterAPI = new FilterServerOuterAPI();
- private final DefaultMQPullConsumer defaultMQPullConsumer = new DefaultMQPullConsumer(
- MixAll.FILTERSRV_CONSUMER_GROUP);
-
- private final ScheduledExecutorService scheduledExecutorService = Executors
- .newSingleThreadScheduledExecutor(new ThreadFactoryImpl("FSScheduledThread"));
- private final FilterServerStatsManager filterServerStatsManager = new FilterServerStatsManager();
-
- private RemotingServer remotingServer;
-
- private ExecutorService remotingExecutor;
- private volatile String brokerName = null;
-
-
- public FiltersrvController(FiltersrvConfig filtersrvConfig, NettyServerConfig nettyServerConfig) {
- this.filtersrvConfig = filtersrvConfig;
- this.nettyServerConfig = nettyServerConfig;
- this.filterClassManager = new FilterClassManager(this);
- }
-
-
- public boolean initialize() {
-
- MixAll.printObjectProperties(log, this.filtersrvConfig);
-
-
- this.remotingServer = new NettyRemotingServer(this.nettyServerConfig);
-
-
- this.remotingExecutor =
- Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(),
- new ThreadFactoryImpl("RemotingExecutorThread_"));
-
- this.registerProcessor();
-
-
- this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
-
- @Override
- public void run() {
- FiltersrvController.this.registerFilterServerToBroker();
- }
- }, 3, 10, TimeUnit.SECONDS);
-
- this.defaultMQPullConsumer.setBrokerSuspendMaxTimeMillis(this.defaultMQPullConsumer
- .getBrokerSuspendMaxTimeMillis() - 1000);
- this.defaultMQPullConsumer.setConsumerTimeoutMillisWhenSuspend(this.defaultMQPullConsumer
- .getConsumerTimeoutMillisWhenSuspend() - 1000);
-
- this.defaultMQPullConsumer.setNamesrvAddr(this.filtersrvConfig.getNamesrvAddr());
- this.defaultMQPullConsumer.setInstanceName(String.valueOf(UtilAll.getPid()));
-
- return true;
- }
-
- private void registerProcessor() {
- this.remotingServer
- .registerDefaultProcessor(new DefaultRequestProcessor(this), this.remotingExecutor);
- }
-
- public void registerFilterServerToBroker() {
- try {
- RegisterFilterServerResponseHeader responseHeader =
- this.filterServerOuterAPI.registerFilterServerToBroker(
- this.filtersrvConfig.getConnectWhichBroker(), this.localAddr());
- this.defaultMQPullConsumer.getDefaultMQPullConsumerImpl().getPullAPIWrapper()
- .setDefaultBrokerId(responseHeader.getBrokerId());
-
- if (null == this.brokerName) {
- this.brokerName = responseHeader.getBrokerName();
- }
-
- log.info("register filter server<{}> to broker<{}> OK, Return: {} {}",
- this.localAddr(),
- this.filtersrvConfig.getConnectWhichBroker(),
- responseHeader.getBrokerName(),
- responseHeader.getBrokerId());
- } catch (Exception e) {
- log.warn("register filter server Exception", e);
-
- log.warn("access broker failed, kill oneself");
- System.exit(-1);
- }
- }
-
- public String localAddr() {
- return String.format("%s:%d", this.filtersrvConfig.getFilterServerIP(),
- this.remotingServer.localListenPort());
- }
-
- public void start() throws Exception {
- this.defaultMQPullConsumer.start();
- this.remotingServer.start();
- this.filterServerOuterAPI.start();
- this.defaultMQPullConsumer.getDefaultMQPullConsumerImpl().getPullAPIWrapper()
- .setConnectBrokerByUser(true);
- this.filterClassManager.start();
- this.filterServerStatsManager.start();
- }
-
-
- public void shutdown() {
- this.remotingServer.shutdown();
- this.remotingExecutor.shutdown();
- this.scheduledExecutorService.shutdown();
- this.defaultMQPullConsumer.shutdown();
- this.filterServerOuterAPI.shutdown();
- this.filterClassManager.shutdown();
- this.filterServerStatsManager.shutdown();
- }
-
-
- public RemotingServer getRemotingServer() {
- return remotingServer;
- }
-
-
- public void setRemotingServer(RemotingServer remotingServer) {
- this.remotingServer = remotingServer;
- }
-
-
- public ExecutorService getRemotingExecutor() {
- return remotingExecutor;
- }
-
-
- public void setRemotingExecutor(ExecutorService remotingExecutor) {
- this.remotingExecutor = remotingExecutor;
- }
-
-
- public FiltersrvConfig getFiltersrvConfig() {
- return filtersrvConfig;
- }
-
-
- public NettyServerConfig getNettyServerConfig() {
- return nettyServerConfig;
- }
-
-
- public ScheduledExecutorService getScheduledExecutorService() {
- return scheduledExecutorService;
- }
-
-
- public FilterServerOuterAPI getFilterServerOuterAPI() {
- return filterServerOuterAPI;
- }
-
-
- public FilterClassManager getFilterClassManager() {
- return filterClassManager;
- }
-
-
- public DefaultMQPullConsumer getDefaultMQPullConsumer() {
- return defaultMQPullConsumer;
- }
-
-
- public String getBrokerName() {
- return brokerName;
- }
-
-
- public void setBrokerName(String brokerName) {
- this.brokerName = brokerName;
- }
-
-
- public FilterServerStatsManager getFilterServerStatsManager() {
- return filterServerStatsManager;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FiltersrvStartup.java
----------------------------------------------------------------------
diff --git a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FiltersrvStartup.java b/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FiltersrvStartup.java
deleted file mode 100644
index 3fe6b22..0000000
--- a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/FiltersrvStartup.java
+++ /dev/null
@@ -1,183 +0,0 @@
-/**
- * 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 com.alibaba.rocketmq.filtersrv;
-
-import ch.qos.logback.classic.LoggerContext;
-import ch.qos.logback.classic.joran.JoranConfigurator;
-import com.alibaba.rocketmq.common.MQVersion;
-import com.alibaba.rocketmq.common.MixAll;
-import com.alibaba.rocketmq.common.constant.LoggerName;
-import com.alibaba.rocketmq.remoting.netty.NettyServerConfig;
-import com.alibaba.rocketmq.remoting.netty.NettySystemConfig;
-import com.alibaba.rocketmq.remoting.protocol.RemotingCommand;
-import com.alibaba.rocketmq.srvutil.ServerUtil;
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.PosixParser;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.BufferedInputStream;
-import java.io.FileInputStream;
-import java.io.InputStream;
-import java.util.Properties;
-import java.util.concurrent.atomic.AtomicInteger;
-
-
-/**
- * @author shijia.wxr
- */
-public class FiltersrvStartup {
- public static Logger log;
-
- public static void main(String[] args) {
- start(createController(args));
- }
-
- public static FiltersrvController start(FiltersrvController controller) {
-
- try {
- controller.start();
- } catch (Exception e) {
- e.printStackTrace();
- System.exit(-1);
- }
-
- String tip = "The Filter Server boot success, " + controller.localAddr();
- log.info(tip);
- System.out.printf("%s%n", tip);
-
- return controller;
- }
-
- public static FiltersrvController createController(String[] args) {
- System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
-
-
- if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE)) {
- NettySystemConfig.socketSndbufSize = 65535;
- }
-
-
- if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE)) {
- NettySystemConfig.socketRcvbufSize = 1024;
- }
-
- try {
- Options options = ServerUtil.buildCommandlineOptions(new Options());
- final CommandLine commandLine =
- ServerUtil.parseCmdLine("mqfiltersrv", args, buildCommandlineOptions(options),
- new PosixParser());
- if (null == commandLine) {
- System.exit(-1);
- return null;
- }
-
- final FiltersrvConfig filtersrvConfig = new FiltersrvConfig();
- final NettyServerConfig nettyServerConfig = new NettyServerConfig();
-
- if (commandLine.hasOption('c')) {
- String file = commandLine.getOptionValue('c');
- if (file != null) {
- InputStream in = new BufferedInputStream(new FileInputStream(file));
- Properties properties = new Properties();
- properties.load(in);
- MixAll.properties2Object(properties, filtersrvConfig);
- System.out.printf("load config properties file OK, " + file + "%n");
- in.close();
-
- String port = properties.getProperty("listenPort");
- if (port != null) {
- filtersrvConfig.setConnectWhichBroker(String.format("127.0.0.1:%s", port));
- }
- }
- }
-
- nettyServerConfig.setListenPort(0);
- nettyServerConfig.setServerAsyncSemaphoreValue(filtersrvConfig.getFsServerAsyncSemaphoreValue());
- nettyServerConfig.setServerCallbackExecutorThreads(filtersrvConfig
- .getFsServerCallbackExecutorThreads());
- nettyServerConfig.setServerWorkerThreads(filtersrvConfig.getFsServerWorkerThreads());
-
- if (commandLine.hasOption('p')) {
- MixAll.printObjectProperties(null, filtersrvConfig);
- MixAll.printObjectProperties(null, nettyServerConfig);
- System.exit(0);
- }
-
- MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), filtersrvConfig);
- if (null == filtersrvConfig.getRocketmqHome()) {
- System.out.printf("Please set the " + MixAll.ROCKETMQ_HOME_ENV
- + " variable in your environment to match the location of the RocketMQ installation%n");
- System.exit(-2);
- }
-
- LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
- JoranConfigurator configurator = new JoranConfigurator();
- configurator.setContext(lc);
- lc.reset();
- configurator.doConfigure(filtersrvConfig.getRocketmqHome() + "/conf/logback_filtersrv.xml");
- log = LoggerFactory.getLogger(LoggerName.FILTERSRV_LOGGER_NAME);
-
- final FiltersrvController controller =
- new FiltersrvController(filtersrvConfig, nettyServerConfig);
- boolean initResult = controller.initialize();
- if (!initResult) {
- controller.shutdown();
- System.exit(-3);
- }
-
- Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
- private volatile boolean hasShutdown = false;
- private AtomicInteger shutdownTimes = new AtomicInteger(0);
-
- @Override
- public void run() {
- synchronized (this) {
- log.info("shutdown hook was invoked, " + this.shutdownTimes.incrementAndGet());
- if (!this.hasShutdown) {
- this.hasShutdown = true;
- long begineTime = System.currentTimeMillis();
- controller.shutdown();
- long consumingTimeTotal = System.currentTimeMillis() - begineTime;
- log.info("shutdown hook over, consuming time total(ms): " + consumingTimeTotal);
- }
- }
- }
- }, "ShutdownHook"));
-
- return controller;
- } catch (Throwable e) {
- e.printStackTrace();
- System.exit(-1);
- }
- return null;
- }
-
- public static Options buildCommandlineOptions(final Options options) {
- Option opt = new Option("c", "configFile", true, "Filter server config properties file");
- opt.setRequired(false);
- options.addOption(opt);
-
- opt = new Option("p", "printConfigItem", false, "Print all config item");
- opt.setRequired(false);
- options.addOption(opt);
-
- return options;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/DynaCode.java
----------------------------------------------------------------------
diff --git a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/DynaCode.java b/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/DynaCode.java
deleted file mode 100644
index e17e5d2..0000000
--- a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/DynaCode.java
+++ /dev/null
@@ -1,393 +0,0 @@
-/**
- * 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 com.alibaba.rocketmq.filtersrv.filter;
-
-import com.alibaba.rocketmq.common.UtilAll;
-import com.alibaba.rocketmq.common.constant.LoggerName;
-import com.alibaba.rocketmq.common.filter.FilterAPI;
-import com.alibaba.rocketmq.remoting.common.RemotingHelper;
-import org.apache.commons.lang3.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.tools.JavaCompiler;
-import javax.tools.ToolProvider;
-import java.io.*;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.net.URLClassLoader;
-import java.net.URLDecoder;
-import java.util.*;
-
-
-public class DynaCode {
- private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.FILTERSRV_LOGGER_NAME);
-
- private static final String FILE_SP = System.getProperty("file.separator");
-
- private static final String LINE_SP = System.getProperty("line.separator");
-
- private String sourcePath = System.getProperty("user.home") + FILE_SP + "rocketmq_filter_class" + FILE_SP
- + UtilAll.getPid();
-
- private String outPutClassPath = sourcePath;
-
-
- private ClassLoader parentClassLoader;
-
-
- private List<String> codeStrs;
-
-
- private Map<String/* fullClassName */, Class<?>/* class */> loadClass;
-
-
- private String classpath;
-
-
- private String bootclasspath;
-
-
- private String extdirs;
-
-
- private String encoding = "UTF-8";
-
-
- private String target;
-
-
- @SuppressWarnings("unchecked")
- public DynaCode(String code) {
- this(Thread.currentThread().getContextClassLoader(), Arrays.asList(code));
- }
-
-
- public DynaCode(ClassLoader parentClassLoader, List<String> codeStrs) {
- this(extractClasspath(parentClassLoader), parentClassLoader, codeStrs);
- }
-
-
- public DynaCode(String classpath, ClassLoader parentClassLoader, List<String> codeStrs) {
- this.classpath = classpath;
- this.parentClassLoader = parentClassLoader;
- this.codeStrs = codeStrs;
- this.loadClass = new HashMap<String, Class<?>>(codeStrs.size());
- }
-
-
- private static String extractClasspath(ClassLoader cl) {
- StringBuffer buf = new StringBuffer();
- while (cl != null) {
- if (cl instanceof URLClassLoader) {
- URL urls[] = ((URLClassLoader) cl).getURLs();
- for (int i = 0; i < urls.length; i++) {
- if (buf.length() > 0) {
- buf.append(File.pathSeparatorChar);
- }
- String s = urls[i].getFile();
- try {
- s = URLDecoder.decode(s, "UTF-8");
- } catch (UnsupportedEncodingException e) {
- continue;
- }
- File f = new File(s);
- buf.append(f.getAbsolutePath());
- }
- }
- cl = cl.getParent();
- }
- return buf.toString();
- }
-
-
- public DynaCode(List<String> codeStrs) {
- this(Thread.currentThread().getContextClassLoader(), codeStrs);
- }
-
- public static Class<?> compileAndLoadClass(final String className, final String javaSource)
- throws Exception {
- String classSimpleName = FilterAPI.simpleClassName(className);
- String javaCode = javaSource;
-
- final String newClassSimpleName = classSimpleName + System.currentTimeMillis();
- String newJavaCode = javaCode.replaceAll(classSimpleName, newClassSimpleName);
-
- List<String> codes = new ArrayList<String>();
- codes.add(newJavaCode);
- DynaCode dc = new DynaCode(codes);
- dc.compileAndLoadClass();
- Map<String, Class<?>> map = dc.getLoadClass();
-
- Class<?> clazz = map.get(getQualifiedName(newJavaCode));
- return clazz;
- }
-
- public void compileAndLoadClass() throws Exception {
- String[] sourceFiles = this.uploadSrcFile();
- this.compile(sourceFiles);
- this.loadClass(this.loadClass.keySet());
- }
-
- public Map<String, Class<?>> getLoadClass() {
- return loadClass;
- }
-
- public static String getQualifiedName(String code) {
- StringBuilder sb = new StringBuilder();
- String className = getClassName(code);
- if (StringUtils.isNotBlank(className)) {
-
- String packageName = getPackageName(code);
- if (StringUtils.isNotBlank(packageName)) {
- sb.append(packageName).append(".");
- }
- sb.append(className);
- }
- return sb.toString();
- }
-
- private String[] uploadSrcFile() throws Exception {
- List<String> srcFileAbsolutePaths = new ArrayList<String>(codeStrs.size());
- for (String code : codeStrs) {
- if (StringUtils.isNotBlank(code)) {
- String packageName = getPackageName(code);
- String className = getClassName(code);
- if (StringUtils.isNotBlank(className)) {
- File srcFile = null;
- BufferedWriter bufferWriter = null;
- try {
- if (StringUtils.isBlank(packageName)) {
- File pathFile = new File(sourcePath);
-
- if (!pathFile.exists()) {
- if (!pathFile.mkdirs()) {
- throw new RuntimeException("create PathFile Error!");
- }
- }
- srcFile = new File(sourcePath + FILE_SP + className + ".java");
- } else {
- String srcPath = StringUtils.replace(packageName, ".", FILE_SP);
- File pathFile = new File(sourcePath + FILE_SP + srcPath);
-
- if (!pathFile.exists()) {
- if (!pathFile.mkdirs()) {
- throw new RuntimeException("create PathFile Error!");
- }
- }
- srcFile = new File(pathFile.getAbsolutePath() + FILE_SP + className + ".java");
- }
- synchronized (loadClass) {
- loadClass.put(getFullClassName(code), null);
- }
- if (null != srcFile) {
- LOGGER.warn("Dyna Create Java Source File:---->" + srcFile.getAbsolutePath());
- srcFileAbsolutePaths.add(srcFile.getAbsolutePath());
- srcFile.deleteOnExit();
- }
- OutputStreamWriter outputStreamWriter =
- new OutputStreamWriter(new FileOutputStream(srcFile), encoding);
- bufferWriter = new BufferedWriter(outputStreamWriter);
- for (String lineCode : code.split(LINE_SP)) {
- bufferWriter.write(lineCode);
- bufferWriter.newLine();
- }
- bufferWriter.flush();
- } finally {
- if (null != bufferWriter) {
- bufferWriter.close();
- }
- }
- }
- }
- }
- return srcFileAbsolutePaths.toArray(new String[srcFileAbsolutePaths.size()]);
- }
-
- private void compile(String[] srcFiles) throws Exception {
- String args[] = this.buildCompileJavacArgs(srcFiles);
- ByteArrayOutputStream err = new ByteArrayOutputStream();
- JavaCompiler compiler = ToolProvider.getSystemJavaCompiler();
- if (compiler == null) {
- throw new NullPointerException(
- "ToolProvider.getSystemJavaCompiler() return null,please use JDK replace JRE!");
- }
- int resultCode = compiler.run(null, null, err, args);
- if (resultCode != 0) {
- throw new Exception(err.toString(RemotingHelper.DEFAULT_CHARSET));
- }
- }
-
- private void loadClass(Set<String> classFullNames) throws ClassNotFoundException, MalformedURLException {
- synchronized (loadClass) {
- ClassLoader classLoader =
- new URLClassLoader(new URL[]{new File(outPutClassPath).toURI().toURL()},
- parentClassLoader);
- for (String key : classFullNames) {
- Class<?> classz = classLoader.loadClass(key);
- if (null != classz) {
- loadClass.put(key, classz);
- LOGGER.info("Dyna Load Java Class File OK:----> className: " + key);
- } else {
- LOGGER.error("Dyna Load Java Class File Fail:----> className: " + key);
- }
- }
- }
- }
-
- public static String getClassName(String code) {
- String className = StringUtils.substringBefore(code, "{");
- if (StringUtils.isBlank(className)) {
- return className;
- }
- if (StringUtils.contains(code, " class ")) {
- className = StringUtils.substringAfter(className, " class ");
- if (StringUtils.contains(className, " extends ")) {
- className = StringUtils.substringBefore(className, " extends ").trim();
- } else if (StringUtils.contains(className, " implements ")) {
- className = StringUtils.trim(StringUtils.substringBefore(className, " implements "));
- } else {
- className = StringUtils.trim(className);
- }
- } else if (StringUtils.contains(code, " interface ")) {
- className = StringUtils.substringAfter(className, " interface ");
- if (StringUtils.contains(className, " extends ")) {
- className = StringUtils.substringBefore(className, " extends ").trim();
- } else {
- className = StringUtils.trim(className);
- }
- } else if (StringUtils.contains(code, " enum ")) {
- className = StringUtils.trim(StringUtils.substringAfter(className, " enum "));
- } else {
- return StringUtils.EMPTY;
- }
- return className;
- }
-
- public static String getPackageName(String code) {
- String packageName =
- StringUtils.substringBefore(StringUtils.substringAfter(code, "package "), ";").trim();
- return packageName;
- }
-
- public static String getFullClassName(String code) {
- String packageName = getPackageName(code);
- String className = getClassName(code);
- return StringUtils.isBlank(packageName) ? className : packageName + "." + className;
- }
-
- private String[] buildCompileJavacArgs(String srcFiles[]) {
- ArrayList<String> args = new ArrayList<String>();
- if (StringUtils.isNotBlank(classpath)) {
- args.add("-classpath");
- args.add(classpath);
- }
- if (StringUtils.isNotBlank(outPutClassPath)) {
- args.add("-d");
- args.add(outPutClassPath);
- }
- if (StringUtils.isNotBlank(sourcePath)) {
- args.add("-sourcepath");
- args.add(sourcePath);
- }
- if (StringUtils.isNotBlank(bootclasspath)) {
- args.add("-bootclasspath");
- args.add(bootclasspath);
- }
- if (StringUtils.isNotBlank(extdirs)) {
- args.add("-extdirs");
- args.add(extdirs);
- }
- if (StringUtils.isNotBlank(encoding)) {
- args.add("-encoding");
- args.add(encoding);
- }
- if (StringUtils.isNotBlank(target)) {
- args.add("-target");
- args.add(target);
- }
- for (int i = 0; i < srcFiles.length; i++) {
- args.add(srcFiles[i]);
- }
- return args.toArray(new String[args.size()]);
- }
-
- public String getOutPutClassPath() {
- return outPutClassPath;
- }
-
- public void setOutPutClassPath(String outPutClassPath) {
- this.outPutClassPath = outPutClassPath;
- }
-
- public String getSourcePath() {
- return sourcePath;
- }
-
- public void setSourcePath(String sourcePath) {
- this.sourcePath = sourcePath;
- }
-
- public ClassLoader getParentClassLoader() {
- return parentClassLoader;
- }
-
- public void setParentClassLoader(ClassLoader parentClassLoader) {
- this.parentClassLoader = parentClassLoader;
- }
-
- public String getClasspath() {
- return classpath;
- }
-
- public void setClasspath(String classpath) {
- this.classpath = classpath;
- }
-
- public String getBootclasspath() {
- return bootclasspath;
- }
-
- public void setBootclasspath(String bootclasspath) {
- this.bootclasspath = bootclasspath;
- }
-
- public String getExtdirs() {
- return extdirs;
- }
-
- public void setExtdirs(String extdirs) {
- this.extdirs = extdirs;
- }
-
- public String getEncoding() {
- return encoding;
- }
-
- public void setEncoding(String encoding) {
- this.encoding = encoding;
- }
-
- public String getTarget() {
- return target;
- }
-
- public void setTarget(String target) {
- this.target = target;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassFetchMethod.java
----------------------------------------------------------------------
diff --git a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassFetchMethod.java b/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassFetchMethod.java
deleted file mode 100644
index 27b19ce..0000000
--- a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassFetchMethod.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/**
- * 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 com.alibaba.rocketmq.filtersrv.filter;
-
-public interface FilterClassFetchMethod {
- public String fetch(final String topic, final String consumerGroup, final String className);
-}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassInfo.java
----------------------------------------------------------------------
diff --git a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassInfo.java b/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassInfo.java
deleted file mode 100644
index f3e747e..0000000
--- a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassInfo.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/**
- * 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 com.alibaba.rocketmq.filtersrv.filter;
-
-import com.alibaba.rocketmq.common.filter.MessageFilter;
-
-
-public class FilterClassInfo {
- private String className;
- private int classCRC;
- private MessageFilter messageFilter;
-
-
- public int getClassCRC() {
- return classCRC;
- }
-
-
- public void setClassCRC(int classCRC) {
- this.classCRC = classCRC;
- }
-
-
- public MessageFilter getMessageFilter() {
- return messageFilter;
- }
-
-
- public void setMessageFilter(MessageFilter messageFilter) {
- this.messageFilter = messageFilter;
- }
-
-
- public String getClassName() {
- return className;
- }
-
-
- public void setClassName(String className) {
- this.className = className;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassLoader.java
----------------------------------------------------------------------
diff --git a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassLoader.java b/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassLoader.java
deleted file mode 100644
index 8966ca2..0000000
--- a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassLoader.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/**
- * 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 com.alibaba.rocketmq.filtersrv.filter;
-
-public class FilterClassLoader extends ClassLoader {
- public final Class<?> createNewClass(String name, byte[] b, int off, int len) throws ClassFormatError {
- return this.defineClass(name, b, off, len);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassManager.java
----------------------------------------------------------------------
diff --git a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassManager.java b/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassManager.java
deleted file mode 100644
index 618db8e..0000000
--- a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/FilterClassManager.java
+++ /dev/null
@@ -1,175 +0,0 @@
-/**
- * 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 com.alibaba.rocketmq.filtersrv.filter;
-
-import com.alibaba.rocketmq.common.MixAll;
-import com.alibaba.rocketmq.common.ThreadFactoryImpl;
-import com.alibaba.rocketmq.common.UtilAll;
-import com.alibaba.rocketmq.common.constant.LoggerName;
-import com.alibaba.rocketmq.common.filter.MessageFilter;
-import com.alibaba.rocketmq.filtersrv.FiltersrvController;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Iterator;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-
-public class FilterClassManager {
- private static final Logger log = LoggerFactory.getLogger(LoggerName.FILTERSRV_LOGGER_NAME);
-
- private final Object compileLock = new Object();
- private final FiltersrvController filtersrvController;
-
- private final ScheduledExecutorService scheduledExecutorService = Executors
- .newSingleThreadScheduledExecutor(new ThreadFactoryImpl("FSGetClassScheduledThread"));
- private ConcurrentHashMap<String/* topic@consumerGroup */, FilterClassInfo> filterClassTable =
- new ConcurrentHashMap<String, FilterClassInfo>(128);
- private FilterClassFetchMethod filterClassFetchMethod;
-
-
- public FilterClassManager(FiltersrvController filtersrvController) {
- this.filtersrvController = filtersrvController;
- this.filterClassFetchMethod =
- new HttpFilterClassFetchMethod(this.filtersrvController.getFiltersrvConfig()
- .getFilterClassRepertoryUrl());
- }
-
-
- public void start() {
- if (!this.filtersrvController.getFiltersrvConfig().isClientUploadFilterClassEnable()) {
- this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
-
- @Override
- public void run() {
- fetchClassFromRemoteHost();
- }
- }, 1, 1, TimeUnit.MINUTES);
- }
- }
-
- private void fetchClassFromRemoteHost() {
- Iterator<Entry<String, FilterClassInfo>> it = this.filterClassTable.entrySet().iterator();
- while (it.hasNext()) {
- try {
- Entry<String, FilterClassInfo> next = it.next();
- FilterClassInfo filterClassInfo = next.getValue();
- String[] topicAndGroup = next.getKey().split("@");
- String responseStr =
- this.filterClassFetchMethod.fetch(topicAndGroup[0], topicAndGroup[1],
- filterClassInfo.getClassName());
- byte[] filterSourceBinary = responseStr.getBytes("UTF-8");
- int classCRC = UtilAll.crc32(responseStr.getBytes("UTF-8"));
- if (classCRC != filterClassInfo.getClassCRC()) {
- String javaSource = new String(filterSourceBinary, MixAll.DEFAULT_CHARSET);
- Class<?> newClass =
- DynaCode.compileAndLoadClass(filterClassInfo.getClassName(), javaSource);
- Object newInstance = newClass.newInstance();
- filterClassInfo.setMessageFilter((MessageFilter) newInstance);
- filterClassInfo.setClassCRC(classCRC);
-
- log.info("fetch Remote class File OK, {} {}", next.getKey(),
- filterClassInfo.getClassName());
- }
- } catch (Exception e) {
- log.error("fetchClassFromRemoteHost Exception", e);
- }
- }
- }
-
- public void shutdown() {
- this.scheduledExecutorService.shutdown();
- }
-
- public boolean registerFilterClass(final String consumerGroup, final String topic,
- final String className, final int classCRC, final byte[] filterSourceBinary) {
- final String key = buildKey(consumerGroup, topic);
-
-
- boolean registerNew = false;
- FilterClassInfo filterClassInfoPrev = this.filterClassTable.get(key);
- if (null == filterClassInfoPrev) {
- registerNew = true;
- } else {
- if (this.filtersrvController.getFiltersrvConfig().isClientUploadFilterClassEnable()) {
- if (filterClassInfoPrev.getClassCRC() != classCRC && classCRC != 0) {
- registerNew = true;
- }
- }
- }
-
- if (registerNew) {
- synchronized (this.compileLock) {
- filterClassInfoPrev = this.filterClassTable.get(key);
- if (null != filterClassInfoPrev && filterClassInfoPrev.getClassCRC() == classCRC) {
- return true;
- }
-
- try {
-
- FilterClassInfo filterClassInfoNew = new FilterClassInfo();
- filterClassInfoNew.setClassName(className);
- filterClassInfoNew.setClassCRC(0);
- filterClassInfoNew.setMessageFilter(null);
-
- if (this.filtersrvController.getFiltersrvConfig().isClientUploadFilterClassEnable()) {
- String javaSource = new String(filterSourceBinary, MixAll.DEFAULT_CHARSET);
- Class<?> newClass = DynaCode.compileAndLoadClass(className, javaSource);
- Object newInstance = newClass.newInstance();
- filterClassInfoNew.setMessageFilter((MessageFilter) newInstance);
- filterClassInfoNew.setClassCRC(classCRC);
- }
-
- this.filterClassTable.put(key, filterClassInfoNew);
- } catch (Throwable e) {
- String info =
- String
- .format(
- "FilterServer, registerFilterClass Exception, consumerGroup: %s topic: %s className: %s",
- consumerGroup, topic, className);
- log.error(info, e);
- return false;
- }
- }
- }
-
- return true;
- }
-
- private static String buildKey(final String consumerGroup, final String topic) {
- return topic + "@" + consumerGroup;
- }
-
- public FilterClassInfo findFilterClass(final String consumerGroup, final String topic) {
- return this.filterClassTable.get(buildKey(consumerGroup, topic));
- }
-
-
- public FilterClassFetchMethod getFilterClassFetchMethod() {
- return filterClassFetchMethod;
- }
-
-
- public void setFilterClassFetchMethod(FilterClassFetchMethod filterClassFetchMethod) {
- this.filterClassFetchMethod = filterClassFetchMethod;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/HttpFilterClassFetchMethod.java
----------------------------------------------------------------------
diff --git a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/HttpFilterClassFetchMethod.java b/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/HttpFilterClassFetchMethod.java
deleted file mode 100644
index 88cb572..0000000
--- a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/filter/HttpFilterClassFetchMethod.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/**
- * 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 com.alibaba.rocketmq.filtersrv.filter;
-
-import com.alibaba.rocketmq.common.constant.LoggerName;
-import com.alibaba.rocketmq.common.utils.HttpTinyClient;
-import com.alibaba.rocketmq.common.utils.HttpTinyClient.HttpResult;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-public class HttpFilterClassFetchMethod implements FilterClassFetchMethod {
- private static final Logger log = LoggerFactory.getLogger(LoggerName.FILTERSRV_LOGGER_NAME);
- private final String url;
-
-
- public HttpFilterClassFetchMethod(String url) {
- this.url = url;
- }
-
-
- @Override
- public String fetch(String topic, String consumerGroup, String className) {
- String thisUrl = String.format("%s/%s.java", this.url, className);
-
- try {
- HttpResult result = HttpTinyClient.httpGet(thisUrl, null, null, "UTF-8", 5000);
- if (200 == result.code) {
- return result.content;
- }
- } catch (Exception e) {
- log.error(
- String.format("call <%s> exception, Topic: %s Group: %s", thisUrl, topic, consumerGroup), e);
- }
-
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/de6f9416/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/processor/DefaultRequestProcessor.java
----------------------------------------------------------------------
diff --git a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/processor/DefaultRequestProcessor.java b/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/processor/DefaultRequestProcessor.java
deleted file mode 100644
index 105cfff..0000000
--- a/filtersrv/src/main/java/com/alibaba/rocketmq/filtersrv/processor/DefaultRequestProcessor.java
+++ /dev/null
@@ -1,355 +0,0 @@
-/**
- * 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 com.alibaba.rocketmq.filtersrv.processor;
-
-import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
-import com.alibaba.rocketmq.client.consumer.PullCallback;
-import com.alibaba.rocketmq.client.consumer.PullResult;
-import com.alibaba.rocketmq.common.MixAll;
-import com.alibaba.rocketmq.common.UtilAll;
-import com.alibaba.rocketmq.common.constant.LoggerName;
-import com.alibaba.rocketmq.common.filter.FilterContext;
-import com.alibaba.rocketmq.common.message.MessageDecoder;
-import com.alibaba.rocketmq.common.message.MessageExt;
-import com.alibaba.rocketmq.common.message.MessageQueue;
-import com.alibaba.rocketmq.common.protocol.RequestCode;
-import com.alibaba.rocketmq.common.protocol.ResponseCode;
-import com.alibaba.rocketmq.common.protocol.header.PullMessageRequestHeader;
-import com.alibaba.rocketmq.common.protocol.header.PullMessageResponseHeader;
-import com.alibaba.rocketmq.common.protocol.header.filtersrv.RegisterMessageFilterClassRequestHeader;
-import com.alibaba.rocketmq.common.sysflag.MessageSysFlag;
-import com.alibaba.rocketmq.filtersrv.FiltersrvController;
-import com.alibaba.rocketmq.filtersrv.filter.FilterClassInfo;
-import com.alibaba.rocketmq.remoting.common.RemotingHelper;
-import com.alibaba.rocketmq.remoting.exception.RemotingCommandException;
-import com.alibaba.rocketmq.remoting.netty.NettyRequestProcessor;
-import com.alibaba.rocketmq.remoting.protocol.RemotingCommand;
-import com.alibaba.rocketmq.store.CommitLog;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
-import io.netty.channel.ChannelHandlerContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.List;
-
-
-/**
- * @author shijia.wxr
- */
-public class DefaultRequestProcessor implements NettyRequestProcessor {
- private static final Logger log = LoggerFactory.getLogger(LoggerName.FILTERSRV_LOGGER_NAME);
-
- private final FiltersrvController filtersrvController;
-
-
- public DefaultRequestProcessor(FiltersrvController filtersrvController) {
- this.filtersrvController = filtersrvController;
- }
-
-
- @Override
- public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws Exception {
- if (log.isDebugEnabled()) {
- log.debug("receive request, {} {} {}",
- request.getCode(),
- RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
- request);
- }
-
- switch (request.getCode()) {
- case RequestCode.REGISTER_MESSAGE_FILTER_CLASS:
- return registerMessageFilterClass(ctx, request);
- case RequestCode.PULL_MESSAGE:
- return pullMessageForward(ctx, request);
- }
-
- return null;
- }
-
- @Override
- public boolean rejectRequest() {
- return false;
- }
-
- private RemotingCommand registerMessageFilterClass(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
- final RemotingCommand response = RemotingCommand.createResponseCommand(null);
- final RegisterMessageFilterClassRequestHeader requestHeader =
- (RegisterMessageFilterClassRequestHeader) request.decodeCommandCustomHeader(RegisterMessageFilterClassRequestHeader.class);
-
- try {
- boolean ok = this.filtersrvController.getFilterClassManager().registerFilterClass(requestHeader.getConsumerGroup(),
- requestHeader.getTopic(),
- requestHeader.getClassName(),
- requestHeader.getClassCRC(),
- request.getBody());
- if (!ok) {
- throw new Exception("registerFilterClass error");
- }
- } catch (Exception e) {
- response.setCode(ResponseCode.SYSTEM_ERROR);
- response.setRemark(RemotingHelper.exceptionSimpleDesc(e));
- return response;
- }
-
- response.setCode(ResponseCode.SUCCESS);
- response.setRemark(null);
- return response;
- }
-
- private RemotingCommand pullMessageForward(final ChannelHandlerContext ctx, final RemotingCommand request) throws Exception {
- final RemotingCommand response = RemotingCommand.createResponseCommand(PullMessageResponseHeader.class);
- final PullMessageResponseHeader responseHeader = (PullMessageResponseHeader) response.readCustomHeader();
- final PullMessageRequestHeader requestHeader =
- (PullMessageRequestHeader) request.decodeCommandCustomHeader(PullMessageRequestHeader.class);
-
- final FilterContext filterContext = new FilterContext();
- filterContext.setConsumerGroup(requestHeader.getConsumerGroup());
-
-
- response.setOpaque(request.getOpaque());
-
- DefaultMQPullConsumer pullConsumer = this.filtersrvController.getDefaultMQPullConsumer();
- final FilterClassInfo findFilterClass =
- this.filtersrvController.getFilterClassManager()
- .findFilterClass(requestHeader.getConsumerGroup(), requestHeader.getTopic());
- if (null == findFilterClass) {
- response.setCode(ResponseCode.SYSTEM_ERROR);
- response.setRemark("Find Filter class failed, not registered");
- return response;
- }
-
- if (null == findFilterClass.getMessageFilter()) {
- response.setCode(ResponseCode.SYSTEM_ERROR);
- response.setRemark("Find Filter class failed, registered but no class");
- return response;
- }
-
- responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
-
-
- MessageQueue mq = new MessageQueue();
- mq.setTopic(requestHeader.getTopic());
- mq.setQueueId(requestHeader.getQueueId());
- mq.setBrokerName(this.filtersrvController.getBrokerName());
- long offset = requestHeader.getQueueOffset();
- int maxNums = requestHeader.getMaxMsgNums();
-
- final PullCallback pullCallback = new PullCallback() {
-
- @Override
- public void onSuccess(PullResult pullResult) {
- responseHeader.setMaxOffset(pullResult.getMaxOffset());
- responseHeader.setMinOffset(pullResult.getMinOffset());
- responseHeader.setNextBeginOffset(pullResult.getNextBeginOffset());
- response.setRemark(null);
-
- switch (pullResult.getPullStatus()) {
- case FOUND:
- response.setCode(ResponseCode.SUCCESS);
-
- List<MessageExt> msgListOK = new ArrayList<MessageExt>();
- try {
- for (MessageExt msg : pullResult.getMsgFoundList()) {
- boolean match = findFilterClass.getMessageFilter().match(msg, filterContext);
- if (match) {
- msgListOK.add(msg);
- }
- }
-
-
- if (!msgListOK.isEmpty()) {
- returnResponse(requestHeader.getConsumerGroup(), requestHeader.getTopic(), ctx, response, msgListOK);
- return;
- } else {
- response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
- }
- } catch (Throwable e) {
- final String error =
- String.format("do Message Filter Exception, ConsumerGroup: %s Topic: %s ",
- requestHeader.getConsumerGroup(), requestHeader.getTopic());
- log.error(error, e);
-
- response.setCode(ResponseCode.SYSTEM_ERROR);
- response.setRemark(error + RemotingHelper.exceptionSimpleDesc(e));
- returnResponse(requestHeader.getConsumerGroup(), requestHeader.getTopic(), ctx, response, null);
- return;
- }
-
- break;
- case NO_MATCHED_MSG:
- response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
- break;
- case NO_NEW_MSG:
- response.setCode(ResponseCode.PULL_NOT_FOUND);
- break;
- case OFFSET_ILLEGAL:
- response.setCode(ResponseCode.PULL_OFFSET_MOVED);
- break;
- default:
- break;
- }
-
- returnResponse(requestHeader.getConsumerGroup(), requestHeader.getTopic(), ctx, response, null);
- }
-
-
- @Override
- public void onException(Throwable e) {
- response.setCode(ResponseCode.SYSTEM_ERROR);
- response.setRemark("Pull Callback Exception, " + RemotingHelper.exceptionSimpleDesc(e));
- returnResponse(requestHeader.getConsumerGroup(), requestHeader.getTopic(), ctx, response, null);
- return;
- }
- };
-
- pullConsumer.pullBlockIfNotFound(mq, null, offset, maxNums, pullCallback);
-
- return null;
- }
-
- private void returnResponse(final String group, final String topic, ChannelHandlerContext ctx, final RemotingCommand response,
- final List<MessageExt> msgList) {
- if (null != msgList) {
- ByteBuffer[] msgBufferList = new ByteBuffer[msgList.size()];
- int bodyTotalSize = 0;
- for (int i = 0; i < msgList.size(); i++) {
- try {
- msgBufferList[i] = messageToByteBuffer(msgList.get(i));
- bodyTotalSize += msgBufferList[i].capacity();
- } catch (Exception e) {
- log.error("messageToByteBuffer UnsupportedEncodingException", e);
- }
- }
-
- ByteBuffer body = ByteBuffer.allocate(bodyTotalSize);
- for (ByteBuffer bb : msgBufferList) {
- bb.flip();
- body.put(bb);
- }
-
- response.setBody(body.array());
-
-
- this.filtersrvController.getFilterServerStatsManager().incGroupGetNums(group, topic, msgList.size());
-
- this.filtersrvController.getFilterServerStatsManager().incGroupGetSize(group, topic, bodyTotalSize);
- }
-
- try {
- ctx.writeAndFlush(response).addListener(new ChannelFutureListener() {
- @Override
- public void operationComplete(ChannelFuture future) throws Exception {
- if (!future.isSuccess()) {
- log.error("FilterServer response to " + future.channel().remoteAddress() + " failed", future.cause());
- log.error(response.toString());
- }
- }
- });
- } catch (Throwable e) {
- log.error("FilterServer process request over, but response failed", e);
- log.error(response.toString());
- }
- }
-
- private ByteBuffer messageToByteBuffer(final MessageExt msg) throws IOException {
- int sysFlag = MessageSysFlag.clearCompressedFlag(msg.getSysFlag());
- if (msg.getBody() != null) {
- if (msg.getBody().length >= this.filtersrvController.getFiltersrvConfig().getCompressMsgBodyOverHowmuch()) {
- byte[] data = UtilAll.compress(msg.getBody(), this.filtersrvController.getFiltersrvConfig().getZipCompressLevel());
- if (data != null) {
- msg.setBody(data);
- sysFlag |= MessageSysFlag.COMPRESSED_FLAG;
- }
- }
- }
-
- final int bodyLength = msg.getBody() != null ? msg.getBody().length : 0;
- byte[] topicData = msg.getTopic().getBytes(MixAll.DEFAULT_CHARSET);
- final int topicLength = topicData.length;
- String properties = MessageDecoder.messageProperties2String(msg.getProperties());
- byte[] propertiesData = properties.getBytes(MixAll.DEFAULT_CHARSET);
- final int propertiesLength = propertiesData.length;
- final int msgLen = 4 // 1 TOTALSIZE
- + 4 // 2 MAGICCODE
- + 4 // 3 BODYCRC
- + 4 // 4 QUEUEID
- + 4 // 5 FLAG
- + 8 // 6 QUEUEOFFSET
- + 8 // 7 PHYSICALOFFSET
- + 4 // 8 SYSFLAG
- + 8 // 9 BORNTIMESTAMP
- + 8 // 10 BORNHOST
- + 8 // 11 STORETIMESTAMP
- + 8 // 12 STOREHOSTADDRESS
- + 4 // 13 RECONSUMETIMES
- + 8 // 14 Prepared Transaction Offset
- + 4 + bodyLength // 14 BODY
- + 1 + topicLength // 15 TOPIC
- + 2 + propertiesLength // 16 propertiesLength
- + 0;
-
- ByteBuffer msgStoreItemMemory = ByteBuffer.allocate(msgLen);
-
- final MessageExt msgInner = msg;
-
- // 1 TOTALSIZE
- msgStoreItemMemory.putInt(msgLen);
- // 2 MAGICCODE
- msgStoreItemMemory.putInt(CommitLog.MESSAGE_MAGIC_CODE);
- // 3 BODYCRC
- msgStoreItemMemory.putInt(UtilAll.crc32(msgInner.getBody()));
- // 4 QUEUEID
- msgStoreItemMemory.putInt(msgInner.getQueueId());
- // 5 FLAG
- msgStoreItemMemory.putInt(msgInner.getFlag());
- // 6 QUEUEOFFSET
- msgStoreItemMemory.putLong(msgInner.getQueueOffset());
- // 7 PHYSICALOFFSET
- msgStoreItemMemory.putLong(msgInner.getCommitLogOffset());
- // 8 SYSFLAG
- msgStoreItemMemory.putInt(sysFlag);
- // 9 BORNTIMESTAMP
- msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
- // 10 BORNHOST
- msgStoreItemMemory.put(msgInner.getBornHostBytes());
- // 11 STORETIMESTAMP
- msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
- // 12 STOREHOSTADDRESS
- msgStoreItemMemory.put(msgInner.getStoreHostBytes());
- // 13 RECONSUMETIMES
- msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());
- // 14 Prepared Transaction Offset
- msgStoreItemMemory.putLong(msgInner.getPreparedTransactionOffset());
- // 15 BODY
- msgStoreItemMemory.putInt(bodyLength);
- if (bodyLength > 0)
- msgStoreItemMemory.put(msgInner.getBody());
- // 16 TOPIC
- msgStoreItemMemory.put((byte) topicLength);
- msgStoreItemMemory.put(topicData);
- // 17 PROPERTIES
- msgStoreItemMemory.putShort((short) propertiesLength);
- if (propertiesLength > 0)
- msgStoreItemMemory.put(propertiesData);
-
- return msgStoreItemMemory;
- }
-}