Merge pull request #17 from c-cube/wip-server-sent

server sent events

examples/dune

@@ -0,0 +1,26 @@
+
+(executable
+  (name sse_server)
+  (modules sse_server)
+  (libraries tiny_httpd unix ptime ptime.clock.os))
+
+(executable
+  (name sse_client)
+  (modules sse_client)
+  (libraries unix))
+
+(rule
+  (targets test_output.txt)
+  (deps (:script ./run_test.sh) ./sse_client.exe ./sse_server.exe)
+  (enabled_if (= %{system} "linux"))
+  (package tiny_httpd)
+  (action
+    (with-stdout-to %{targets} (run %{script}))))
+
+(rule
+  (alias runtest)
+  (package tiny_httpd)
+  (enabled_if (= %{system} "linux"))
+  (deps test_output.txt)
+  (action
+    (diff test_output.txt.expected test_output.txt)))

examples/run_test.sh

@@ -0,0 +1,10 @@
+#!/usr/bin/env bash
+
+PORT=8082
+
+./sse_server.exe -p $PORT &
+sleep 0.1
+./sse_client.exe -p $PORT --alarm=1 /count | tr -d '\r' || true
+
+kill %1
+echo "success"

examples/sse_client.ml

@@ -0,0 +1,31 @@
+let addr = ref "127.0.0.1"
+let port = ref 8080
+let path = ref "/clock"
+
+let bufsize = 1024
+
+let () =
+  Arg.parse (Arg.align [
+      "-h", Arg.Set_string addr, " address to connect to";
+      "-p", Arg.Set_int port, " port to connect to";
+      "--alarm", Arg.Int (fun i->Unix.alarm i|>ignore), " set alarm (in seconds)";
+    ]) (fun s -> path := s) "sse_client [opt]* path?";
+
+  Format.printf "connect to %s:%d@." !addr !port;
+  let sock = Unix.socket Unix.PF_INET Unix.SOCK_STREAM 0 in
+  Unix.connect sock (Unix.ADDR_INET (Unix.inet_addr_of_string !addr, !port));
+  Unix.setsockopt sock Unix.TCP_NODELAY false;
+
+  let ic = Unix.in_channel_of_descr sock in
+  let oc = Unix.out_channel_of_descr sock in
+  Printf.fprintf oc "GET %s HTTP/1.1\r\nHost: localhost\r\n\r\n" !path;
+  flush oc;
+
+  let continue = ref true in
+  let buf = Bytes.create bufsize in
+  while !continue do
+    let n = input ic buf 0 bufsize in
+    if n=0 then continue := false;
+    output stdout buf 0 n; flush stdout
+  done;
+  Format.printf "exit!@."

examples/sse_server.ml

@@ -0,0 +1,54 @@
+
+(* serves some streams of events *)
+
+module S = Tiny_httpd
+
+let port = ref 8080
+
+let () =
+  Arg.parse (Arg.align [
+      "-p", Arg.Set_int port, " port to listen on";
+      "--debug", Arg.Bool S._enable_debug, " toggle debug";
+    ]) (fun _ -> ()) "sse_clock [opt*]";
+  let server = S.create ~port:!port () in
+
+  let extra_headers = [
+    "Access-Control-Allow-Origin", "*";
+    "Access-Control-Allow-Methods", "POST, GET, OPTIONS";
+  ] in
+
+  (* tick/tock goes the clock *)
+  S.add_route_server_sent_handler server S.Route.(exact "clock" @/ return)
+    (fun _req (module EV : S.SERVER_SENT_GENERATOR) ->
+       S._debug (fun k->k"new connection");
+       EV.set_headers extra_headers;
+       let tick = ref true in
+       while true do
+         let now = Ptime_clock.now() in
+         S._debug (fun k->k"send clock ev %s" (Format.asprintf "%a" Ptime.pp now));
+         EV.send_event ~event:(if !tick then "tick" else "tock")
+           ~data:(Ptime.to_rfc3339 now) ();
+         tick := not !tick;
+
+         Unix.sleepf 1.0;
+       done;
+    );
+
+  (* just count *)
+  S.add_route_server_sent_handler server S.Route.(exact "count" @/ return)
+    (fun _req (module EV : S.SERVER_SENT_GENERATOR)  ->
+       let n = ref 0 in
+       while true do
+         EV.send_event ~data:(string_of_int !n) ();
+         incr n;
+         Unix.sleepf 0.1;
+       done;
+    );
+
+  Printf.printf "listening on http://localhost:%d/\n%!" (S.port server);
+  match S.run server with
+  | Ok () -> ()
+  | Error e ->
+    Printf.eprintf "error: %s\n%!" (Printexc.to_string e); exit 1
+
+

examples/test_output.txt.expected

@@ -0,0 +1,26 @@
+listening on http://localhost:8082/
+connect to 127.0.0.1:8082
+HTTP/1.1 200 OK
+content-type: text/event-stream
+
+data: 0
+
+data: 1
+
+data: 2
+
+data: 3
+
+data: 4
+
+data: 5
+
+data: 6
+
+data: 7
+
+data: 8
+
+data: 9
+
+success