vernedeng commented on code in PR #10727:
URL: https://github.com/apache/inlong/pull/10727#discussion_r1694558122
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/MqttSource.java:
##########
@@ -20,63 +20,57 @@
import org.apache.inlong.agent.conf.InstanceProfile;
import org.apache.inlong.agent.conf.TaskProfile;
import org.apache.inlong.agent.constant.CommonConstants;
+import org.apache.inlong.agent.constant.TaskConstants;
+import org.apache.inlong.agent.except.FileException;
import org.apache.inlong.agent.plugin.Message;
import org.apache.inlong.agent.plugin.file.Reader;
import org.apache.inlong.agent.plugin.sources.file.AbstractSource;
import org.apache.inlong.agent.plugin.sources.reader.MqttReader;
-import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
-import java.util.Objects;
public class MqttSource extends AbstractSource {
private static final Logger LOGGER =
LoggerFactory.getLogger(MqttSource.class);
- private static final String JOB_MQTTJOB_PARAM_PREFIX = "job.mqttJob.";
+ private MqttReader mqttReader;
- private static final String JOB_MQTTJOB_SERVERURI = "";
-
- private static final String JOB_MQTTJOB_CLIENTID = "";
-
- public static final String JOB_MQTTJOB_TOPICS = "job.mqttJob.topic";
+ private String topic;
public MqttSource() {
+ LOGGER.debug("MqttSource init!!!!!!!!!!!");
}
private List<Reader> splitSqlJob(String topics, String instanceId) {
if (StringUtils.isEmpty(topics)) {
return null;
}
- final List<Reader> result = new ArrayList<>();
+ final List<Reader> readers = new ArrayList<>();
String[] topicList = topics.split(CommonConstants.COMMA);
- if (Objects.nonNull(topicList)) {
- Arrays.stream(topicList).forEach(topic -> {
- MqttReader mqttReader = new MqttReader(topic);
- mqttReader.setReadSource(instanceId);
- result.add(mqttReader);
- });
- }
- return result;
+ Arrays.stream(topicList).forEach(topic -> {
+ MqttReader mqttReader = new MqttReader(topic);
+ mqttReader.setReadSource(instanceId);
+ readers.add(mqttReader);
+ });
+ return readers;
}
@Override
public List<Reader> split(TaskProfile conf) {
- String topics = conf.get(JOB_MQTTJOB_TOPICS, StringUtils.EMPTY);
- List<Reader> readerList = null;
- if (StringUtils.isNotEmpty(topics)) {
- }
- if (CollectionUtils.isNotEmpty(readerList)) {
- sourceMetric.sourceSuccessCount.incrementAndGet();
- } else {
- sourceMetric.sourceFailCount.incrementAndGet();
- }
+ LOGGER.info("splited!!!");
+ String topics = conf.get(TaskConstants.TASK_MQTT_TOPIC,
StringUtils.EMPTY);
+ List<Reader> readerList = splitSqlJob(topics, instanceId);
+ // if (CollectionUtils.isNotEmpty(readerList)) {
Review Comment:
please remove these code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
