Merge remote-tracking branch 'origin/master'

1 files changed, 9 insertions, 111 deletions

diff --git a/src/main/java/com/juick/server/ServerManager.java b/src/main/java/com/juick/server/ServerManager.java
index 7a95ec43..46faf9eb 100644
--- a/src/main/java/com/juick/server/ServerManager.java
+++ b/src/main/java/com/juick/server/ServerManager.java
@@ -16,7 +16,6 @@
  */
 package com.juick.server;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.juick.Message;
 import com.juick.User;
@@ -24,25 +23,18 @@ import com.juick.model.AnonymousUser;
 import com.juick.service.MessagesService;
 import com.juick.service.SubscriptionService;
 import com.juick.service.UserService;
-import com.juick.service.component.LikeEvent;
-import com.juick.service.component.MessageEvent;
-import com.juick.service.component.MessageReadEvent;
-import com.juick.service.component.NotificationListener;
-import com.juick.service.component.PingEvent;
-import com.juick.service.component.SubscribeEvent;
-import com.juick.service.component.TopEvent;
+import com.juick.service.component.*;
 import com.juick.util.MessageUtils;
+import org.apache.commons.collections4.ListUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
-import org.springframework.web.socket.TextMessage;
 
 import javax.annotation.PostConstruct;
 import javax.inject.Inject;
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -62,8 +54,6 @@ public class ServerManager implements NotificationListener {
     @Inject
     private MessagesService messagesService;
     @Inject
-    private WebsocketManager wsHandler;
-    @Inject
     private SubscriptionService subscriptionService;
     @Inject
     private UserService userService;
@@ -81,96 +71,15 @@ public class ServerManager implements NotificationListener {
 
 
     private void onJuickPM(final User to, final com.juick.Message jmsg) {
-        try {
-            String json = jsonMapper.writeValueAsString(jmsg);
-            synchronized (wsHandler.getClients()) {
-                wsHandler.getClients().stream().filter(c ->
-                        (!c.legacy && c.visitor.getUid() == to.getUid()) || c.visitor.equals(serviceUser))
-                        .forEach(c -> {
-                            try {
-                                logger.debug("sending pm to {}", c.visitor.getUid());
-                                c.sendMessage(new TextMessage(json));
-                            } catch (IOException e) {
-                                logger.warn("ws error", e);
-                            }
-                        });
-            }
-        } catch (JsonProcessingException e) {
-            logger.warn("Invalid JSON", e);
-        }
-        messageEvent(jmsg, Collections.singletonList(to));
+        messageEvent(jmsg, Arrays.asList(to, jmsg.getUser()));
     }
 
     private void onJuickMessagePost(final com.juick.Message jmsg, List<User> subscribedUsers) {
-        try {
-            String json = jsonMapper.writeValueAsString(jmsg);
-            List<Integer> uids = subscribedUsers
-                    .stream().map(User::getUid).collect(Collectors.toList());
-            synchronized (wsHandler.getClients()) {
-                wsHandler.getClients().stream().filter(c ->
-                        (!c.legacy && c.visitor.isAnonymous()) // anonymous users
-                                || c.visitor.equals(serviceUser) // services
-                                || (!c.legacy && uids.contains(c.visitor.getUid()))) // subscriptions
-                        .forEach(c -> {
-                            try {
-                                logger.debug("sending message to {}", c.visitor.getUid());
-                                c.sendMessage(new TextMessage(json));
-                            } catch (IOException e) {
-                                logger.warn("ws error", e);
-                            }
-                        });
-                wsHandler.getClients().stream().filter(c ->
-                        c.legacy && c.allMessages) // legacy all posts
-                        .forEach(c -> {
-                            try {
-                                logger.debug("sending message to legacy client {}", c.visitor.getUid());
-                                c.sendMessage(new TextMessage(json));
-                            } catch (IOException e) {
-                                logger.warn("ws error", e);
-                            }
-                        });
-            }
-        } catch (JsonProcessingException e) {
-            logger.warn("Invalid JSON", e);
-        }
         messageEvent(jmsg, subscribedUsers);
         messageEvent(jmsg, Collections.singletonList(AnonymousUser.INSTANCE));
     }
 
     private void onJuickMessageReply(final com.juick.Message jmsg, final List<User> subscribedUsers) {
-        try {
-
-            String json = jsonMapper.writeValueAsString(jmsg);
-            List<Integer> threadUsers =
-                    subscribedUsers
-                            .stream().map(User::getUid).collect(Collectors.toList());
-            synchronized (wsHandler.getClients()) {
-                wsHandler.getClients().stream().filter(c ->
-                        (!c.legacy && c.visitor.isAnonymous()) // anonymous users
-                                || c.visitor.equals(serviceUser) // services
-                                || (!c.legacy && threadUsers.contains(c.visitor.getUid()))) // subscriptions
-                        .forEach(c -> {
-                            try {
-                                logger.debug("sending reply to {}", c.visitor.getUid());
-                                c.sendMessage(new TextMessage(json));
-                            } catch (IOException e) {
-                                logger.warn("ws error", e);
-                            }
-                        });
-                wsHandler.getClients().stream().filter(c ->
-                        (c.legacy && c.allReplies) || (c.legacy && c.MID == jmsg.getMid())) // legacy replies
-                        .forEach(c -> {
-                            try {
-                                logger.debug("sending reply to legacy client {}", c.visitor.getUid());
-                                c.sendMessage(new TextMessage(json));
-                            } catch (IOException e) {
-                                logger.warn("ws error", e);
-                            }
-                        });
-            }
-        } catch (JsonProcessingException e) {
-            logger.warn("Invalid JSON", e);
-        }
         messageEvent(jmsg, subscribedUsers);
         messageEvent(jmsg, Collections.singletonList(AnonymousUser.INSTANCE));
     }
@@ -178,7 +87,7 @@ public class ServerManager implements NotificationListener {
     @Override
     public void processMessageEvent(MessageEvent event) {
         com.juick.Message jmsg = event.getMessage();
-        List<User> subscribedUsers = event.getUsers();
+        List<User> subscribedUsers = ListUtils.union(event.getUsers(), Collections.singletonList(jmsg.getUser()));
         if (jmsg.isService()) {
             readEvent(jmsg, Collections.singletonList(serviceUser));
             return;
@@ -186,16 +95,14 @@ public class ServerManager implements NotificationListener {
         if (MessageUtils.isPM(jmsg)) {
             onJuickPM(jmsg.getTo(), jmsg);
         } else if (!MessageUtils.isReply(jmsg)) {
-            // to get full message with attachment, etc.
-            onJuickMessagePost(messagesService.getMessage(jmsg.getMid()).orElseThrow(IllegalStateException::new), subscribedUsers);
+            onJuickMessagePost(jmsg, subscribedUsers);
         } else {
             // to get quote and attachment
             Message op = messagesService.getMessage(jmsg.getMid()).orElseThrow(IllegalStateException::new);
-            com.juick.Message reply = messagesService.getReply(jmsg.getMid(), jmsg.getRid());
-            subscriptionService.getUsersSubscribedToComments(op, reply, true).stream()
-                    .filter(u -> userService.isReplyToBL(u, reply))
-                    .forEach(b -> messagesService.setLastReadComment(b, reply.getMid(), reply.getRid()));
-            onJuickMessageReply(reply, subscribedUsers);
+            subscriptionService.getUsersSubscribedToComments(op, jmsg, true).stream()
+                    .filter(u -> userService.isReplyToBL(u, jmsg))
+                    .forEach(b -> messagesService.setLastReadComment(b, jmsg.getMid(), jmsg.getRid()));
+            onJuickMessageReply(jmsg, subscribedUsers);
         }
         messageEvent(jmsg, Collections.singletonList(serviceUser));
     }
@@ -226,15 +133,6 @@ public class ServerManager implements NotificationListener {
         serviceMessage.setUser(user);
         serviceMessage.setMid(source.getMid());
         serviceMessage.setUnread(false);
-        wsHandler.getClients().stream().filter(c ->
-                (!c.legacy && c.visitor == user) || c.visitor.equals(serviceUser)
-        ).forEach(u -> {
-            try {
-                u.sendMessage(new TextMessage(jsonMapper.writeValueAsString(serviceMessage)));
-            } catch (IOException e) {
-                logger.error("JSON error", e);
-            }
-        });
         readEvent(serviceMessage, Collections.singletonList(serviceUser));
     }