当前位置：网站首页>Practice of real-time push demo of three web messages: long polling, iframe and SSE

Practice of real-time push demo of three web messages: long polling, iframe and SSE

2022-07-28 20:16:00 【Fire king】

Long polling ,iframe and sse Three web Message push in real time demo practice

Long polling

@Controller
@RequestMapping("/polling")
public class PollingController {
    

    private static final Long TIME_OUT = 10000l;
    //  Store and monitor a Id Long polling set for 
    //  Thread synchronization structure 
    public static Multimap<String, DeferredResult<String>> watchRequests = Multimaps.synchronizedMultimap(HashMultimap.create());

    /** * *  Set listening  */
    @GetMapping(path = "watch/{id}")
    @ResponseBody
    public DeferredResult<String> watch(@PathVariable String id) {
    
        //  Delay object setting timeout 
        DeferredResult<String> deferredResult = new DeferredResult<>(TIME_OUT);
        //  Remove when the asynchronous request completes  key, Prevent memory overflow 
        deferredResult.onCompletion(() -> {
    
            watchRequests.remove(id, deferredResult);
        });
        //  Registrar polling request 
        watchRequests.put(id, deferredResult);
        return deferredResult;
    }

    /** * *  Change data  */
    @GetMapping(path = "publish/{id}")
    @ResponseBody
    public String publish(@PathVariable String id) {
    
        //  Data changes   Take out the monitor ID All long polling requests ( Multiple browsers , Different windows of a single browser only think of one , Every time publish Just change one ), And respond one by one 
        if (watchRequests.containsKey(id)) {
    
            Collection<DeferredResult<String>> deferredResults = watchRequests.get(id);
            System.out.println(deferredResults.size());
            for (DeferredResult<String> deferredResult : deferredResults) {
    
                // as long as set, Think it's changed 
                deferredResult.setResult(" I updated " + new Date());
            }
        }
        return "success";
    }
    // When the request exceeds the set timeout , Will throw out AsyncRequestTimeoutException abnormal , It's directly used here @ControllerAdvice Global capture and unified return , After obtaining the agreed status code, the front end sends a long polling request again , Call back and forth like this .

    @ControllerAdvice
    public class AsyncRequestTimeoutHandler {
    

        @ResponseStatus(HttpStatus.NOT_MODIFIED)
        @ResponseBody
        @ExceptionHandler(AsyncRequestTimeoutException.class)
        public String asyncRequestTimeoutHandler(AsyncRequestTimeoutException e) {
    
            System.out.println(" Asynchronous request timed out ");
            return "304";
        }
    }
}

iframe

IframeController

@Controller
@RequestMapping("/iframe")
public class IframeController {
    
    @GetMapping(path = "message")
    public void message(HttpServletResponse response) throws IOException, InterruptedException {
    
       int count = 0;
        while (true) {
    
            count++;
            response.setHeader("Pragma", "no-cache");
            response.setDateHeader("Expires", 0);
            response.setHeader("Cache-Control", "no-cache,no-store");
            response.setStatus(HttpServletResponse.SC_OK);
            response.getWriter().print(" <script type=\"text/javascript\">\n" +
                    "parent.document.getElementById('clock').innerHTML = \"" + count + "\";" +
                    "parent.document.getElementById('count').innerHTML = \"" + count + "\";" +
                    "</script>");
        }
    }

    @GetMapping(path = "toIframePage")
    public String toIframePage() {
    
        return "/iframe";
    }
}

iframe.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>iframe Long polling </title>
</head>
<body>
<iframe src="/iframe/message" style="display:none"></iframe>
<h1 id="clock"></h1>
<h1 id="count"></h1>
</body>
</html>

sse

SseEmitterServer

@Slf4j
public class SseEmitterServer {
    
    private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();
    /** *  Current connections  */
    private static AtomicInteger count = new AtomicInteger(0);
    /** *  Create connection  * * @date: 2022/7/12 14:51 * @auther: */
    public static SseEmitter connect(String userId) {
    
        try {
    
            //  Set timeout ,0 No expiration date . Default 30 second 
            SseEmitter sseEmitter = new SseEmitter(0L);
            //  Register callback 
            sseEmitter.onCompletion(completionCallBack(userId));
            sseEmitter.onError(errorCallBack(userId));
            sseEmitter.onTimeout(timeoutCallBack(userId));
            sseEmitterMap.put(userId, sseEmitter);
            count.getAndIncrement();
                return sseEmitter;
        } catch (Exception e) {
    
            log.info(" Create a new sse Abnormal connection , The current user ：{}", userId);
        }
        return null;
    }

