Remove Lwt and use Poll instead

14 files changed, 316 insertions, 382 deletions

diff --git a/bin/spatial/dune b/bin/spatial/dune
index 3682cc9..f0fd72b 100644
--- a/bin/spatial/dune
+++ b/bin/spatial/dune
@@ -1,3 +1,3 @@
 (executable
  (name main)
- (libraries spatial-sway.ipc sway-ipc lwt lwt.unix))
+ (libraries poll spatial-sway.ipc sway-ipc unix))
diff --git a/bin/spatial/main.ml b/bin/spatial/main.ml
index af62117..ee2efac 100644
--- a/bin/spatial/main.ml
+++ b/bin/spatial/main.ml
@@ -1,7 +1,5 @@
 open Sway_ipc_types
 
-type input = From_sway of Event.t | From_client of Spatial_ipc.socket
-
 let workspace_handle (ev : Event.workspace_event) state =
   match ev.change with
   | Focus ->
@@ -10,29 +8,23 @@ let workspace_handle (ev : Event.workspace_event) state =
         | Some workspace -> State.set_current_workspace workspace state
         | None -> state
       in
-      Lwt.return (state, true, None)
-  | Init | Empty | Move | Rename | Urgent | Reload ->
-      Lwt.return (state, false, None)
+      (state, true, None)
+  | Init | Empty | Move | Rename | Urgent | Reload -> (state, false, None)
 
 let window_handle (ev : Event.window_event) state =
-  let open Lwt.Syntax in
   match ev.change with
   | Event.New ->
-      let* () =
-        Lwt_io.printf "created window %Ld (%s)\n" ev.container.id
-          (Option.value ~default:"<meh>" ev.container.app_id)
-      in
       let state =
         State.register_window false 2 state.State.current_workspace state
           ev.container
       in
-      Lwt.return (state, true, None)
+      (state, true, None)
   | Event.Close ->
       let state = State.unregister_window state ev.container.id in
-      Lwt.return (state, true, None)
+      (state, true, None)
   | Event.Focus | Event.Title | Event.Fullscreen_mode | Event.Move | Event.Mark
   | Event.Urgent ->
-      Lwt.return (state, false, None)
+      (state, false, None)
   | Event.Floating -> (
       match ev.container.node_type with
       | Con ->
@@ -40,77 +32,83 @@ let window_handle (ev : Event.window_event) state =
             State.register_window false 2 state.State.current_workspace state
               ev.container
           in
-          Lwt.return (state, true, None)
+          (state, true, None)
       | Floating_con ->
-          let* () =
-            Lwt_io.printf "window %Ld (%s) turned floating\n" ev.container.id
-              (Option.value ~default:"<meh>" ev.container.app_id)
-          in
           let state = State.unregister_window state ev.container.id in
-          Lwt.return (state, true, Some ev.container.id)
-      | _ -> Lwt.return (state, false, None))
+          (state, true, Some ev.container.id)
+      | _ -> (state, false, None))
 
-let event_handle ev state =
-  let open Lwt.Syntax in
-  Lwt.try_bind
-    (fun () ->
-      let* state, arrange, force_focus =
-        match ev with
-        | From_sway (Event.Workspace ev) -> workspace_handle ev state
-        | From_sway (Window ev) -> window_handle ev state
-        | From_sway _ -> assert false
-        | From_client socket ->
-            Lwt.try_bind
-              (fun () ->
-                let+ handle_res =
-                  Spatial_ipc.(
-                    handle_next_command ~socket state
-                      { handler = State.client_command_handle })
-                in
-                match handle_res with Some x -> x | _ -> (state, false, None))
-              Lwt.return
-              (fun exn ->
-                let* () = Spatial_ipc.close socket in
-                raise exn)
-      in
-      let+ () =
-        if arrange then
-          let* () = State.arrange_current_workspace ?force_focus state in
-          (* TODO: Make this more general *)
-          let* _ =
-            Lwt_process.(exec @@ shell "/usr/bin/pkill -SIGRTMIN+8 waybar")
-          in
-          Lwt.return ()
-        else Lwt.return ()
-      in
-      state)
-    Lwt.return
-    (fun exn ->
-      let+ _ =
-        Lwt_io.printf "something went wrong with an event:\n%s\n"
-          (Printexc.to_string exn)
-      in
-      state)
+let with_nonblock_socket socket f =
+  Unix.clear_nonblock socket;
+  let res = f () in
+  Unix.set_nonblock socket;
+  res
+
+let rec go poll state sway_socket server_socket =
+  try
+    match Poll.wait poll (Poll.Timeout.after 10_000_000_000_000L) with
+    | `Timeout -> go poll state sway_socket server_socket
+    | `Ok ->
+        let ref_state = ref state in
+        let ref_arrange = ref false in
+        let ref_force_focus = ref None in
+        Poll.iter_ready poll ~f:(fun fd _event ->
+            try
+              let state, arrange, force_focus =
+                if fd = sway_socket then
+                  with_nonblock_socket fd @@ fun () ->
+                  match Sway_ipc.read_event fd with
+                  | Workspace ev -> workspace_handle ev !ref_state
+                  | Window ev -> window_handle ev !ref_state
+                  | _ -> assert false
+                else if fd = server_socket then (
+                  let client = Spatial_ipc.accept fd in
+                  Unix.set_nonblock client;
+                  Poll.(set poll client Event.read);
+                  (!ref_state, false, None))
+                else
+                  with_nonblock_socket fd @@ fun () ->
+                  match
+                    Spatial_ipc.handle_next_command ~socket:fd !ref_state
+                      { handler = State.client_command_handle }
+                  with
+                  | Some res -> res
+                  | None -> (!ref_state, false, None)
+              in
+              Poll.(set poll fd Event.read);
+              ref_arrange := arrange || !ref_arrange;
+              ref_state := state;
+              ref_force_focus := force_focus
+            with
+            | Mltp_ipc.Socket.Connection_closed
+            when fd <> sway_socket && fd <> server_socket
+            ->
+              Unix.set_nonblock fd;
+              Poll.(set poll fd Event.none);
+              Unix.close fd);
+        if !ref_arrange then (
+          State.arrange_current_workspace ?force_focus:!ref_force_focus
+            !ref_state;
+          (* TODO: Be more configurable about that *)
+          ignore (Unix.system "/usr/bin/pkill -SIGRTMIN+8 waybar"));
+        Poll.clear poll;
+        go poll !ref_state sway_socket server_socket
+  with Unix.Unix_error (EINTR, _, _) ->
+    go poll state sway_socket server_socket
+
+let () =
+  Printexc.record_backtrace true;
+  let poll = Poll.create () in
+
+  let sway_socket = Sway_ipc.subscribe [ Window; Workspace ] in
+  Unix.set_nonblock sway_socket;
+  Poll.(set poll sway_socket Event.read);
 
