Implementation for [RIP-83 Lite Topic: A New Message Model]

Change-Id: I102d2bf7b5d573f41416333135074cde1a9f1fec

Author: zhouli11

broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSharding.java

@@ -0,0 +1,23 @@
+/*
+ * 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.broker.lite;
+
+public interface LiteSharding {
+
+    String shardingByLmqName(String parentTopic, String lmqName);
+}

broker/src/main/java/org/apache/rocketmq/broker/lite/LiteShardingImpl.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.broker.lite;
+
+import com.google.common.hash.Hashing;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.topic.TopicRouteInfoManager;
+import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
+import org.apache.rocketmq.common.lite.LiteUtil;
+import org.apache.rocketmq.common.message.MessageQueue;
+
+import java.util.List;
+
+public class LiteShardingImpl implements LiteSharding {
+
+    private final BrokerController brokerController;
+    private final TopicRouteInfoManager topicRouteInfoManager;
+
+    public LiteShardingImpl(BrokerController brokerController, TopicRouteInfoManager topicRouteInfoManager) {
+        this.brokerController = brokerController;
+        this.topicRouteInfoManager = topicRouteInfoManager;
+    }
+
+    @Override
+    public String shardingByLmqName(String parentTopic, String lmqName) {
+        TopicPublishInfo topicPublishInfo = topicRouteInfoManager.tryToFindTopicPublishInfo(parentTopic);
+        if (topicPublishInfo == null) {
+            // if topic not exist, return current broker
+            return brokerController.getBrokerConfig().getBrokerName();
+        }
+        List<MessageQueue> writeQueues = topicPublishInfo.getMessageQueueList();
+        if (CollectionUtils.isEmpty(writeQueues)) {
+            return brokerController.getBrokerConfig().getBrokerName();
+        }
+        String liteTopic = LiteUtil.getLiteTopic(lmqName);
+        if (StringUtils.isEmpty(liteTopic)) {
+            return brokerController.getBrokerConfig().getBrokerName();
+        }
+        int bucket = Hashing.consistentHash(liteTopic.hashCode(), writeQueues.size());
+        MessageQueue targetQueue = writeQueues.get(bucket);
+        return targetQueue.getBrokerName();
+    }
+}

broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistry.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.broker.lite;
+
+import io.netty.channel.Channel;
+
+import java.util.List;
+import java.util.Set;
+import org.apache.rocketmq.common.entity.ClientGroup;
+import org.apache.rocketmq.common.lite.LiteSubscription;
+
+public interface LiteSubscriptionRegistry {
+
+    void updateClientChannel(String clientId, Channel channel);
+
+    LiteSubscription getLiteSubscription(String clientId);
+
+    int getActiveSubscriptionNum();
+
+    void addPartialSubscription(String clientId, String group, String topic, Set<String> lmqNameSet, boolean exclusive);
+
+    void removePartialSubscription(String clientId, String group, String topic, Set<String> lmqNameSet);
+
+    void addCompleteSubscription(String clientId, String group, String topic, Set<String> newLmqNameSet, long version);
+
+    void removeCompleteSubscription(String clientId);
+
+    void addListener(LiteCtlListener listener);
+
+    Set<ClientGroup> getSubscriber(String lmqName);
+
+    List<String> getAllClientIdByGroup(String group);
+
+    void cleanSubscription(String lmqName, boolean notifyClient);
+
+    void start();
+
+    void shutdown();
+}