删除无用的api

hsweb-message/hsweb-message-api/src/main/java/org/hswebframework/web/message/MessagePublish.java

@@ -18,19 +18,11 @@
 
 package org.hswebframework.web.message;
 
-import java.util.function.Consumer;
-
 /**
  * @author zhouhao
  */
 public interface MessagePublish {
-    MessagePublish from(MessageSubject subject);
-
     MessagePublish to(MessageSubject subject);
 
-    MessagePublish deleteOnTimeout(long timeOutSecond);
-
-    <T> T send();
-
-    <T> void send(Consumer<T> responseConsumer);
+    void send();
 }

hsweb-message/hsweb-message-api/src/main/java/org/hswebframework/web/message/MessageSubscribe.java

@@ -29,4 +29,6 @@ public interface MessageSubscribe<M extends Message> {
     MessageSubscribe<M> iam(MessageSubject iam);
 
     MessageSubscribe<M> onMessage(Consumer<M> consumer);
+
+    void cancel();
 }

hsweb-message/hsweb-message-api/src/test/java/org/hswebframework/web/message/builder/SimpleMessageBuilder.java

@@ -23,12 +23,14 @@ import org.hswebframework.web.message.support.ObjectMessage;
 import org.hswebframework.web.message.support.ServiceInvokerMessage;
 import org.hswebframework.web.message.support.TextMessage;
 
+import java.io.Serializable;
+
 /**
  * TODO 完成注释
  *
  * @author zhouhao
  */
