opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-30 18:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #5089] Delegate plugin store configuration by broker

Browse Source
This commit is contained in:
SSpirits
2022-09-20 14:34:54 +08:00
committed by GitHub
parent 39df11994c
commit d5170e5ba3
4 changed files with 114 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+3 -3
View File
@@ -69,8 +69,8 @@ import org.apache.rocketmq.broker.offset.ConsumerOrderInfoManager;
import org.apache.rocketmq.broker.offset.LmqConsumerOffsetManager;
import org.apache.rocketmq.broker.out.BrokerOuterAPI;
import org.apache.rocketmq.broker.plugin.BrokerAttachedPlugin;
import org.apache.rocketmq.broker.plugin.MessageStoreFactory;
import org.apache.rocketmq.broker.plugin.MessageStorePluginContext;
import org.apache.rocketmq.store.plugin.MessageStoreFactory;
import org.apache.rocketmq.store.plugin.MessageStorePluginContext;
import org.apache.rocketmq.broker.processor.AckMessageProcessor;
import org.apache.rocketmq.broker.processor.AdminBrokerProcessor;
import org.apache.rocketmq.broker.processor.ChangeInvisibleTimeProcessor;
@@ -729,7 +729,7 @@ public class BrokerController {
}
this.brokerStats = new BrokerStats(defaultMessageStore);
//load plugin
MessageStorePluginContext context = new MessageStorePluginContext(this, messageStoreConfig, brokerStatsManager, messageArrivingListener);
MessageStorePluginContext context = new MessageStorePluginContext(messageStoreConfig, brokerStatsManager, messageArrivingListener, brokerConfig, configuration);
this.messageStore = MessageStoreFactory.build(context, defaultMessageStore);
this.messageStore.getDispatcherList().addFirst(new CommitLogDispatcherCalcBitMap(this.brokerConfig, this.consumerFilterManager));
if (this.brokerConfig.isEnableControllerMode()) {
broker/src/main/java/org/apache/rocketmq/broker/plugin/AbstractPluginMessageStore.java → store/src/main/java/org/apache/rocketmq/store/plugin/AbstractPluginMessageStore.java
+1 -1
View File
@@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.broker.plugin;
package org.apache.rocketmq.store.plugin;
import java.nio.ByteBuffer;
import java.util.HashMap;
broker/src/main/java/org/apache/rocketmq/broker/plugin/MessageStoreFactory.java → store/src/main/java/org/apache/rocketmq/store/plugin/MessageStoreFactory.java
+45 -46
View File
@@ -1,46 +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.broker.plugin;
import java.io.IOException;
import java.lang.reflect.Constructor;
import org.apache.rocketmq.store.MessageStore;
public final class MessageStoreFactory {
public final static MessageStore build(MessageStorePluginContext context,
MessageStore messageStore) throws IOException {
String plugin = context.getBrokerConfig().getMessageStorePlugIn();
if (plugin != null && plugin.trim().length() != 0) {
String[] pluginClasses = plugin.split(",");
for (int i = pluginClasses.length - 1; i >= 0; --i) {
String pluginClass = pluginClasses[i];
try {
@SuppressWarnings("unchecked")
Class<AbstractPluginMessageStore> clazz = (Class<AbstractPluginMessageStore>)Class.forName(pluginClass);
Constructor<AbstractPluginMessageStore> construct = clazz.getConstructor(MessageStorePluginContext.class, MessageStore.class);
AbstractPluginMessageStore pluginMessageStore = construct.newInstance(context, messageStore);
messageStore = pluginMessageStore;
}
catch (Throwable e) {
throw new RuntimeException("Initialize plugin's class: " + pluginClass + " not found!", e);
}
}
}
return messageStore;
}
}
/*
* 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.store.plugin;
import java.io.IOException;
import java.lang.reflect.Constructor;
import org.apache.rocketmq.store.MessageStore;
public final class MessageStoreFactory {
public static MessageStore build(MessageStorePluginContext context,
MessageStore messageStore) throws IOException {
String plugin = context.getBrokerConfig().getMessageStorePlugIn();
if (plugin != null && plugin.trim().length() != 0) {
String[] pluginClasses = plugin.split(",");
for (int i = pluginClasses.length - 1; i >= 0; --i) {
String pluginClass = pluginClasses[i];
try {
@SuppressWarnings("unchecked")
Class<AbstractPluginMessageStore> clazz = (Class<AbstractPluginMessageStore>) Class.forName(pluginClass);
Constructor<AbstractPluginMessageStore> construct = clazz.getConstructor(MessageStorePluginContext.class, MessageStore.class);
AbstractPluginMessageStore pluginMessageStore = construct.newInstance(context, messageStore);
messageStore = pluginMessageStore;
} catch (Throwable e) {
throw new RuntimeException("Initialize plugin's class: " + pluginClass + " not found!", e);
}
}
}
return messageStore;
}
}
broker/src/main/java/org/apache/rocketmq/broker/plugin/MessageStorePluginContext.java → store/src/main/java/org/apache/rocketmq/store/plugin/MessageStorePluginContext.java
+65 -61
View File
@@ -1,61 +1,65 @@
/*
* 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.plugin;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.store.MessageArrivingListener;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
public class MessageStorePluginContext {
private BrokerController controller;
private MessageStoreConfig messageStoreConfig;
private BrokerStatsManager brokerStatsManager;
private MessageArrivingListener messageArrivingListener;
public MessageStorePluginContext(BrokerController controller, MessageStoreConfig messageStoreConfig,
BrokerStatsManager brokerStatsManager, MessageArrivingListener messageArrivingListener) {
super();
this.messageStoreConfig = messageStoreConfig;
this.brokerStatsManager = brokerStatsManager;
this.messageArrivingListener = messageArrivingListener;
this.controller = controller;
}
public MessageStoreConfig getMessageStoreConfig() {
return messageStoreConfig;
}
public BrokerStatsManager getBrokerStatsManager() {
return brokerStatsManager;
}
public MessageArrivingListener getMessageArrivingListener() {
return messageArrivingListener;
}
public BrokerConfig getBrokerConfig() {
return controller.getBrokerConfig();
}
public BrokerController getController() {
return controller;
}
}
/*
* 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.store.plugin;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.Configuration;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.store.MessageArrivingListener;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
public class MessageStorePluginContext {
private MessageStoreConfig messageStoreConfig;
private BrokerStatsManager brokerStatsManager;
private MessageArrivingListener messageArrivingListener;
private BrokerConfig brokerConfig;
private final Configuration configuration;
public MessageStorePluginContext(MessageStoreConfig messageStoreConfig,
BrokerStatsManager brokerStatsManager, MessageArrivingListener messageArrivingListener,
BrokerConfig brokerConfig, Configuration configuration) {
super();
this.messageStoreConfig = messageStoreConfig;
this.brokerStatsManager = brokerStatsManager;
this.messageArrivingListener = messageArrivingListener;
this.brokerConfig = brokerConfig;
this.configuration = configuration;
}
public MessageStoreConfig getMessageStoreConfig() {
return messageStoreConfig;
}
public BrokerStatsManager getBrokerStatsManager() {
return brokerStatsManager;
}
public MessageArrivingListener getMessageArrivingListener() {
return messageArrivingListener;
}
public BrokerConfig getBrokerConfig() {
return brokerConfig;
}
public void registerConfiguration(Object config) {
MixAll.properties2Object(configuration.getAllConfigs(), config);
configuration.registerConfig(config);
}
}