-let merge_streams l =
-  Lwt_stream.from (fun () -> Lwt.pick (List.map Lwt_stream.get l))
+  let server_socket = Spatial_ipc.create_server () in
+  Unix.set_nonblock server_socket;
+  Poll.(set poll server_socket Event.read);
 
-let main () =
-  let open Lwt.Syntax in
-  let* stream_sway = Sway_ipc.subscribe [ Window; Workspace ] in
-  let* stream_client = Spatial_ipc.create_server () in
-  let stream =
-    merge_streams
-      [
-        Lwt_stream.map (fun x -> From_sway x) stream_sway;
-        Lwt_stream.map (fun x -> From_client x) stream_client;
-      ]
-  in
-  let* state = State.init false 2 in
-  let* () = State.arrange_current_workspace state in
-  let string = Format.asprintf "%a" State.pp state in
-  let* () = Lwt_io.printf "%s\n" string in
-  let* _ = Lwt_stream.fold_s event_handle stream state in
-  Lwt_io.printf "one of the stream has ended\n"
+  let state = State.init false 2 in
+  State.arrange_current_workspace state;
 
-let () = Lwt_main.run @@ main ()
+  go poll state sway_socket server_socket
diff --git a/bin/spatial/ribbon.ml b/bin/spatial/ribbon.ml
index 155b44b..24777ac 100644
--- a/bin/spatial/ribbon.ml
+++ b/bin/spatial/ribbon.ml
@@ -169,7 +169,7 @@ let split_visible ribbon =
   let rec split_visible acc f = function
     | x :: rst when f = 0 -> Some (List.rev acc, x, rst)
     | x :: rst -> split_visible (x :: acc) (f - 1) rst
-    | [] -> None
+    | [] -> raise (Invalid_argument "Ribbon.split_visible")
   in
   match ribbon.visible with None -> None | Some (f, l) -> split_visible [] f l
 
