More sophisticated socket handling.

Now we can handle non-blocking socket operations.

Author: conormcd

resources/darwin/libsimplesocket.dylib

@@ -7,4 +7,4 @@
 (def config (jna/to-fn Void ssh2/libssh2_keepalive_config))
 
 ; int libssh2_keepalive_send (LIBSSH2_SESSION *session, int *seconds_to_next);
-(def send (jna/to-fn Void ssh2/libssh2_keepalive_send))
+(def send (jna/to-fn Integer ssh2/libssh2_keepalive_send))

resources/linux-x86-64/libsimplesocket.so

@@ -90,6 +90,7 @@
          (when (> 0 (:socket session))
            (destroy-session session "Shutting down due to bad socket." true))
          (try
+           (libssh2-session/set-blocking (:session session) 0)
            (handshake session)
            (known-hosts/check session)
            (authenticate session credentials)
@@ -98,3 +99,7 @@
            (catch Exception e
              (close session)
              (throw e))))))))
+
+(defn get-timeout
+  [session]
+  (libssh2-session/get-timeout (:session session)))

src/clj_libssh2/libssh2/keepalive.clj

@@ -1,5 +1,20 @@
 (ns clj-libssh2.socket
-  (:require [net.n01se.clojure-jna :as jna]))
+  (:require [net.n01se.clojure-jna :as jna]
+            [clj-libssh2.error :refer [handle-errors]]
+            [clj-libssh2.libssh2 :as libssh2]
+            [clj-libssh2.libssh2.keepalive :as libssh2-keepalive]
+            [clj-libssh2.libssh2.session :as libssh2-session])
+  (:import [com.sun.jna.ptr IntByReference]))
+
+;; Constants from libsimplesocket.h
+
+(def SIMPLE_SOCKET_BAD_ADDRESS -1)
+(def SIMPLE_SOCKET_SOCKET_FAILED -2)
+(def SIMPLE_SOCKET_CONNECT_FAILED -3)
+
+(def SIMPLE_SOCKET_SELECT_READ 1)
+(def SIMPLE_SOCKET_SELECT_WRITE 2)
+(def SIMPLE_SOCKET_SELECT_ERROR 4)
 
 (def close
   "Close a socket"
@@ -14,9 +29,84 @@
                            port)]
     (when (> 0 socket)
       ;; Magic numbers are from libsimplesocket.h
-      (throw (Exception. (case socket
-                           -1 (format "%s is not a valid IP address" address)
-                           -2 "Failed to create a TCP socket"
-                           -3 (format "Failed to connect to %s:%d" address port)
+      (throw (Exception. (condp = socket
+                           SIMPLE_SOCKET_BAD_ADDRESS
+                           (format "%s is not a valid IP address" address)
+
+                           SIMPLE_SOCKET_SOCKET_FAILED
+                           "Failed to create a TCP socket"
+
+                           SIMPLE_SOCKET_CONNECT_FAILED
+                           (format "Failed to connect to %s:%d" address port)
+
                            "An unknown error occurred"))))
     socket))
+
+(defn select
+  "Call select on a socket from a clj-libssh2 Session."
+  [session select-read select-write timeout]
+  (jna/invoke Integer
+              simplesocket/simple_socket_select
+              (:socket session)
+              (bit-or
+                (if select-read SIMPLE_SOCKET_SELECT_READ 0)
+                (if select-write SIMPLE_SOCKET_SELECT_WRITE 0))
+              timeout))
+
+(defn send-keepalive
+  "Send a keepalive message and return the number of seconds until the next
+   time we should send a keepalive."
+  [session]
+  (let [seconds-to-wait (IntByReference.)]
+    (handle-errors session
+      (libssh2-keepalive/send (:session session) seconds-to-wait))
+    (.getValue seconds-to-wait)))
+
+(defn wait
+  "Roughly equivalent to _libssh2_wait_socket in libssh2. Will raise an error
+  on timeout or just block until it's time to try again."
+  ([session]
+   (wait session (System/currentTimeMillis)))
+  ([session start-time]
+   (let [ms-until-next-keepalive (* (send-keepalive session) 1000)
+         block-directions (libssh2-session/block-directions (:session session))
+         block-for-read (boolean (bit-and block-directions libssh2/SESSION_BLOCK_INBOUND))
+         block-for-write (boolean (bit-and block-directions libssh2/SESSION_BLOCK_OUTBOUND))
+         libssh2-timeout-ms (libssh2-session/get-timeout (:session session))
+         select-until (partial select session block-for-read block-for-write)
+         select-result (cond (and (< 0 libssh2-timeout-ms)
+                                  (or (= 0 ms-until-next-keepalive)
+                                      (> ms-until-next-keepalive libssh2-timeout-ms)))
+                             (let [elapsed (- (System/currentTimeMillis) start-time)]
+                               (when (> elapsed libssh2-timeout-ms)
+                                 (handle-errors session libssh2/ERROR_TIMEOUT))
+                               (select-until (- libssh2-timeout-ms elapsed)))
+
+                             (< 0 ms-until-next-keepalive)
+                             (select-until ms-until-next-keepalive)
+
+                             :else
+                             (select-until 0))]
+     (when (>= 0 select-result)
+       (handle-errors session libssh2/ERROR_TIMEOUT)))))
+
+(defmacro block
+  "Turn a non-blocking call that returns EAGAIN into a blocking one."
+  [session & body]
+  `(let [start-time# (-> (System/currentTimeMillis) (/ 1000.0) Math/round)]
+     (while (= libssh2/ERROR_EAGAIN (do ~@body))
+       (handle-errors ~session
+         (wait ~session start-time#)))))
+
+(defmacro block-return
+  "Similar to block, but for functions that return a pointer"
+  [session & body]
+  `(let [start-time# (-> (System/currentTimeMillis) (/ 1000.0) Math/round)]
+     (loop [result# (do ~@body)]
+       (if (nil? result#)
+         (let [errno# (libssh2-session/last-errno (:session ~session))]
+           (handle-errors ~session errno#)
+           (when (= libssh2/ERROR_EAGAIN errno#)
+             (wait ~session start-time#))
+           (recur (do ~@body)))
+         result#))))