implemented a basic tcp server and tests

halgari · halgari · commit 0cccf33e7dda · 2015-04-16T15:43:32.000-06:00
diff --git a/pixie/io.pxi b/pixie/io.pxi
@@ -111,6 +111,11 @@
   IDisposable
   (-dispose! [this]
     (set-buffer-count! buffer idx)
+    (write downstream buffer))
+  IFlushableStream
+  (flush [this]
+    (set-buffer-count! buffer idx)
+    (set-field! this :idx 0)
     (write downstream buffer)))
 
 (deftype BufferedInputStream [upstream idx buffer]
@@ -124,16 +129,21 @@
       val))
   IDisposable
   (-dispose! [this]
-    (dispose! upstream)
     (dispose! buffer)))
 
-(defn buffered-output-stream [downstream size]
-  (->BufferedOutputStream downstream 0 (buffer size)))
+(defn buffered-output-stream
+  ([downstream]
+   (buffered-output-stream downstream DEFAULT-BUFFER-SIZE))
+  ([downstream size]
+    (->BufferedOutputStream downstream 0 (buffer size))))
 
-(defn buffered-input-stream [upstream size]
-  (let [b (buffer size)]
-    (set-buffer-count! b size)
-    (->BufferedInputStream upstream size b)))
+(defn buffered-input-stream
+  ([upstream]
+   (buffered-input-stream upstream DEFAULT-BUFFER-SIZE))
+  ([upstream size]
+   (let [b (buffer size)]
+     (set-buffer-count! b size)
+     (->BufferedInputStream upstream size b))))
 
 (defn throw-on-error [result]
   (when (neg? result)
@@ -180,25 +190,6 @@
 (defn run-command [command]
   (st/apply-blocking io-blocking/run-command command))
 
-(comment
-
-  (defn tcp-server [ip port on-connection]
-    (assert (string? ip) "Ip should be a string")
-    (assert (integer? port) "Port should be a int")
-    (let [server (uv/uv_tcp_t)
-          bind-addr (uv/sockaddr_in)
-          _ (uv/throw-on-error (uv/uv_ip4_addr ip port bind-addr))
-          on-new-connetion (atom nil)]
-      (reset! on-new-connetion
-              (ffi/ffi-prep-callback
-               uv/uv_connection_cb
-               (fn [server status]
-                 (when (not (= status -1))
-                   (println "Got Client!!!!!!!")))))
-      (uv/uv_tcp_init (uv/uv_default_loop) server)
-      (uv/uv_tcp_bind server bind-addr 0)
-      (uv/throw-on-error (uv/uv_listen server 128 @on-new-connetion))
-      (st/yield-control))))
 
 (comment
   (st/apply-blocking println "FROM OTHER THREAD <---!!!!!")
diff --git a/pixie/io/tcp.pxi b/pixie/io/tcp.pxi
@@ -4,15 +4,19 @@
             [pixie.uv :as uv]
             [pixie.ffi :as ffi]))
 
-(defrecord TCPServer [ip port on-connect uv-server bind-addr on-connection-cb])
+(defrecord TCPServer [ip port on-connect uv-server bind-addr on-connection-cb]
+  IDisposable
+  (-dispose! [this]
+    (uv/uv_close uv-server st/close_cb)
+    (dispose! @on-connection-cb)
+    (dispose! bind-addr)))
 
 (defn -prep-uv-buffer-fn [buf read-bytes]
   (ffi/ffi-prep-callback
    uv/uv_alloc_cb
    (fn [handle suggested-size uv-buf]
      (try
        (let [casted (ffi/cast uv-buf uv/uv_buf_t)]
-         (println "Alloc " handle suggested-size buf read-bytes)
          (ffi/set! casted :base buf)
          (ffi/set! casted :len (min suggested-size
                                     (buffer-capacity buf)
@@ -30,37 +34,40 @@
                  (reset! read-cb (ffi/ffi-prep-callback
                                   uv/uv_read_cb
                                   (fn [stream nread uv-buf]
-                                    (println "-<<< nread <-- " nread)
                                     (set-buffer-count! buffer nread)
                                     (try
                                       (dispose! alloc-cb)
                                       (dispose! @read-cb)
                                       ;(dispose! uv-buf)
                                       (uv/uv_read_stop stream)
-                                      (st/run-and-process k nread)
+                                      (st/run-and-process k (or
+                                                             (st/exception-on-uv-error nread)
+                                                             nread))
                                       (catch ex
                                           (println ex))))))
                  (uv/uv_read_start uv-client alloc-cb @read-cb)))))
   IOutputStream
   (write [this buffer]
     (let [write-cb (atom nil)
           uv_write (uv/uv_write_t)]
-      (println "writing " (count buffer))
       (ffi/set! uv-write-buf :base buffer)
       (ffi/set! uv-write-buf :len (count buffer))
       (st/call-cc
        (fn [k]
          (reset! write-cb (ffi/ffi-prep-callback
                           uv/uv_write_cb
                           (fn [req status]
-                            (println status "<<-- status")
                             (try
                               (dispose! @write-cb)
                               ;(uv/uv_close uv_write st/close_cb)
                               (st/run-and-process k status)
                               (catch ex
                                   (println ex))))))
-         (uv/uv_write uv_write uv-client uv-write-buf 1 @write-cb))))))
+         (uv/uv_write uv_write uv-client uv-write-buf 1 @write-cb)))))
+  IDisposable
+  (-dispose! [this]
+    (dispose! uv-write-buf)
+    (uv/uv_close uv-client st/close_cb)))
 
 (defn launch-tcp-client-from-server [svr]
   (assert (instance? TCPServer svr) "Requires a TCPServer as the first argument")
@@ -94,3 +101,27 @@
       (uv/throw-on-error (uv/uv_listen server 128 @on-new-connetion))
       (st/yield-control)
       tcp-server))
