Merge pull request #78 from c-cube/wip-ws

add a websocket library

.github/workflows/gh-pages.yml

@@ -3,7 +3,7 @@ name: github pages
 on:
   push:
     branches:
-      - master
+      - main
 
 jobs:
   deploy:

.github/workflows/main.yml

@@ -3,9 +3,8 @@ name: build
 on:
   pull_request:
   push:
-  schedule:
-    # Prime the caches every Monday
-    - cron: 0 1 * * MON
+    branches:
+      - main
 
 jobs:
   build:
@@ -32,9 +31,6 @@ jobs:
         with:
           ocaml-compiler: ${{ matrix.ocaml-compiler }}
           allow-prerelease-opam: true
-          opam-local-packages: |
-            ./tiny_httpd.opam
-            ./tiny_httpd_camlzip.opam
           opam-depext-flags: --with-test
 
       - run: opam install ./tiny_httpd.opam ./tiny_httpd_camlzip.opam --deps-only --with-test

echo_ws.sh

@@ -0,0 +1,2 @@
+#!/bin/sh
+exec dune exec --display=quiet --profile=release "examples/echo_ws.exe" -- $@

examples/dune

@@ -20,6 +20,12 @@
  (modules writer)
  (libraries tiny_httpd logs))
 
+(executable
+ (name echo_ws)
+ (flags :standard -warn-error -a+8)
+ (modules echo_ws)
+ (libraries tiny_httpd tiny_httpd.ws logs))
+
 (rule
  (targets test_output.txt)
  (deps

examples/echo_ws.ml

@@ -0,0 +1,67 @@
+module S = Tiny_httpd
+module Log = Tiny_httpd.Log
+module IO = Tiny_httpd_io
+
+let setup_logging ~debug () =
+  Logs.set_reporter @@ Logs.format_reporter ();
+  Logs.set_level ~all:true
+  @@ Some
+       (if debug then
+         Logs.Debug
+       else
+         Logs.Info)
+
+let handle_ws _client_addr ic oc =
+  Log.info (fun k ->
+      k "new client connection from %s"
+        (Tiny_httpd_util.show_sockaddr _client_addr));
+
+  let (_ : Thread.t) =
+    Thread.create
+      (fun () ->
+        while true do
+          Thread.delay 3.;
+          IO.Output.output_string oc "(special ping!)";
+          IO.Output.flush oc
+        done)
+      ()
+  in
+
+  let buf = Bytes.create 32 in
+  let continue = ref true in
+  while !continue do
+    let n = IO.Input.input ic buf 0 (Bytes.length buf) in
+    Log.debug (fun k ->
+        k "echo %d bytes from websocket: %S" n (Bytes.sub_string buf 0 n));
+
+    if n = 0 then continue := false;
+    IO.Output.output oc buf 0 n;
+    IO.Output.flush oc
+  done;
+  Log.info (fun k -> k "client exiting")
+
+let () =
+  let port_ = ref 8080 in
+  let j = ref 32 in
+  let debug = ref false in
+  Arg.parse
+    (Arg.align
+       [
+         "--port", Arg.Set_int port_, " set port";
+         "-p", Arg.Set_int port_, " set port";
+         "--debug", Arg.Set debug, " enable debug";
+         "-j", Arg.Set_int j, " maximum number of connections";
+       ])
+    (fun _ -> raise (Arg.Bad ""))
+    "echo [option]*";
+  setup_logging ~debug:!debug ();
+
+  let server = S.create ~port:!port_ ~max_connections:!j () in
+  Tiny_httpd_ws.add_route_handler server
+    S.Route.(exact "echo" @/ return)
+    handle_ws;
+
+  Printf.printf "listening on http://%s:%d\n%!" (S.addr server) (S.port server);
+  match S.run server with
+  | Ok () -> ()
+  | Error e -> raise e

src/Tiny_httpd_io.ml

@@ -45,12 +45,60 @@ module Input = struct
             Unix.close fd);
     }
 
+  let of_slice (i_bs : bytes) (i_off : int) (i_len : int) : t =
+    let i_off = ref i_off in
+    let i_len = ref i_len in
+    {
+      input =
+        (fun buf i len ->
+          let n = min len !i_len in
+          Bytes.blit i_bs !i_off buf i n;
+          i_off := !i_off + n;
+          i_len := !i_len - n;
+          n);
+      close = ignore;
+    }
+
   (** Read into the given slice.
       @return the number of bytes read, [0] means end of input. *)
   let[@inline] input (self : t) buf i len = self.input buf i len
 
+  (** Read exactly [len] bytes.
+      @raise End_of_file if the input did not contain enough data. *)
+  let really_input (self : t) buf i len : unit =
+    let i = ref i in
+    let len = ref len in
+    while !len > 0 do
+      let n = input self buf !i !len in
+      if n = 0 then raise End_of_file;
+      i := !i + n;
+      len := !len - n
+    done
+
   (** Close the channel. *)
   let[@inline] close self : unit = self.close ()
+
+  let append (i1 : t) (i2 : t) : t =
+    let use_i1 = ref true in
+    let rec input buf i len : int =
+      if !use_i1 then (
+        let n = i1.input buf i len in
+        if n = 0 then (
+          use_i1 := false;
+          input buf i len
+        ) else
+          n
+      ) else
+        i2.input buf i len
+    in
+
+    {
+      input;
+      close =
+        (fun () ->
+          close i1;
+          close i2);
+    }
 end
 
 (** Output channel (byte sink) *)