@@ -244,10 +244,12 @@ let move_window_right ribbon =
               match pop_front left with
               | Some (x, left) ->
                   (* Case:   [|a b {f}|]
-                     Result: [|b a {f}|] *)
+                     Result: [|b a {f}|]
+                     In this case, focus remains in the same place, so
+                     [f - 1] *)
                   {
                     ribbon with
-                    visible = Some (f, push_back x left @ [ focus ] @ right);
+                    visible = Some (f - 1, push_back x left @ [ focus ] @ right);
                   }
               | None ->
                   (* Case:   [|{f}|]
diff --git a/bin/spatial/state.ml b/bin/spatial/state.ml
index 495436b..e978166 100644
--- a/bin/spatial/state.ml
+++ b/bin/spatial/state.ml
@@ -85,10 +85,9 @@ let arrange_workspace_commands ?force_focus workspace state =
   | None -> []
 
 let arrange_workspace ?force_focus ~socket workspace state =
-  let open Lwt.Syntax in
   let cmds = arrange_workspace_commands ?force_focus workspace state in
-  let* _replies = Sway_ipc.send_command ~socket (Run_command cmds) in
-  Lwt.return ()
+  let _replies = Sway_ipc.send_command ~socket (Run_command cmds) in
+  ()
 
 let arrange_current_workspace ?force_focus state =
   Sway_ipc.with_socket (fun socket ->
@@ -117,9 +116,8 @@ let unregister_window state window =
   | None -> state
 
 let init default_full_view default_maximum_visible =
-  let open Lwt.Syntax in
-  let* cw = Sway_ipc.get_current_workspace () in
-  let+ tree = Sway_ipc.get_tree () in
+  let cw = Sway_ipc.get_current_workspace () in
+  let tree = Sway_ipc.get_tree () in
   let workspaces = Node.filter (fun x -> x.node_type = Workspace) tree in
   List.fold_left
     (fun state workspace ->
@@ -133,80 +131,76 @@ let init default_full_view default_maximum_visible =
     (empty cw.name) workspaces
 
 let client_command_handle :
-    type a.
-    state -> a Spatial_ipc.t -> ((state * bool * int64 option) option * a) Lwt.t
-    =
+    type a. state -> a Spatial_ipc.t -> (state * bool * int64 option) * a =
  fun state cmd ->
   let open Spatial_ipc in
-  Lwt.return
-  @@ (match cmd with
-      | Run_command cmd ->
-          let res =
-            match cmd with
-            | Focus Left ->
-                ( {
-                    state with
-                    workspaces =
-                      Workspaces_registry.update state.current_workspace
-                        (function
-                          | Some ribbon -> Some (Ribbon.move_focus_left ribbon)
-                          | None -> None)
-                        state.workspaces;
-                  },
-                  true,
-                  None )
-            | Focus Right ->
-                ( {
-                    state with
-                    workspaces =
-                      Workspaces_registry.update state.current_workspace
-                        (function
-                          | Some ribbon -> Some (Ribbon.move_focus_right ribbon)
-                          | None -> None)
-                        state.workspaces;
-                  },
-                  true,
-                  None )
-            | Move Left ->
-                (move_window_left state.current_workspace state, true, None)
-            | Move Right ->
-                (move_window_right state.current_workspace state, true, None)
-            | Maximize Toggle ->
-                (toggle_full_view state.current_workspace state, true, None)
-            | Maximize _ ->
-                (* TODO: implement [On] and [Off] cases *)
-                (state, false, None)
-            | Split Incr ->
-                ( incr_maximum_visible_size state.current_workspace state,
-                  true,
-                  None )
-            | Split Decr ->
-                ( decr_maximum_visible_size state.current_workspace state,
-                  true,
-                  None )
-          in
-          (Some res, { success = true })
-      | Get_windows -> (
-          let ribbon =
-            Workspaces_registry.find_opt state.current_workspace
-              state.workspaces
-          in
-          ( None,
-            match ribbon with
-            | None -> { focus = None; windows = [] }
-            | Some ribbon -> (
-                match ribbon.visible with
-                | Some (f, l) ->
-                    {
-                      focus = Some f;
-                      windows =
-                        List.map
-                          (fun id ->
-                            (Windows_registry.find id state.windows).app_id)
-                          (l @ ribbon.hidden);
-                    }
-                | None -> { focus = None; windows = [] }) ))
-       : _ * a)
+  (match cmd with
+   | Run_command cmd ->
+       let res =
+         match cmd with
+         | Focus Left ->
+             ( {
+                 state with
+                 workspaces =
+                   Workspaces_registry.update state.current_workspace
+                     (function
+                       | Some ribbon -> Some (Ribbon.move_focus_left ribbon)
+                       | None -> None)
+                     state.workspaces;
+               },
+               true,
+               None )
+         | Focus Right ->
+             ( {
+                 state with
+                 workspaces =
+                   Workspaces_registry.update state.current_workspace
+                     (function
+                       | Some ribbon -> Some (Ribbon.move_focus_right ribbon)
+                       | None -> None)
+                     state.workspaces;
+               },
+               true,
+               None )
+         | Move Left ->
+             (move_window_left state.current_workspace state, true, None)
+         | Move Right ->
+             (move_window_right state.current_workspace state, true, None)
+         | Maximize Toggle ->
+             (toggle_full_view state.current_workspace state, true, None)
+         | Maximize _ ->
+             (* TODO: implement [On] and [Off] cases *)
+             (state, false, None)
+         | Split Incr ->
+             ( incr_maximum_visible_size state.current_workspace state,
+               true,
+               None )
+         | Split Decr ->
+             ( decr_maximum_visible_size state.current_workspace state,
+               true,
+               None )
+       in
+       (res, { success = true })
+   | Get_windows -> (
+       let ribbon =
+         Workspaces_registry.find_opt state.current_workspace state.workspaces
+       in
+       ( (state, false, None),
+         match ribbon with
+         | None -> { focus = None; windows = [] }
+         | Some ribbon -> (
+             match ribbon.visible with
+             | Some (f, l) ->
+                 {
+                   focus = Some f;
+                   windows =
+                     List.map
+                       (fun id ->
+                         (Windows_registry.find id state.windows).app_id)
+                       (l @ ribbon.hidden);
+                 }
+             | None -> { focus = None; windows = [] }) ))
+    : _ * a)
 
 let pp fmt state =
   Format.(
diff --git a/bin/spatialmsg/dune b/bin/spatialmsg/dune
index 23ecccc..3f5fa06 100644
--- a/bin/spatialmsg/dune
+++ b/bin/spatialmsg/dune
@@ -1,3 +1,3 @@
 (executable
  (name main)
- (libraries mltp-ipc spatial-sway.ipc lwt lwt.unix clap))
+ (libraries mltp-ipc spatial-sway.ipc clap unix))
diff --git a/bin/spatialmsg/main.ml b/bin/spatialmsg/main.ml
index 50b40c4..9e7be1b 100644
--- a/bin/spatialmsg/main.ml
+++ b/bin/spatialmsg/main.ml
@@ -11,13 +11,13 @@ let () =
       Clap.close ();
 
       let cmd = command_of_string_exn cmd in
-      let { success } = Lwt_main.run (send_command (Run_command cmd)) in
+      let { success } = send_command (Run_command cmd) in
       if not success then exit 1
   | "get_windows" -> (
       let cmd = Clap.optional_int ~placeholder:"INDEX" () in
       Clap.close ();
 
-      let reply = Lwt_main.run (send_command Get_windows) in
+      let reply = send_command Get_windows in
 
       match (cmd, reply.focus) with
       | Some idx, Some focus when idx < List.length reply.windows ->
diff --git a/lib/mltp_ipc/dune b/lib/mltp_ipc/dune
index 6289fab..00ab551 100644
--- a/lib/mltp_ipc/dune
+++ b/lib/mltp_ipc/dune
@@ -1,4 +1,4 @@
 (library
  (name mltp_ipc)
  (public_name mltp-ipc)
- (libraries lwt lwt.unix))
+ (libraries unix))
diff --git a/lib/mltp_ipc/socket.ml b/lib/mltp_ipc/socket.ml
index e90ff65..7a1f50a 100644
--- a/lib/mltp_ipc/socket.ml
+++ b/lib/mltp_ipc/socket.ml
@@ -2,90 +2,85 @@
  * License, v. 2.0. If a copy of the MPL was not distributed with this
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. *)
 
-type socket = Lwt_io.input_channel * Lwt_io.output_channel * Lwt_unix.file_descr
-type error = Bad_magic_string of string | Connection_closed
+type socket = Unix.file_descr
 
-let ( let*! ) x k = Lwt.bind x k
+exception Bad_magic_string of string
+exception Connection_closed
 
-let connect socket_path : socket Lwt.t =
-  let open Lwt.Syntax in
-  let socket = Lwt_unix.socket PF_UNIX SOCK_STREAM 0 in
-  let+ () = Lwt_unix.connect socket (ADDR_UNIX socket_path) in
-  let socket_in = Lwt_io.of_fd ~mode:Input socket in
-  let socket_out = Lwt_io.of_fd ~mode:Output socket in
-  (socket_in, socket_out, socket)
+let connect socket_path : socket =
+  let socket = Unix.socket PF_UNIX SOCK_STREAM 0 in
+  let () = Unix.connect socket (ADDR_UNIX socket_path) in
+  socket
 
-let close (_, _, s) = Lwt_unix.close s
+let close s = Unix.close s
 
 let with_socket socket_path f =
-  let open Lwt.Syntax in
-  let* socket = connect socket_path in
-  Lwt.try_bind
-    (fun () ->
-      let* res = f socket in
-      let* () = close socket in
-      Lwt.return res)
-    Lwt.return
-    (fun exn ->
-      let* () = close socket in
-      raise exn)
+  let socket = connect socket_path in
+  try
+    let res = f socket in
+    close socket;
+    res
+  with exn ->
+    close socket;
+    raise exn
 
-let catch_end_of_file f =
-  Lwt.try_bind f Lwt_result.return @@ function
-  | End_of_file -> Lwt_result.fail Connection_closed
-  | exn -> Lwt.fail exn
-
-let rec read_all ~count ((socket, _, _) as s) =
-  let open Lwt_result.Syntax in
-  let* payload = catch_end_of_file (fun () -> Lwt_io.read ~count socket) in
-  if String.length payload = count then Lwt_result.return payload
-  else
-    let+ rest = read_all ~count:(count - String.length payload) s in
-    payload ^ rest
+let read_all ~count s =
+  let buffer = Bytes.create count in
+  let rec aux ~left =
+    let read = Unix.read s buffer (count - left) left in
+    if read = 0 then raise End_of_file;
+    let remainder = left - read in
+    if remainder = 0 then buffer else aux ~left:remainder
+  in
+  aux ~left:count |> Bytes.to_string
 
 let read_magic_string ~magic_string socket =
-  let open Lwt_result.Syntax in
-  let* msg = read_all ~count:(String.length magic_string) socket in
-  if msg <> magic_string then
-    let*! () = close socket in
-    Lwt_result.fail (Bad_magic_string msg)
-  else Lwt_result.return ()
+  let msg = read_all ~count:(String.length magic_string) socket in
+  if msg <> magic_string then (
+    close socket;
+    raise (Bad_magic_string msg))
 
-let write_raw_message ~magic_string (_, socket, _) raw =
-  let msg = Raw_message.to_string ~magic_string raw in
-  catch_end_of_file @@ fun () -> Lwt_io.write socket msg
+let write_raw_message ~magic_string s raw =
+  try
+    let msg = Raw_message.to_string ~magic_string raw in
+    let _ = Unix.write_substring s msg 0 (String.length msg) in
+    ()
+  with
+  | End_of_file -> raise Connection_closed
+  | exn ->
+      close s;
+      raise exn
 
 let read_raw_message ~magic_string socket =
-  let open Lwt_result.Syntax in
-  let* () = read_magic_string ~magic_string socket in
-  let* msg = read_all ~count:4 socket in
-  let size = Raw_message.string_to_int32 msg in
-  let* msg = read_all ~count:4 socket in
-  let msg_type = Raw_message.string_to_int32 msg in
-  if size <> 0l then
-    let* payload = read_all ~count:(Int32.to_int size) socket in
-    Lwt_result.return (msg_type, payload)
-  else Lwt_result.return (msg_type, "")
+  try
+    read_magic_string ~magic_string socket;
+    let msg = read_all ~count:4 socket in
+    let size = Raw_message.string_to_int32 msg in
+    let msg = read_all ~count:4 socket in
+    let msg_type = Raw_message.string_to_int32 msg in
+    if size <> 0l then
+      let payload = read_all ~count:(Int32.to_int size) socket in
+      (msg_type, payload)
+    else (msg_type, "")
+  with
+  | End_of_file -> raise Connection_closed
+  | exn ->
+      close socket;
+      raise exn
 
 let rec read_next_raw_message ~magic_string socket f =
-  let open Lwt_result.Syntax in
-  let* raw = read_raw_message ~magic_string socket in
-  if f raw then Lwt_result.return raw
-  else read_next_raw_message ~magic_string socket f
+  let raw = read_raw_message ~magic_string socket in
+  if f raw then raw else read_next_raw_message ~magic_string socket f
 
-let socket_handler server () =
-  let open Lwt.Syntax in
-  let+ socket, _ = Lwt_unix.accept server in
-  let socket_in = Lwt_io.of_fd ~mode:Input socket in
-  let socket_out = Lwt_io.of_fd ~mode:Output socket in
-  Some (socket_in, socket_out, socket)
+let accept server =
+  let socket, _ = Unix.accept server in
+  socket
 
 let create_server path =
-  let open Lwt.Syntax in
-  let socket = Lwt_unix.socket PF_UNIX SOCK_STREAM 0 in
-  let* socket_exists = Lwt_unix.file_exists path in
-  let* () = if socket_exists then Lwt_unix.unlink path else Lwt.return () in
-  let sockaddr = Lwt_unix.ADDR_UNIX path in
-  let+ () = Lwt_unix.bind socket sockaddr in
-  let () = Lwt_unix.listen socket 100 in
-  Lwt_stream.from (socket_handler socket)
+  let socket = Unix.socket PF_UNIX SOCK_STREAM 0 in
+  let socket_exists = Sys.file_exists path in
+  if socket_exists then Unix.unlink path;
+  let sockaddr = Unix.ADDR_UNIX path in
+  Unix.bind socket sockaddr;
+  Unix.listen socket 100;
+  socket
diff --git a/lib/mltp_ipc/socket.mli b/lib/mltp_ipc/socket.mli
index f351676..77d01ce 100644
--- a/lib/mltp_ipc/socket.mli
+++ b/lib/mltp_ipc/socket.mli
@@ -10,56 +10,48 @@
         {li A 32-bit integer representing the {b L}ength of the payload}
         {li The {b P}ayload itself}} *)
 