+
+
+(defn tcp-client [ip port]
+  (let [client-addr (uv/sockaddr_in)
+        uv-connect (uv/uv_connect_t)
+        client (uv/uv_tcp_t)
+        cb (atom nil)]
+    (uv/throw-on-error (uv/uv_ip4_addr ip port client-addr))
+    (uv/uv_tcp_init (uv/uv_default_loop) client)
+    (st/call-cc (fn [k]
+               (reset! cb (ffi/ffi-prep-callback
+                           uv/uv_connect_cb
+                           (fn [_ status]
+                             (try
+                               (dispose! @cb)
+                               (dispose! uv-connect)
+                               (dispose! client-addr)
+                               (st/run-and-process k (or (st/exception-on-uv-error status)
+                                                         (->TCPStream client (uv/uv_buf_t))))
+                               (catch ex
+                                   (println ex))))))
+               (uv/uv_tcp_connect uv-connect client client-addr @cb)))
+
+    ))
diff --git a/pixie/stacklets.pxi b/pixie/stacklets.pxi
@@ -16,19 +16,28 @@
 
 ;; Yield
 
+(defrecord ThrowException [ex])
+
 (defn run-and-process
   ([k]
    (run-and-process k nil))
   ([k val]
    (let [[h f] (k val)]
      (f h))))
 
+(defn exception-on-uv-error [result]
+  (when (neg? result)
+    (->ThrowException (str "UV Error: " (uv/uv_err_name result)))))
+
+
 (defn call-cc [f]
   (let [frames (-get-current-var-frames nil)
         [h val] (@stacklet-loop-h f)]
     (reset! stacklet-loop-h h)
     (-set-current-var-frames nil frames)
-    val))
+    (if (instance? ThrowException val)
+      (throw (:ex val))
+      val)))
 
 (defn -run-later [f]
   (let [a (uv/uv_async_t)
diff --git a/pixie/stdlib.pxi b/pixie/stdlib.pxi
@@ -2324,4 +2324,4 @@ Calling this function on something that is not ISeqable returns a seq with that
   "Returns the largest of all the arguments to this function. Assumes arguments are numeric"
   ([x] x)
   ([x y] (if (> x y) x y))
-  ([x y & zs] (apply min (min x y) zs)))
+  ([x y & zs] (apply max (max x y) zs)))
diff --git a/pixie/streams.pxi b/pixie/streams.pxi
@@ -1,5 +1,8 @@
 (ns pixie.streams)
 
+(defprotocol IFlushableStream
+  (flush [this] "Flushes all buffers in this stream and applies writes to any parent streams"))
+
 (defprotocol IInputStream
   (read [this buffer len] "Reads multiple bytes into a buffer, returns the number of bytes read"))
 
diff --git a/pixie/uv.pxi b/pixie/uv.pxi
@@ -124,6 +124,8 @@
     (f/defconst UV_DIRENT_CHAR)
     (f/defconst UV_DIRENT_BLOCK)
 
+    (f/defconst UV_EOF)
+
     (f/defcstruct uv_dirent_t [:name
                                :type])
 
diff --git a/tests/pixie/tests/io/test-tcp.pxi b/tests/pixie/tests/io/test-tcp.pxi
@@ -1,15 +1,40 @@
 (ns pixie.test.io.test-tcp
   (:require [pixie.io.tcp :refer :all]
-            [pixie.io :refer [read write]]
+            [pixie.io :refer [buffered-input-stream buffered-output-stream read-byte write-byte]]
+            [pixie.streams :refer :all]
             [pixie.stacklets :as st]
-            [pixie.async :as async]))
+            [pixie.async :as async]
+            [pixie.uv :as uv]
+            [pixie.test :refer :all]))
 
-(defn on-client [conn]
-  (let [b (buffer 1024)]
-    (read conn b 1024)
-    (write conn b)
-    (dotimes [x 1000]
-      (st/yield-control))
-    (println "Done Writing..")))
+(deftest test-echo-server
+  (let [client-done (async/promise)
+        on-client (fn on-client [conn]
+                    (let [in (buffered-input-stream conn)
+                          out (buffered-output-stream conn)]
+                      (try
+                        (loop []
+                          (let [val (read-byte in)]
+                            (write-byte out val)
+                            (flush out)
+                            (recur)))
+                        (catch ex
+                            (dispose! in)
+                          (dispose! out)
 
-(tcp-server "0.0.0.0" 4242 on-client)
+                          (dispose! conn)
+                          (client-done true)))))
+
+        server (tcp-server "0.0.0.0" 4242 on-client)]
+
+    (let [client-stream (tcp-client "127.0.0.1" 4242)
+          in (buffered-input-stream client-stream)
+          out (buffered-output-stream client-stream)]
+
+      (dotimes [x 255]
+        (write-byte out x)
+        (flush out)
+        (assert= x (read-byte in)))
+      (dispose! client-stream)
+      (assert @client-done)
+      (dispose! server))))
