V03.01+sse (#19)

* Started SSE

* SSE Success!

* Added closed and removed debug

* Update pom.xml

Co-authored-by: Katsute <Katsute@users.noreply.github.com>

Author: Katsute

.gitignore

@@ -5,8 +5,8 @@ out
 # Project ignore
 .docs
 src/LICENSE.txt
-src/_ignore
+/src/main/java/com/kttdevelopment/simplehttpserver/_ignore/
 # Gradle
 .gradle
 # Maven
-target
+target

README.md

@@ -22,6 +22,8 @@ Get requests are parsed into a hashmap.
 Post requests are parsed into a hashmap (`multipart/form-data` as well!).
 
 ## Http Handlers
+### SSE Handler
+Sever sent event events made easy.
 ### Redirect Handler
 A simple handler to redirect links without leaving a mark in the history.
 ### Predicate Handler

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.kttdevelopment</groupId>
     <artifactId>simplehttpserver</artifactId>
-    <version>03.00.01</version>
+    <version>03.01.00</version>
     <packaging>jar</packaging>
 
     <dependencies>

src/main/java/com/kttdevelopment/simplehttpserver/SimpleHttpExchange.java

@@ -5,6 +5,7 @@
 import com.kttdevelopment.simplehttpserver.var.RequestMethod;
 
 import java.io.IOException;
+import java.io.OutputStream;
 import java.net.InetSocketAddress;
 import java.net.URI;
 import java.util.HashMap;
@@ -335,6 +336,15 @@ static SimpleHttpExchange create(final HttpExchange exchange){
 
 //
 
+    /**
+     * Returns the exchange response stream.
+     * @return exchange output stream
+     *
+     * @since 03.01.00
+     * @author Ktt Development
+     */
+    public abstract OutputStream getOutputStream();
+
     /**
      * Sends response headers to the client.
      *

src/main/java/com/kttdevelopment/simplehttpserver/SimpleHttpExchangeImpl.java

@@ -62,6 +62,8 @@ static SimpleHttpExchange createSimpleHttpExchange(final HttpExchange exchange){
 
             private final HashMap<String,String> cookies;
 
+            private final OutputStream outputStream;
+
         //
 
             private final Function<String,HashMap<String,String>> parseWwwFormEnc = s -> {
@@ -191,6 +193,8 @@ static SimpleHttpExchange createSimpleHttpExchange(final HttpExchange exchange){
                         cookies.put(value[0],value[1]);
                     }
                 }
+
+                outputStream = exchange.getResponseBody();
             }
 
         //
@@ -325,6 +329,11 @@ public synchronized final HttpSession getHttpSession(){
                 return session;
             }
 
+            @Override
+            public final OutputStream getOutputStream(){
+                return outputStream;
+            }
+
             //
 
             @Override
@@ -343,7 +352,7 @@ public synchronized final void send(final byte[] response) throws IOException{
             }
 
             @Override
-            public void send(final byte[] response, final boolean gzip) throws IOException{
+            public final void send(final byte[] response, final boolean gzip) throws IOException{
                 send(response, HttpCode.HTTP_OK, gzip);
             }
 
@@ -353,7 +362,7 @@ public synchronized final void send(final byte[] response, final int responseCod
             }
 
             @Override
-            public void send(final byte[] response, final int responseCode, final boolean gzip) throws IOException{
+            public final void send(final byte[] response, final int responseCode, final boolean gzip) throws IOException{
                 if(gzip){
                     exchange.getResponseHeaders().set("Accept-Encoding","gzip");
                     exchange.getResponseHeaders().set("Content-Encoding","gzip");
@@ -379,7 +388,7 @@ public synchronized final void send(final String response) throws IOException{
             }
 
             @Override
-            public void send(final String response, final boolean gzip) throws IOException{
+            public final void send(final String response, final boolean gzip) throws IOException{
                 send(response.getBytes(StandardCharsets.UTF_8), HttpCode.HTTP_OK, gzip);
             }
 
@@ -389,14 +398,17 @@ public synchronized final void send(final String response, final int responseCod
             }
 
             @Override
-            public void send(final String response, final int responseCode, final boolean gzip) throws IOException{
+            public final void send(final String response, final int responseCode, final boolean gzip) throws IOException{
                 send(response.getBytes(StandardCharsets.UTF_8),responseCode,gzip);
             }
 
             //
 
             @Override
             public synchronized final void close(){
+                try{
+                    outputStream.close();
+                }catch(final IOException ignored){ }
                 exchange.close();
             }
 

src/main/java/com/kttdevelopment/simplehttpserver/handler/FileHandler.java

@@ -676,6 +676,7 @@ public final void handle(final SimpleHttpExchange exchange) throws IOException{
             }
             handle(exchange,null,null); // not found
         }
+        exchange.close();
     }
 
     /**

src/main/java/com/kttdevelopment/simplehttpserver/handler/RedirectHandler.java

@@ -33,13 +33,9 @@ public RedirectHandler(final String link){
 
     @Override
     public final void handle(final SimpleHttpExchange exchange) throws IOException{
-        try{
-            System.out.println("started");
-            exchange.getResponseHeaders().set("Location", link);
-            exchange.send(HttpCode.HTTP_Found);
-        }catch(final Exception e){
-            e.printStackTrace();
-        }
+        exchange.getResponseHeaders().set("Location", link);
+        exchange.send(HttpCode.HTTP_Found);
+        exchange.close();
     }
 
 //

src/main/java/com/kttdevelopment/simplehttpserver/handler/SSEHandler.java

@@ -0,0 +1,89 @@
+package com.kttdevelopment.simplehttpserver.handler;
+
+import com.kttdevelopment.simplehttpserver.SimpleHttpExchange;
+import com.kttdevelopment.simplehttpserver.SimpleHttpHandler;
+import com.kttdevelopment.simplehttpserver.var.HttpCode;
+import com.kttdevelopment.simplehttpserver.var.RequestMethod;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class SSEHandler extends SimpleHttpHandler {
+
+    private final List<OutputStream> listeners = new ArrayList<>();
+    private final AtomicInteger      eventId   = new AtomicInteger(-1);
+    private final LinkedList<EventStreamRecord> queue = new LinkedList<>();
+
+    @Override
+    public final void handle(final SimpleHttpExchange exchange) throws IOException{
+        exchange.getResponseHeaders().add("Access-Control-Allow-Headers", "Content-Type");
+        if(exchange.getRequestHeaders().getFirst("origin") != null)
+            exchange.getResponseHeaders().add("Access-Control-Allow-Origin", exchange.getRequestHeaders().getFirst("origin"));
+        exchange.getResponseHeaders().add("Access-Control-Allow-Methods","GET, HEAD, POST, PUT, DELETE");
+        exchange.getResponseHeaders().add("Access-Control-Max-Age", String.valueOf(TimeUnit.HOURS.toSeconds(1)));
+
+        if(exchange.getRequestMethod() == RequestMethod.OPTIONS){
+            exchange.sendResponseHeaders(HttpCode.HTTP_OK,0);
+            return;
+        }
+
+        exchange.getResponseHeaders().put("content-type", Collections.singletonList("text/event-stream"));
+
+        int latest = 0;
+        try{
+            latest = Integer.parseInt(exchange.getRequestHeaders().getFirst("Last_Event-ID"));
+        }catch(final NumberFormatException | NullPointerException ignored){ }
+
+        exchange.sendResponseHeaders(200,0);
+        for(int index = latest; index < queue.size(); index++){
+            exchange.getOutputStream().write(queue.get(index).toString(eventId.get()).getBytes(StandardCharsets.UTF_8));
+            exchange.getOutputStream().flush();
+        }
+
+        listeners.add(exchange.getOutputStream());
+    }
+
+    public synchronized final void push(final String data){
+        push(data,0,"");
+    }
+
+    public synchronized final void push(final String data, final int retry, final String event){
+        eventId.addAndGet(1);
+        final EventStreamRecord record = new EventStreamRecord(retry,event,data);
+        queue.add(record);
+        listeners.forEach(stream -> {
+            try{
+                stream.write(record.toString(eventId.get()).getBytes(StandardCharsets.UTF_8));
+                stream.flush();
+            }catch(final IOException ignored){ }
+        });
+    }
+
+    private static class EventStreamRecord {
+
+            private final int retry;
+            private final String event;
+            private final String data;
+
+            public EventStreamRecord(final int retry, final String event, final String data){
+                this.retry = retry;
+                this.event = event;
+                this.data = data;
+            }
+
+            public final String toString(final int id){
+                return
+                    "id: " + id + '\n' +
+                    (retry > 0 ? "retry: " + retry + '\n' : "") +
+                    (!event.isBlank() ? "event: " + event + '\n' : "") +
+                    (!data.isBlank() ? "data: " + data + '\n' : "") +
+                    '\n';
+            }
+
+        }
+
+}