-type socket
+type socket = Unix.file_descr
 (** A socket to communicate with a peer using the so-called MTLP protocol. *)
 
-val connect : string -> socket Lwt.t
+val connect : string -> socket
 (** Establish a bi-directional connection with a peer. *)
 
-val close : socket -> unit Lwt.t
+val close : socket -> unit
 (** Close a bi-directional connection with a peer. *)
 
-val with_socket : string -> (socket -> 'a Lwt.t) -> 'a Lwt.t
+val with_socket : string -> (socket -> 'a) -> 'a
 (** [with_socket path k] establishes a bi-connection with a peer using
     the UNIX socket located at [path], hands over the socket to the
     continuation [k], and takes care of closing the connection prior
     to returning the result, even in case of an exception. *)
 
-type error =
-  | Bad_magic_string of string
-      (** When trying to read a MTLP message, the magic string was not
-          correct. *)
-  | Connection_closed
-      (** When trying to receive from or send a message to a closed
-          bi-directional connection. *)
+exception Bad_magic_string of string
+(** When trying to read a MTLP message, the magic string was not
+    correct. *)
 
-val read_raw_message :
-  magic_string:string -> socket -> (Raw_message.t, error) result Lwt.t
+exception Connection_closed
+(** When trying to receive from or send a message to a closed
+    bi-directional connection. *)
+
+val read_raw_message : magic_string:string -> socket -> Raw_message.t
 (** [read_raw_message ~magic_string socket] reads a MTLP
     message from [socket].
 
-    This function may fail with the following errors:
-
-    {ul {li [Bad_magic_string] (closes [socket] when it happens)}
-        {li [Connection_closed]}} *)
+    @raise Bad_magic_string (closes [socket] when it happens)
+    @raise Connection_closed *)
 
 val read_next_raw_message :
-  magic_string:string ->
-  socket ->
-  (Raw_message.t -> bool) ->
-  (Raw_message.t, error) result Lwt.t
+  magic_string:string -> socket -> (Raw_message.t -> bool) -> Raw_message.t
 (** [read_next_raw_message ~magic_string socket f] returns the next
     raw message received by [socket] which satisfies [f]’s
     conditions. Messages that don’t satisfy [f]’s conditions are
     ignored.
 
-    This function may fail with the following errors:
-
-    {ul {li [Bad_magic_string] (closes [socket] when it happens)}
-        {li [Connection_closed]}} *)
+    @raise Bad_magic_string (closes [socket] when it happens)
+    @raise Connection_closed *)
 
