Queue up observe messages

Regular data messages are queued when there are less than 100 messages
for that client. Build finish messages are always enqueued. This ensures
end messages are never lost, and enhances the user experience for the
client as a busy build will drop less messages.

app/server.ml

@@ -562,16 +562,38 @@ let client_loop t fd =
         Lwt_list.iter_s (fun (ts, l) ->
             write_cmd fd (output id ts l) >|= ignore)
           (List.rev out) >>= fun () ->
+        let q = Queue.create () in
+        let q_cond = Lwt_condition.create () in
         let rec more () =
           Lwt_condition.wait cond >>= function
-          | `End (ts, data) ->
-            write_cmd fd (output id ts data)
-          | `Data (ts, data) ->
-            write_cmd fd (output id ts data) >>= function
-            | Ok () -> more ()
-            | Error _ -> Lwt.return (Ok ())
+          | `End _ as ev ->
+            Queue.add ev q;
+            Lwt_condition.signal q_cond ();
+            Lwt.return_unit
+          | `Data _ as ev ->
+            if Queue.length q < 100 then begin
+              Queue.add ev q;
+              Lwt_condition.signal q_cond ();
+              more ()
+            end else
+              (* Drop data messages if the client can't keep up *)
+              Lwt.return_unit
         in
-        more ()
+        let rec send () =
+          if Queue.is_empty q then
+            Lwt_condition.wait q_cond >>= fun () ->
+            send ()
+          else
+            match Queue.take q with
+            | `End (ts, data) ->
+              write_cmd fd (output id ts data)
+            | `Data (ts, data) ->
+              write_cmd fd (output id ts data) >>= function
+              | Ok () -> send ()
+              | Error _ -> Lwt.return (Ok ())
+        in
+        let more = more () in
+        send () >>= fun r -> Lwt.cancel more; Lwt.return r
       | None -> Lwt.return (Error (`Msg "uuid not found"))
     end
   | Builder.Drop_platform p ->