-public class SimpleMessageBuilder implements MessageBuilder {
+public class SimpleMessageBuilder implements MessageBuilder ,Serializable {
     @Override
     public TextMessage text(String msg) {
         return new TextMessage() {

hsweb-message/hsweb-message-api/src/test/java/org/hswebframework/web/message/builder/SimpleMessageSubjectBuilder.java

@@ -23,6 +23,7 @@ import org.hswebframework.web.message.support.MultipleUserMessageSubject;
 import org.hswebframework.web.message.support.TopicMessageSubject;
 import org.hswebframework.web.message.support.UserMessageSubject;
 
+import java.io.Serializable;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Set;
@@ -32,7 +33,7 @@ import java.util.Set;
  *
  * @author zhouhao
  */
-public class SimpleMessageSubjectBuilder implements MessageSubjectBuilder {
+public class SimpleMessageSubjectBuilder implements MessageSubjectBuilder,Serializable {
     @Override
     public UserMessageSubject user(String userId) {
         return () -> userId;

hsweb-message/hsweb-message-api/src/test/java/org/hswebframework/web/message/builder/StaticMessageBuilder.java

@@ -35,15 +35,15 @@ public class StaticMessageBuilder {
         return messageBuilder.text(msg);
     }
 
-    public <T> ObjectMessage object(T msg) {
+    public static <T> ObjectMessage object(T msg) {
         return messageBuilder.object(msg);
     }
 
-    public DataMessage data(byte[] msg) {
+    public static DataMessage data(byte[] msg) {
         return messageBuilder.data(msg);
     }
 
-    public ServiceInvokerMessage service(String serviceName) {
+    public static ServiceInvokerMessage service(String serviceName) {
         return messageBuilder.service(serviceName);
     }
 }

hsweb-message/hsweb-message-api/src/test/java/org/hswebframework/web/message/builder/StaticMessageSubjectBuilder.java

@@ -43,15 +43,15 @@ public class StaticMessageSubjectBuilder {
         return messageSubjectBuilder.users(userIds);
     }
 
-    public MultipleUserMessageSubject users(Set<String> userIds) {
+    public static MultipleUserMessageSubject users(Set<String> userIds) {
         return messageSubjectBuilder.users(userIds);
     }
 
-    public MessageSubject system() {
+    public static MessageSubject system() {
         return messageSubjectBuilder.system();
     }
 
-    public TopicMessageSubject topic(String topic) {
+    public static TopicMessageSubject topic(String topic) {
         return messageSubjectBuilder.topic(topic);
     }
 }

hsweb-message/hsweb-message-api/src/main/java/org/hswebframework/web/message/support/TopicMessageSubject.java

@@ -18,12 +18,13 @@
 
 package org.hswebframework.web.message.support;
 
+import org.hswebframework.web.message.MessageSubject;
+
 /**
  * TODO 完成注释
  *
  * @author zhouhao
  */
-public interface TopicMessageSubject {
+public interface TopicMessageSubject extends MessageSubject {
     String getTopic();
-
 }

hsweb-message/hsweb-message-api/src/test/java/org/hswebframework/web/message/MessagePublishHanlder.java

@@ -23,7 +23,7 @@ package org.hswebframework.web.message;
  *
  * @author zhouhao
  */
-public interface MessagePublishHanlder {
+public interface MessagePublishHandler {
     boolean isSupport(Message message);
 
     MessagePublish handle(Message message);

hsweb-message/hsweb-message-api/src/test/java/org/hswebframework/web/message/SimpleMessagePublish.java

@@ -18,39 +18,18 @@
 
 package org.hswebframework.web.message;
 
-import java.util.function.Consumer;
-
 /**
  * TODO 完成注释
  *
  * @author zhouhao
  */
 public abstract class SimpleMessagePublish implements MessagePublish {
-    MessageSubject from;
 
     MessageSubject to;
 
-    @Override
-    public MessagePublish from(MessageSubject subject) {
-        this.from = subject;
-        return this;
-    }
-
     @Override
     public MessagePublish to(MessageSubject subject) {
         this.to = subject;
-
         return this;
     }
-
-    @Override
-    public MessagePublish deleteOnTimeout(long timeOutSecond) {
-        //not support now
-        return this;
-    }
-
-    @Override
-    public <T> void send(Consumer<T> responseConsumer) {
-        responseConsumer.accept(send());
-    }
 }

hsweb-message/hsweb-message-api/src/test/java/org/hswebframework/web/message/SimpleMessageSubscribe.java

@@ -61,6 +61,11 @@ public class SimpleMessageSubscribe<T extends Message> implements MessageSubscri
         return this;
     }
 
+    @Override
+    public void cancel() {
+        stop = true;
+    }
+
     public void startConsumer() {
         if (started) return;
         new Thread(() -> {

hsweb-message/hsweb-message-api/src/test/java/org/hswebframework/web/message/SimpleMessageTests.java

@@ -18,7 +18,6 @@
 
 package org.hswebframework.web.message;
 
-import org.hswebframework.web.message.support.TextMessage;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -41,7 +40,6 @@ public class SimpleMessageTests {
 
         messager.publish(text("hello2"))
                 .to(user("test"))
-                .from(user("admin"))
                 .send();
 
         Thread.sleep(1000);

hsweb-message/hsweb-message-api/src/test/java/org/hswebframework/web/message/SimpleMessager.java

@@ -39,11 +39,11 @@ public class SimpleMessager implements Messager {
         return queueStorage.computeIfAbsent(key, k -> new LinkedBlockingQueue<>());
     }
 
-    List<MessagePublishHanlder> publishHanlders = new ArrayList<>();
+    List<MessagePublishHandler> publishHanlders = new ArrayList<>();
 
     public SimpleMessager() {
         //just support TextMessage
-        publishHanlders.add(new MessagePublishHanlder() {
+        publishHanlders.add(new MessagePublishHandler() {
             @Override
             public boolean isSupport(Message message) {
                 return message instanceof TextMessage;
@@ -53,11 +53,9 @@ public class SimpleMessager implements Messager {
             public MessagePublish handle(Message message) {
                 return new SimpleMessagePublish() {
                     @Override
-                    public <T> T send() {
-                        getQueue(buildKey(to)).add(message);
-                        return null;
+                    public void send() {
+                        getQueue(buildKey(to)).offer(message);
                     }
-
                 };
             }
         });

hsweb-message/hsweb-message-redis/pom.xml

@@ -13,6 +13,10 @@
 
 
     <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.hswebframework.web</groupId>
             <artifactId>hsweb-message-api</artifactId>

hsweb-message/hsweb-message-redis/src/main/java/org/hswebframework/web/message/redis/RedissionMessageSubscribe.java

@@ -18,10 +18,10 @@ import java.util.function.Consumer;
 public class RedissionMessageSubscribe<M extends Message> implements MessageSubscribe<M> {
     private MessageSubject iam;
     private RedissonClient redisson;
-
-    private boolean running = false;
-
-    private List<Consumer<M>> consumers = new ArrayList<>();
+    private boolean           running    = false;
+    private int               listenerId = 0;
+    private List<Consumer<M>> consumers  = new ArrayList<>();
+    private RTopic<M> topic;
 
     public RedissionMessageSubscribe(MessageSubject iam, RedissonClient redisson) {
         this.iam = iam;
@@ -47,6 +47,16 @@ public class RedissionMessageSubscribe<M extends Message> implements MessageSubs
         return this;
     }
 
+    @Override
+    public void cancel() {
+        running = false;
+        if (listenerId != 0 && topic != null) {
+            topic.removeListener(listenerId);
+            topic = null;
+        }
+        consumers.clear();
+    }
+
     private static SerializationCodec codec = new SerializationCodec();
 
     private void doRun() {
@@ -59,7 +69,11 @@ public class RedissionMessageSubscribe<M extends Message> implements MessageSubs
                     try {
                         countDownLatch.trySetCount(1);
                         countDownLatch.await();
-                        consumers.forEach(cons -> cons.accept(queue.peek()));
+                        consumers.forEach(cons -> {
+                            M message = queue.poll();
+                            if (null != message)
+                                cons.accept(message);
+                        });
                     } catch (InterruptedException e) {
                         try {
                             Thread.sleep(1000);
@@ -74,8 +88,8 @@ public class RedissionMessageSubscribe<M extends Message> implements MessageSubs
             return;
         }
         if (iam instanceof TopicMessageSubject) {
-            RTopic<M> topic = redisson.getTopic("topic_" + ((TopicMessageSubject) iam).getTopic(), codec);
-            topic.addListener((channel, msg) -> consumers.forEach(cons -> cons.accept(msg)));
+            topic = redisson.getTopic("topic_" + ((TopicMessageSubject) iam).getTopic(), codec);
+            listenerId = topic.addListener((channel, msg) -> consumers.forEach(cons -> cons.accept(msg)));
         }
         running = true;
     }

hsweb-message/hsweb-message-redis/src/main/java/org/hswebframework/web/message/redis/RedissonMessagePublish.java

@@ -20,7 +20,6 @@ import java.util.function.Consumer;
  * @author zhouhao
  */
 public class RedissonMessagePublish implements MessagePublish {
-    private MessageSubject from;
     private MessageSubject to;
     private RedissonClient redissonClient;
     private Message        message;
@@ -30,23 +29,12 @@ public class RedissonMessagePublish implements MessagePublish {
         this.message = message;
     }
 
-    @Override
-    public MessagePublish from(MessageSubject subject) {
-        this.from = subject;
-        return this;
-    }
-
     @Override
     public MessagePublish to(MessageSubject subject) {
         this.to = subject;
         return this;
     }
 
-    @Override
-    public MessagePublish deleteOnTimeout(long timeOutSecond) {
-        return null;
-    }
-
     private boolean useQueue() {
         return to instanceof UserMessageSubject || to instanceof MultipleUserMessageSubject;
     }
@@ -61,7 +49,10 @@ public class RedissonMessagePublish implements MessagePublish {
     };
 
     @Override
-    public <T> T send() {
+    public void send() {
+        if (redissonClient.isShutdown() || redissonClient.isShuttingDown()) {
+            return;
+        }
         if (to instanceof UserMessageSubject) {
             queueConsumer.accept(((UserMessageSubject) to).getUserId());
         }
@@ -72,12 +63,7 @@ public class RedissonMessagePublish implements MessagePublish {
             RTopic<Message> topic = redissonClient.getTopic("topic_" + ((TopicMessageSubject) to).getTopic(), codec);
             topic.publish(message);
         }
-        throw new UnsupportedOperationException();
     }
 
-    @Override
-    public <T> void send(Consumer<T> responseConsumer) {
-        responseConsumer.accept(send());
-    }
 
 }

hsweb-message/hsweb-message-redis/src/main/java/org/hswebframework/web/message/redis/starter/RedissonMessagerAutoConfiguration.java

@@ -0,0 +1,24 @@
+package org.hswebframework.web.message.redis.starter;
+
+import org.hswebframework.web.message.Messager;
+import org.hswebframework.web.message.redis.RedissonMessager;
+import org.redisson.api.RedissonClient;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+@Configuration
+@ConditionalOnBean(RedissonClient.class)
+@ConditionalOnMissingBean(Messager.class)
+public class RedissonMessagerAutoConfiguration {
+    @Bean
+    public Messager messager(RedissonClient client) {
+        return new RedissonMessager(client);
+    }
+}

hsweb-message/hsweb-message-redis/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,3 @@
+# Auto Configure
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.hswebframework.web.message.redis.starter.RedissonMessagerAutoConfiguration