-val write_raw_message :
-  magic_string:string -> socket -> Raw_message.t -> (unit, error) result Lwt.t
-(** This function may fail with [Connection_closed]. *)
+val write_raw_message : magic_string:string -> socket -> Raw_message.t -> unit
+(** @raise Connection_closed *)
 
-val create_server : string -> socket Lwt_stream.t Lwt.t
+val create_server : string -> socket
+val accept : socket -> socket
diff --git a/lib/spatial_ipc/spatial_ipc.ml b/lib/spatial_ipc/spatial_ipc.ml
index 8d00d70..662582a 100644
--- a/lib/spatial_ipc/spatial_ipc.ml
+++ b/lib/spatial_ipc/spatial_ipc.ml
@@ -121,60 +121,38 @@ let of_raw_message (op, payload) =
   | 1l -> Some (Packed Get_windows)
   | _ -> None
 
-exception Spatial_ipc_error of Socket.error
-
 type socket = Socket.socket
 
-let connect () : socket Lwt.t = Socket.connect socket_path
+let connect () : socket = Socket.connect socket_path
 let close socket = Socket.close socket
 
-let trust_spatial f =
-  let open Lwt.Syntax in
-  let* x = f () in
-  match x with Ok x -> Lwt.return x | Error e -> raise (Spatial_ipc_error e)
-
-let with_socket f = Socket.with_socket socket_path f
-
-let socket_from_option = function
-  | Some socket -> Lwt.return socket
-  | None -> connect ()
+let with_socket ?socket f =
+  match socket with
+  | Some socket -> f socket
+  | None -> Socket.with_socket socket_path f
 
 let send_command ?socket cmd =
