#24 simple sns: alarm with sse

simple_sns/front-end/src/layouts/alarm/index.js

@@ -47,6 +47,7 @@ import Slide from '@mui/material/Slide';
 // Data
 import axios from 'axios';
 
+
 const Transition = React.forwardRef(function Transition(
   props: TransitionProps & {
     children: React.ReactElement<any, any>,
@@ -61,8 +62,10 @@ function Alarm() {
   const [render, setRender] = useState(false);
   const [alarms, setAlarms] = useState([]);
   const [totalPage, setTotalPage] = useState(0);
+  const [alarmEvent, setAlarmEvent] = useState(undefined);
 
   const navigate = useNavigate();
+  let eventSource = undefined;
 
   const changePage = (pageNum) => {
     console.log('change pages');
@@ -75,7 +78,7 @@ function Alarm() {
   const handleGetAlarm = (pageNum, event) => {
     console.log('handleGetAlarm');
     axios({
-      url: '/api/v1/users/alarm?size=5&sort=id&page=' + pageNum,
+      url: '/api/v1/users/alarm?size=5&sort=id,desc&page=' + pageNum,
       method: 'GET',
       headers: {
         Authorization: 'Bearer ' + localStorage.getItem('token'),
@@ -95,6 +98,28 @@ function Alarm() {
 
   useEffect(() => {
     handleGetAlarm();
+
+    eventSource = new EventSource("http://localhost:8080/api/v1/users/alarm/subscribe?token=" + localStorage.getItem('token'));
+
+    setAlarmEvent(eventSource);
+
+    eventSource.addEventListener("open", function (event) {
+      console.log("connection opened");
+    });
+
+    eventSource.addEventListener("alarm", function (event) {
+       console.log(event.data);
+       handleGetAlarm();
+    });
+
+    eventSource.addEventListener("error", function (event) {
+      console.log(event.target.readyState);
+      if (event.target.readyState === EventSource.CLOSED) {
+        console.log("eventsource closed (" + event.target.readyState + ")");
+      }
+      eventSource.close();
+    });
+
   }, []);
 
   return (

simple_sns/src/main/java/com/example/sns/config/filter/JwtTokenFilter.java

@@ -16,6 +16,7 @@ import javax.servlet.ServletException;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import java.io.IOException;
+import java.util.List;
 
 @Slf4j
 @RequiredArgsConstructor
@@ -23,10 +24,20 @@ public class JwtTokenFilter extends OncePerRequestFilter {
 
     private final String key;
     private final UserService userService;
+    private final static List<String> TOKEN_IN_PARAM_URLS = List.of(
+            "/api/v1/users/alarm/subscribe"
+    );
 
     @Override
     protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain) throws ServletException, IOException {
 
+        final String token;
+
+        try {
+            if (TOKEN_IN_PARAM_URLS.contains(request.getRequestURI())) {
+                log.info("Request with {} check the query param", request.getRequestURI());
+                token = request.getQueryString().split("=")[1].trim();
+            } else {
                 final String header = request.getHeader(HttpHeaders.AUTHORIZATION);
 
                 if (header == null || !header.startsWith("Bearer ")) {
@@ -34,9 +45,8 @@ public class JwtTokenFilter extends OncePerRequestFilter {
                     filterChain.doFilter(request, response);
                     return;
                 }
-
+                token = header.split(" ")[1].trim();
-        try {
+            }
-            final String token = header.split(" ")[1].trim();
 
             if (JwtTokenUtils.isExpired(token, key)) {
                 log.error("Token is expired");

simple_sns/src/main/java/com/example/sns/controller/UserController.java

@@ -9,6 +9,7 @@ import com.example.sns.controller.response.UserLoginResponse;
 import com.example.sns.exception.ErrorCode;
 import com.example.sns.exception.SnsApplicationException;
 import com.example.sns.model.User;
+import com.example.sns.service.AlarmService;
 import com.example.sns.service.UserService;
 import com.example.sns.util.ClassUtils;
 import lombok.RequiredArgsConstructor;
@@ -16,6 +17,7 @@ import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
 import org.springframework.security.core.Authentication;
 import org.springframework.web.bind.annotation.*;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
 @RestController
 @RequestMapping("/api/v1/users")
@@ -23,6 +25,7 @@ import org.springframework.web.bind.annotation.*;
 public class UserController {
 
     private final UserService userService;
+    private final AlarmService alarmService;
 
     @PostMapping("/join")
     public Response<UserJoinResponse> join(@RequestBody UserJoinRequest request) {
@@ -48,4 +51,17 @@ public class UserController {
                 );
         return Response.success(userService.alarmList(user.getId(), pageable).map(AlarmResponse::fromAlarm));
     }
+
+    @GetMapping("/alarm/subscribe")
+    public SseEmitter subscribe(Authentication authentication) {
+        User user = ClassUtils.getSafeCastInstance(authentication.getPrincipal(), User.class)
+                .orElseThrow(
+                        () ->  new SnsApplicationException(
+                                ErrorCode.INTERNAL_SERVER_ERROR,
+                                "Casting to User class failed"
+                        )
+                );
+
+        return alarmService.connectAlarm(user.getId());
+    }
 }

simple_sns/src/main/java/com/example/sns/exception/ErrorCode.java

@@ -16,6 +16,7 @@ public enum ErrorCode {
     INVALID_PERMISSION(HttpStatus.UNAUTHORIZED, "Permission is invalid."),
     INTERNAL_SERVER_ERROR(HttpStatus.INTERNAL_SERVER_ERROR, "Internal server error."),
     ALREADY_LIKED(HttpStatus.CONFLICT, "User already liked the post."),
+    ALARM_CONNECT_ERROR(HttpStatus.INTERNAL_SERVER_ERROR, "Connecting alarm occurs error."),
 
     ;
 

simple_sns/src/main/java/com/example/sns/repository/EmitterRepository.java

@@ -0,0 +1,38 @@
+package com.example.sns.repository;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Repository;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+@Slf4j
+@Repository
+public class EmitterRepository {
+
+    private final Map<String, SseEmitter> emitterMap = new HashMap<>();
+
+    public SseEmitter save(Integer userId, SseEmitter sseEmitter) {
+        final String key = getKey(userId);
+        emitterMap.put(key, sseEmitter);
+        log.info("Set sseEmitter for {}", userId);
+        return sseEmitter;
+    }
+
+    public Optional<SseEmitter> get(Integer userId) {
+        final String key = getKey(userId);
+        SseEmitter sseEmitter = emitterMap.get(key);
+        log.info("Get sseEmitter for {}", userId);
+        return Optional.ofNullable(sseEmitter);
+    }
+
+    public void delete(Integer userid) {
+        emitterMap.remove(getKey(userid));
+    }
+
+    private String getKey(Integer userId) {
+        return "Emitter:UID:" + userId;
+    }
+}

simple_sns/src/main/java/com/example/sns/service/AlarmService.java

@@ -0,0 +1,53 @@
+package com.example.sns.service;
+
+import com.example.sns.exception.ErrorCode;
+import com.example.sns.exception.SnsApplicationException;
+import com.example.sns.repository.EmitterRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.io.IOException;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class AlarmService {
+
+    private final static Long DEFAULT_TIMEOUT = 60L * 1000 * 60;
+    private final static String ALARM_NAME = "alarm";
+    private final EmitterRepository emitterRepository;
+
+    public SseEmitter connectAlarm(Integer userId) {
+        SseEmitter sseEmitter = new SseEmitter(DEFAULT_TIMEOUT);
+        emitterRepository.save(userId, sseEmitter);
+
+        sseEmitter.onCompletion(() -> {
+            emitterRepository.delete(userId);
+        });
+
+        sseEmitter.onTimeout(() -> {
+            emitterRepository.delete(userId);
+        });
+
+        try {
+            sseEmitter.send(SseEmitter.event().id("").name(ALARM_NAME).data("connect completed"));
+        } catch (IOException e) {
+            throw new SnsApplicationException(ErrorCode.ALARM_CONNECT_ERROR);
+        }
+
+        return sseEmitter;
+    }
+
+    public void send(Integer alarmId, Integer userId) {
+        emitterRepository.get(userId).ifPresentOrElse(sseEmitter -> {
+            try {
+                sseEmitter.send(SseEmitter.event().id(alarmId.toString()).name(ALARM_NAME).data("new alarm"));
+            } catch (IOException e) {
+                emitterRepository.delete(userId);
+                throw new SnsApplicationException(ErrorCode.ALARM_CONNECT_ERROR);
+            }
+        }, () -> log.info("No emitter founded."));
+    }
+}

simple_sns/src/main/java/com/example/sns/service/PostService.java

@@ -24,6 +24,7 @@ public class PostService {
     private final LikeEntityRepository likeEntityRepository;
     private final CommentEntityRepository commentEntityRepository;
     private final AlarmEntityRepository alarmEntityRepository;
+    private final AlarmService alarmService;
 
     @Transactional
     public void create(String title, String body, String username) {
@@ -94,12 +95,14 @@ public class PostService {
         // save like
         likeEntityRepository.save(LikeEntity.of(userEntity, postEntity));
 
-        alarmEntityRepository.save(AlarmEntity.of(
+        AlarmEntity alarmEntity = alarmEntityRepository.save(AlarmEntity.of(
                         postEntity.getUser(),
                         AlarmType.NEW_LIKE_ON_POST,
                         new AlarmArgs(userEntity.getId(), postEntity.getId())
                 )
         );
+
+        alarmService.send(alarmEntity.getId(), postEntity.getUser().getId());
     }
 
     public long likeCount(Integer postId) {
@@ -116,12 +119,14 @@ public class PostService {
 
         commentEntityRepository.save(CommentEntity.of(userEntity, postEntity, comment));
 
-        alarmEntityRepository.save(AlarmEntity.of(
+        AlarmEntity alarmEntity = alarmEntityRepository.save(AlarmEntity.of(
                         postEntity.getUser(),
                         AlarmType.NEW_COMMENT_ON_POST,
                         new AlarmArgs(userEntity.getId(), postEntity.getId())
                 )
         );
+
+        alarmService.send(alarmEntity.getId(), postEntity.getUser().getId());
     }
 
     public Page<Comment> getComments(Integer postId, Pageable pageable) {