    private static Consumer<Throwable> errorCallBack(String userId) {
    
        return throwable -> {
    
            log.info(" Abnormal connection ：{}", userId);
            removeUser(userId);
        };
    }

    private static Runnable timeoutCallBack(String userId) {
    
        return () -> {
    
            log.info(" Connection timeout ：{}", userId);
            removeUser(userId);
        };
    }

    private static Runnable completionCallBack(String userId) {
    
        return () -> {
    
            log.info(" End connection ：{}", userId);
            removeUser(userId);
        };
    }

    public static void removeUser(String userId) {
    
        sseEmitterMap.remove(userId);
        //  Number -1
        count.getAndDecrement();
        log.info(" Remove users ：{}", userId);
    }

    /** *  Send a message to the specified user  * * @date: 2022/7/12 14:51 * @auther: */
    public static void sendMessage(String userId, String message) {
    

        if (sseEmitterMap.containsKey(userId)) {
    
            try {
    
                sseEmitterMap.get(userId).send(message);
                // Different SpringBoot The version of may need to add 
                // sseEmitterMap.get(userId).complete();
            } catch (IOException e) {
    
                log.error(" user [{}] Push exception :{}", userId, e.getMessage());
                removeUser(userId);
            }
        }
    }
}

SseEmitterController

@CrossOrigin
@RestController
@RequestMapping("/sse")
public class SseEmitterController {
    
    /** *  Used to create a connection  */
    @GetMapping("/connect/{userId}")
    public SseEmitter connect(@PathVariable String userId) {
    
        return SseEmitterServer.connect(userId);
    }

    /** *  Push to everyone  * * @param message * @return */
   /* @GetMapping("/push/{message}") public ResponseEntity<String> push(@PathVariable(name = "message") String message) { SseEmitterServer.batchSendMessage(message); return ResponseEntity.ok("WebSocket  Push messages to everyone "); }*/

    /** *  Send to a single person  * * @param message * @param userid * @return */
    @RequestMapping("/push_one")
    public ResponseEntity<String> pushOne( String message,  String userid) {
    
        SseEmitterServer.sendMessage(userid, message);
        return ResponseEntity.ok("WebSocket  Push a message to " + userid);
    }

    /** *  Close the connection  */
    @GetMapping("/close/{userid}")
    public ResponseEntity<String> close(@PathVariable("userid") String userid) {
    
        SseEmitterServer.removeUser(userid);
        return ResponseEntity.ok(" Connection is closed ");
    }

}

PageController

@Controller
public class PageController {
    

    @RequestMapping("/toSsePage")
    public String toSsePage() {
    
        return "/sse";
    }
}

sse.html

<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <title>SseEmitter</title>
</head>

<body>
<button onclick="closeSse()"> Close the connection </button>
<div id="message"></div>
</body>
<script> let source = null; //  Simulate login user with timestamp  //const userId = new Date().getTime(); const userId = "1"; if (window.EventSource) {
       //  Establishing a connection  source = new EventSource('http://localhost:8080/sse/connect/' + userId); /** *  Once the connection is established , It will trigger open event  *  Another way of writing ：source.onopen = function (event) {} */ source.addEventListener('open', function(e) {
       setMessageInnerHTML(" Establishing a connection ..."); }, false); /** *  The client receives the data from the server  *  Another way of writing ：source.onmessage = function (event) {} */ source.addEventListener('message', function(e) {
       setMessageInnerHTML(e.data); }); /** *  If a communication error occurs （ For example, the connection is broken ）, It will trigger error event  *  perhaps ： *  Another way of writing ：source.onerror = function (event) {} */ source.addEventListener('error', function(e) {
       if (e.readyState === EventSource.CLOSED) {
       setMessageInnerHTML(" Connection is closed "); } else {
       console.log(e); } }, false); } else {
       setMessageInnerHTML(" Your browser does not support it SSE"); } //  Listen for window closing events , Active close sse Connect , If the server settings never expire , After the browser is closed, manually clean up the server data  window.onbeforeunload = function() {
       closeSse(); }; //  close Sse Connect  function closeSse() {
       source.close(); const httpRequest = new XMLHttpRequest(); httpRequest.open('GET', 'http://localhost:8080/sse/close/' + userId, true); httpRequest.send(); console.log("close"); } //  Displays the message on the web page  function setMessageInnerHTML(innerHTML) {
       document.getElementById('message').innerHTML += innerHTML + '<br/>'; } </script>

</html>