-  let open Lwt.Syntax in
-  let* s = socket_from_option socket in
+  with_socket ?socket @@ fun socket ->
   let ((op, _) as raw) = to_raw_message cmd in
-  let* () =
-    trust_spatial @@ fun () -> Socket.write_raw_message ~magic_string s raw
-  in
-  let* op', payload =
-    trust_spatial @@ fun () -> Socket.read_raw_message ~magic_string s
-  in
+  Socket.write_raw_message ~magic_string socket raw;
+  let op', payload = Socket.read_raw_message ~magic_string socket in
   assert (op = op');
-  let* () = if Option.is_none socket then close s else Lwt.return () in
-
-  Lwt.return @@ reply_of_string_exn cmd payload
+  reply_of_string_exn cmd payload
 
-type ('a, 'b) handler = { handler : 'r. 'a -> 'r t -> ('b option * 'r) Lwt.t }
+type ('a, 'b) handler = { handler : 'r. 'a -> 'r t -> 'b * 'r }
 
 let handle_next_command ~socket input { handler } =
-  let open Lwt.Syntax in
-  let* res = Socket.read_raw_message ~magic_string socket in
-  match res with
-  | Ok ((op, _) as raw) -> (
-      let cmd = of_raw_message raw in
-      match cmd with
-      | Some (Packed cmd) ->
-          let* output, reply = handler input cmd in
-          let* _ =
-            Socket.write_raw_message ~magic_string socket
-              (op, reply_to_string cmd reply)
-          in
-          Lwt.return output
-      | None ->
-          let* _ = Socket.write_raw_message ~magic_string socket (op, "") in
-          Lwt.return None)
-  | Error _ ->
-      let* _ = Socket.write_raw_message ~magic_string socket (-1l, "") in
-      Lwt.return None
+  let ((op, _) as raw) = Socket.read_raw_message ~magic_string socket in
+  let cmd = of_raw_message raw in
+  match cmd with
+  | Some (Packed cmd) ->
+      let output, reply = handler input cmd in
+      Socket.write_raw_message ~magic_string socket
+        (op, reply_to_string cmd reply);
+      Some output
+  | None ->
+      Socket.write_raw_message ~magic_string socket (op, "");
+      None
 
 let create_server () = Socket.create_server socket_path
+let accept = Socket.accept
diff --git a/lib/spatial_ipc/spatial_ipc.mli b/lib/spatial_ipc/spatial_ipc.mli
index c60d5e8..d742867 100644
--- a/lib/spatial_ipc/spatial_ipc.mli
+++ b/lib/spatial_ipc/spatial_ipc.mli
@@ -26,16 +26,15 @@ type 'a t =
   | Run_command : command -> run_command_reply t
   | Get_windows : get_windows_reply t
 
-type socket
+type socket = Unix.file_descr
 
-val connect : unit -> socket Lwt.t
-val close : socket -> unit Lwt.t
-val with_socket : (socket -> 'a Lwt.t) -> 'a Lwt.t
-val send_command : ?socket:socket -> 'a t -> 'a Lwt.t
+val connect : unit -> socket
+val close : socket -> unit
+val with_socket : ?socket:socket -> (socket -> 'a) -> 'a
+val send_command : ?socket:socket -> 'a t -> 'a
 
-type ('a, 'b) handler = { handler : 'r. 'a -> 'r t -> ('b option * 'r) Lwt.t }
+type ('a, 'b) handler = { handler : 'r. 'a -> 'r t -> 'b * 'r }
 
-val handle_next_command :
-  socket:socket -> 'u -> ('u, 'v) handler -> 'v option Lwt.t
-
-val create_server : unit -> socket Lwt_stream.t Lwt.t
+val handle_next_command : socket:socket -> 'u -> ('u, 'v) handler -> 'v option
+val create_server : unit -> socket
+val accept : socket -> socket
diff --git a/lib/sway_ipc/dune b/lib/sway_ipc/dune
index 86ccb22..5ebc71d 100644
--- a/lib/sway_ipc/dune
+++ b/lib/sway_ipc/dune
@@ -1,4 +1,4 @@
 (library
  (name sway_ipc)
  (public_name sway-ipc)
- (libraries json-decoder mltp-ipc sway-ipc.types lwt lwt.unix))
+ (libraries json-decoder mltp-ipc sway-ipc.types unix))
diff --git a/lib/sway_ipc/sway_ipc.ml b/lib/sway_ipc/sway_ipc.ml
index 13ec1af..73554c8 100644
--- a/lib/sway_ipc/sway_ipc.ml
+++ b/lib/sway_ipc/sway_ipc.ml
@@ -5,8 +5,6 @@
 open Sway_ipc_types
 open Mltp_ipc
 
-exception Sway_ipc_error of Socket.error
-
 let magic_string = "i3-ipc"
 
 let sway_sock_path () =
@@ -16,54 +14,36 @@ let sway_sock_path () =
 
 type socket = Socket.socket
 
-let connect () : socket Lwt.t = Socket.connect (sway_sock_path ())
+let connect () = Socket.connect (sway_sock_path ())
 let close socket = Socket.close socket
 
-let trust_sway f =
-  let open Lwt.Syntax in
-  let* x = f () in
-  match x with Ok x -> Lwt.return x | Error e -> raise (Sway_ipc_error e)
-
-let with_socket f = Socket.with_socket (sway_sock_path ()) f
+let with_socket ?socket f =
+  match socket with
+  | Some socket -> f socket
+  | None -> Socket.with_socket (sway_sock_path ()) f
 
-let socket_from_option = function
-  | Some socket -> Lwt.return socket
-  | None -> connect ()
-
-let send_command ?socket cmd =
-  let open Lwt.Syntax in
-  let* socket = socket_from_option socket in
+let send_command : type a. ?socket:socket -> a Message.t -> a =
+ fun ?socket cmd ->
+  with_socket ?socket @@ fun socket ->
   let ((op, _) as raw) = Message.to_raw_message cmd in
-  let* () =
-    trust_sway @@ fun () -> Socket.write_raw_message ~magic_string socket raw
-  in
-  let* op', payload =
-    trust_sway @@ fun () -> Socket.read_raw_message ~magic_string socket
-  in
+  Socket.write_raw_message ~magic_string socket raw;
+  let op', payload = Socket.read_raw_message ~magic_string socket in
   assert (op = op');
-  Lwt.return @@ Json_decoder.of_string_exn (Message.reply_decoder cmd) payload
+  Json_decoder.of_string_exn (Message.reply_decoder cmd) payload
+
+let subscribe events =
+  let socket = connect () in
+  let ({ success } : Message.subscribe_reply) =
+    send_command ~socket (Subscribe events)
+  in
+  if success then socket else failwith "could not subscribe"
 
-let subscribe ?socket events =
-  let open Lwt.Syntax in
-  let* socket = socket_from_option socket in
-  let+ { success } = send_command ~socket (Subscribe events) in
-  if success then
-    Lwt_stream.from (fun () ->
-        let open Lwt.Syntax in
-        let+ ev =
-          Socket.read_next_raw_message ~magic_string socket (fun (code, _) ->
-              List.exists
-                (fun ev_type -> ev_type = Event.event_type_of_code code)
-                events)
-        in
-        match ev with
-        | Ok ev -> Some (Event.event_of_raw_message ev)
-        | Error _ -> None)
-  else failwith "Something went wrong"
+let read_event socket =
+  let raw = Socket.read_next_raw_message ~magic_string socket (fun _ -> true) in
+  Event.event_of_raw_message raw
 
 let get_tree ?socket () = send_command ?socket Get_tree
 
 let get_current_workspace ?socket () =
-  let open Lwt.Syntax in
-  let+ workspaces = send_command ?socket Get_workspaces in
+  let workspaces = send_command ?socket Get_workspaces in
   List.find (fun w -> w.Workspace.focused) workspaces
diff --git a/lib/sway_ipc/sway_ipc.mli b/lib/sway_ipc/sway_ipc.mli
index 753a0df..1c04b69 100644
--- a/lib/sway_ipc/sway_ipc.mli
+++ b/lib/sway_ipc/sway_ipc.mli
@@ -2,27 +2,25 @@
  * License, v. 2.0. If a copy of the MPL was not distributed with this
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. *)
 
-type socket
+type socket = Unix.file_descr
 (** A socket to interact with Sway. *)
 
-exception Sway_ipc_error of Mltp_ipc.Socket.error
-
-val connect : unit -> socket Lwt.t
+val connect : unit -> socket
 (** [connect ()] establishes a connection with Sway. This connection
     can be ended by using {!close}.
 
     When possible, it is advised to use {!with_socket}. *)
 
-val close : socket -> unit Lwt.t
+val close : socket -> unit
 (** [close socket] puts an end to a connection with Sway. *)
 
-val with_socket : (socket -> 'a Lwt.t) -> 'a Lwt.t
-(** [with_socket k] establishes a bi-connection with Sway, hands over
-    the socket to the continuation [k], and takes care of closing the
-    connection prior to returning the result, even in case of an
-    exception. *)
+val with_socket : ?socket:socket -> (socket -> 'a) -> 'a
+(** [with_socket ?socket k] establishes a bi-connection with Sway,
+    hands over the socket to the continuation [k], and takes care of
+    closing the connection prior to returning the result, even in case
+    of an exception. *)
 
-val send_command : ?socket:socket -> 'a Sway_ipc_types.Message.t -> 'a Lwt.t
+val send_command : ?socket:socket -> 'a Sway_ipc_types.Message.t -> 'a
 (** [send_command ?socket msg] sends the command [msg] to Sway (by
     establishing (either by using [socket], or by establishing a fresh
     connection if [socket] is omitted), and returns the result sent
@@ -31,25 +29,23 @@ val send_command : ?socket:socket -> 'a Sway_ipc_types.Message.t -> 'a Lwt.t
     This is a low-level helpers. It is advised to use specialized
     helpers whenever they are available. *)
 
-val subscribe :
-  ?socket:socket ->
-  Sway_ipc_types.Event.event_type list ->
-  Sway_ipc_types.Event.t Lwt_stream.t Lwt.t
+val read_event : socket -> Sway_ipc_types.Event.t
+
+val subscribe : Sway_ipc_types.Event.event_type list -> socket
 (** [subscribe ?socket evs] returns a stream of events sent by Sway,
     matching the event types listed in [evs].
 
     The socket passed as argument should not be used to send commands
     afterwards. *)
 
-val get_tree : ?socket:socket -> unit -> Sway_ipc_types.Node.t Lwt.t
+val get_tree : ?socket:socket -> unit -> Sway_ipc_types.Node.t
 (** [get_tree ?socket ()] returns the current state of the tree
     manipulated by Sway.
 
     If [socket] is omitted, a fresh connection is established with
     Sway. *)
 
-val get_current_workspace :
-  ?socket:socket -> unit -> Sway_ipc_types.Workspace.t Lwt.t
+val get_current_workspace : ?socket:socket -> unit -> Sway_ipc_types.Workspace.t
 (** [get_current_workspace ?socket ()] returns the workspace currently
     focused by Sway.