The goal of the patch is to enable the use of non-blocking

connections, to make async integration easier.  To that end, we made
the following changes:

- For putline, putnbytes and endcopy, don't throw exceptions on
  non-blocking connections.

- Return meaningful status code from flush

Also, we added a binding for the server_version call.

Written by Max Wolter

Author: yminsky-janestreet

lib/postgresql.ml

@@ -254,6 +254,10 @@ type polling_status =
   | Polling_writing
   | Polling_ok
 
+type flush_status =
+  | Successful
+  | Data_left_to_send
+
 type conninfo_option =
   {
     cio_keyword : string;
@@ -340,7 +344,7 @@ module Stub = struct
 
   external error_message : connection -> string = "PQerrorMessage_stub"
   external backend_pid : connection -> int = "PQbackendPID_stub" "noalloc"
-
+  external server_version : connection -> int = "PQserverVersion_stub" "noalloc"
 
   (* Command Execution Functions *)
 
@@ -773,6 +777,20 @@ object (self)
   method status = wrap_conn Stub.connection_status
   method error_message = wrap_conn Stub.error_message
   method backend_pid = wrap_conn Stub.backend_pid
+  method server_version = wrap_conn (fun conn ->
+    let v = Stub.server_version conn in
+    if v = 0 then begin
+      let message = if Stub.connection_status conn = Bad
+        then "server_version failed because the connection was bad"
+        else "server_version failed for an unknown reason"
+      in
+      raise (Error (Connection_failure message))
+    end;
+    let revision = v mod 100 in
+    let minor    = (v / 100) mod 100 in
+    let major    = v / (100 * 100) in
+    major, minor, revision
+  )
 
 
   (* Commands and Queries *)
@@ -882,18 +900,26 @@ object (self)
       | _ -> assert false)
 
   method putline buf =
-    wrap_conn (fun conn -> if Stub.putline conn buf <> 0 then signal_error conn)
+    wrap_conn (fun conn ->
+      if (Stub.putline conn buf <> 0) && not (Stub.is_nonblocking conn) then
+        signal_error conn
+    )
 
   method putnbytes ?(pos = 0) ?len buf =
     let buf_len = String.length buf in
     let len = match len with Some len -> len | None -> buf_len - pos in
     if len < 0 || pos < 0 || pos + len > buf_len then
       invalid_arg "Postgresql.connection#putnbytes";
     wrap_conn (fun conn ->
-      if Stub.putnbytes conn buf pos len <> 0 then signal_error conn)
+      if (Stub.putnbytes conn buf pos len <> 0) && not (Stub.is_nonblocking conn) then
+        signal_error conn
+    )
 
   method endcopy =
-    wrap_conn (fun conn -> if Stub.endcopy conn <> 0 then signal_error conn)
+    wrap_conn (fun conn ->
+      if (Stub.endcopy conn <> 0) && not (Stub.is_nonblocking conn) then
+        signal_error conn
+    )
 
 
   (* High level *)
@@ -948,7 +974,12 @@ object (self)
   method is_busy = wrap_conn Stub.is_busy
 
   method flush =
-    wrap_conn (fun conn -> if Stub.flush conn <> 0 then signal_error conn)
+    wrap_conn (fun conn ->
+      match Stub.flush conn with
+      | 0 -> Successful
+      | 1 -> Data_left_to_send
+      | _ -> signal_error conn
+    )
 
   method socket =
     wrap_conn (fun conn ->

lib/postgresql.mli

@@ -380,6 +380,11 @@ type polling_status =
   | Polling_writing
   | Polling_ok
 
+(** Result of a call to flush on nonblocking connections. *)
+type flush_status =
+  | Successful
+  | Data_left_to_send
+
 (** Record of connection options *)
 type conninfo_option =
   {
@@ -526,6 +531,7 @@ object
       @raise Error if there is a connection error.
   *)
 
+  method server_version : int * int * int (* major, minor, revision *)
 
   (** Commands and Queries *)
 
@@ -801,7 +807,7 @@ object
       @raise Error if there is a connection error.
   *)
 
-  method flush : unit
+  method flush : flush_status
   (** [flush] attempts to flush any data queued to the backend.
 
       @raise Error if there is a connection error.

lib/postgresql_stubs.c

@@ -369,7 +369,7 @@ conn_info(PQoptions, make_string)
 noalloc_conn_info(PQstatus, Val_int)
 conn_info(PQerrorMessage, make_string)
 noalloc_conn_info(PQbackendPID, Val_int)
-
+noalloc_conn_info(PQserverVersion, Val_int)
 
 /* Command Execution Functions */