2 files changed, 29 insertions, 12 deletions

diff --git a/juick-server/src/main/assets/scripts.js b/juick-server/src/main/assets/scripts.js
index 95cbe22f..36678f97 100644
--- a/juick-server/src/main/assets/scripts.js
+++ b/juick-server/src/main/assets/scripts.js
@@ -143,7 +143,7 @@ function initES() {
             pageTitle = document.title;
         }
     };
-    es.onmessage = function(msg) {
+    es.addEventListener("msg",  msg => {
         try {
             var jsonMsg = JSON.parse(msg.data);
             console.log('data: ' + msg.data);
@@ -154,7 +154,7 @@ function initES() {
         } catch (err) {
             console.log(err);
         }
-    };
+    });
 }
 
 function wsIncomingReply(msg) {
diff --git a/juick-server/src/main/java/com/juick/server/ServerManager.java b/juick-server/src/main/java/com/juick/server/ServerManager.java
index 6aba7212..be7a5dfa 100644
--- a/juick-server/src/main/java/com/juick/server/ServerManager.java
+++ b/juick-server/src/main/java/com/juick/server/ServerManager.java
@@ -43,6 +43,7 @@ import javax.annotation.PostConstruct;
 import javax.inject.Inject;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -96,7 +97,7 @@ public class ServerManager implements NotificationListener {
         } catch (JsonProcessingException e) {
             logger.warn("Invalid JSON", e);
         }
-        sendSseEvent(jmsg, Collections.singletonList(to));
+        messageEvent(jmsg, Arrays.asList(to, serviceUser));
     }
 
     private void onJuickMessagePost(final com.juick.Message jmsg, List<User> subscribedUsers) {
@@ -131,8 +132,8 @@ public class ServerManager implements NotificationListener {
         } catch (JsonProcessingException e) {
             logger.warn("Invalid JSON", e);
         }
-        sendSseEvent(jmsg, subscribedUsers);
-        sendSseEvent(jmsg, Collections.singletonList(AnonymousUser.INSTANCE));
+        messageEvent(jmsg, subscribedUsers);
+        messageEvent(jmsg, Collections.singletonList(AnonymousUser.INSTANCE));
     }
 
     private void onJuickMessageReply(final com.juick.Message jmsg, final List<User> subscribedUsers) {
@@ -170,8 +171,8 @@ public class ServerManager implements NotificationListener {
         } catch (JsonProcessingException e) {
             logger.warn("Invalid JSON", e);
         }
-        sendSseEvent(jmsg, subscribedUsers);
-        sendSseEvent(jmsg, Collections.singletonList(AnonymousUser.INSTANCE));
+        messageEvent(jmsg, subscribedUsers);
+        messageEvent(jmsg, Collections.singletonList(AnonymousUser.INSTANCE));
     }
 
     @Override
@@ -179,7 +180,7 @@ public class ServerManager implements NotificationListener {
         com.juick.Message jmsg = event.getMessage();
         List<User> subscribedUsers = event.getUsers();
         if (jmsg.isService()) {
-            sendSseEvent(jmsg, Collections.singletonList(serviceUser));
+            readEvent(jmsg, Collections.singletonList(serviceUser));
             return;
         }
         if (MessageUtils.isPM(jmsg)) {
@@ -196,7 +197,7 @@ public class ServerManager implements NotificationListener {
                     .forEach(b -> messagesService.setLastReadComment(b, reply.getMid(), reply.getRid()));
             onJuickMessageReply(reply, subscribedUsers);
         }
-        sendSseEvent(jmsg, Collections.singletonList(serviceUser));
+        messageEvent(jmsg, Collections.singletonList(serviceUser));
     }
 
     @Override
@@ -234,19 +235,35 @@ public class ServerManager implements NotificationListener {
                 logger.error("JSON error", e);
             }
         });
-        sendSseEvent(serviceMessage, Collections.singletonList(serviceUser));
+        readEvent(serviceMessage, Collections.singletonList(serviceUser));
     }
 
     @Override
     public void processTopEvent(TopEvent topEvent) {
+        User topUser = topEvent.getMessage().getUser();
+        topEvent(topEvent.getMessage(), Arrays.asList(topUser, serviceUser));
+    }
+
+    public void topEvent(Message msg, List<User> subscribers){
+        sendSseEvent(msg, "top", subscribers);
+    }
+
+    public void readEvent(Message msg, List<User> subscribers){
+        sendSseEvent(msg, "read", subscribers);
+    }
 
+    public void messageEvent(Message msg, List<User> subscribers){
+        sendSseEvent(msg, "msg", subscribers);
     }
 
-    public void sendSseEvent(Message msg, List<User> subscribers) {
+    private void sendSseEvent(Message msg, String name, List<User> subscribers) {
         List<EventSession> deadEmitters = new ArrayList<>();
         this.sessions.stream().filter(s -> subscribers.contains(s.user)).forEach(session -> {
             try {
-                session.getEmitter().send(msg);
+                SseEmitter.SseEventBuilder builder = SseEmitter.event()
+                        .name(name)
+                        .data(msg);
+                session.getEmitter().send(builder);
             } catch (Exception e) {
                 deadEmitters.add(session);
             }