删除MessageStreams

src/META-INF/application-template.xml

@@ -72,9 +72,6 @@
             <producer>
                 <property name="xxxxxx" value="XXXXXXXX"/>
             </producer>
-            <streams>
-                <property name="xxxxxx" value="XXXXXXXX"/>
-            </streams>
         </mq>
         -->
         <!--

src/org/redkale/mq/MessageAgent.java

@@ -73,9 +73,6 @@ public abstract class MessageAgent {
     //创建指定topic的生产处理器
     public abstract MessageProducer createProducer();
 
-    //创建指定topic的流处理器
-    public abstract MessageStreams createStreams(String topic, Function<MessageRecord, MessageRecord> processor);
-
     //格式: sncp:req:user
     protected static String generateSncpReqTopic(NodeServer ns, Service service) {
         String resname = Sncp.getResourceName(service);

src/org/redkale/mq/MessageStreams.java

@@ -1,49 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkale.mq;
-
-import java.util.Objects;
-import java.util.function.Function;
-import java.util.logging.Logger;
-
-/**
- *
- * <p>
- * 详情见: https://redkale.org
- *
- * @author zhangjx
- */
-public abstract class MessageStreams extends Thread {
-
-    protected final String topic;
-
-    protected final Function<MessageRecord, MessageRecord> processor;
-
-    protected final Logger logger = Logger.getLogger(this.getClass().getSimpleName());
-
-    protected volatile boolean closed;
-
-    protected MessageStreams(String topic, Function<MessageRecord, MessageRecord> processor) {
-        Objects.requireNonNull(topic);
-        Objects.requireNonNull(processor);
-        this.topic = topic;
-        this.processor = processor;
-    }
-
-    public Function<MessageRecord, MessageRecord> getProcessor() {
-        return processor;
-    }
-
-    public String getTopic() {
-        return topic;
-    }
-
-    public boolean isClosed() {
-        return closed;
-    }
-
-    public abstract void close();
-}