TCP seems to be working now!

2 files changed, 43 insertions, 16 deletions

diff --git a/src/org/gnu/clojure/gnunet/tcp.clj b/src/org/gnu/clojure/gnunet/tcp.clj
index 672feb9..e261688 100644
--- a/src/org/gnu/clojure/gnunet/tcp.clj
+++ b/src/org/gnu/clojure/gnunet/tcp.clj
@@ -59,23 +59,31 @@
       [:when-let [welcome (first (parse-welcome (:bytes message)))]
        connection (fetch-val [:connection encoded-address])
        :when-not (nil? connection)
+       ;; XXX: This is weird because for an outgoing connection we don't check
+       ;; that the peer-id matches who we thought we connected to.
        _ (set-val [:connection encoded-address]
-           (assoc connection :remote-peer-id (:peer-id welcome)))]
+           (conj connection {:remote-peer-id (:peer-id welcome)
+                             :received-welcome true}))]
       (do
-        (when (and (nil? (:remote-peer-id connection))
-                (:incoming connection))
+        (when (nil? (:remote-peer-id connection))  
           (.add (:send-queue connection)
             {:bytes (generate-welcome-message peer)
              :continuation! identity}))
-        (update-selection-key-async! peer selection-key
-          (bit-or SelectionKey/OP_READ SelectionKey/OP_WRITE))))
-    (when-let [remote-peer-id (:remote-peer-id
-                                ((deref (:sessions-agent transport))
-                                  [:connection encoded-address]))]
-      (let [address {:transport "tcp"
-                     :encoded-address encoded-address
-                     :expiration (idle-connection-timeout)}]
-        (admit-message! peer remote-peer-id address message)))))
+        ;; We have to send again to make sure our update to the sessions-agent
+        ;; is finished before updating the selection-key
+        (send (:sessions-agent transport)
+          (fn [sessions]
+            (update-selection-key-async! peer selection-key
+              (bit-or SelectionKey/OP_READ SelectionKey/OP_WRITE))
+            sessions))))
+    (send-do-exception-m! (:sessions-agent transport)
+      [connection (fetch-val [:connection encoded-address])
+       :when-let [remote-peer-id (:remote-peer-id connection)]
+       :when (:received-welcome connection)
+       :let [address {:transport "tcp"
+                      :encoded-address encoded-address
+                      :expiration (idle-connection-timeout)}]]
+      (admit-message! peer remote-peer-id address message))))
 
 (defn handle-socket-channel-connectable!
   [peer transport encoded-address selection-key]
@@ -178,6 +186,7 @@
                    {:socket-channel socket-channel
                     :selection-key selection-key
                     :send-queue send-queue
+                    :remote-peer-id (:id remote-peer)
                     :received-bytes []})]
               (do
                 (.connect socket-channel address)
diff --git a/src/org/gnu/clojure/gnunet/transport.clj b/src/org/gnu/clojure/gnunet/transport.clj
index ac1f923..127fc80 100644
--- a/src/org/gnu/clojure/gnunet/transport.clj
+++ b/src/org/gnu/clojure/gnunet/transport.clj
@@ -45,7 +45,7 @@
 
 (defn encode-pong-signed-material
   [pong]
-  (let [transport (encode-utf8 (:transport pong))
+  (let [transport (when (:transport pong) (encode-utf8 (:transport pong)))
         address-length (+ (count transport) (count (:encoded-address pong)))]
     (encode-signed (:signature-purpose pong)
       (concat
@@ -252,13 +252,31 @@
 
 (defn send-pong-using!
   [peer remote-peer ping]
-  (.write *out* "We don't send PONG_USING yet!\n"))
+  (.execute (:cpu-bound-executor peer)
+    (fn []
+      (let [pong {:challenge (:challenge ping)
+                  :signature-purpose signature-purpose-pong-using
+                  :expiration (pong-expiration)
+                  :peer-id (:id peer)}
+            signed-material (encode-pong-signed-material pong)
+            signature (rsa-sign (:private-key peer) signed-material)
+            pong (assoc pong :signature signature)
+            encoded-pong (encode-pong pong signed-material)]
+        ;; XXX: gnunet looks for a "reliable" connection for the pong before
+        ;; sending to every known address.
+        (doseq [transports (deref (:transport-addresses-agent remote-peer))
+                address (val transports)]
+          (when-let [transport ((deref (:transports-agent peer))
+                                 (key transports))]
+            ((:emit-messages! transport) transport remote-peer (key address)
+              nil
+              [{:message-type message-type-pong :bytes encoded-pong}])))))))
 
 (defn handle-ping!
   [peer remote-peer message]
   (when-let [ping (first (parse-ping (:bytes message)))]
     (cond
-      (not (= (:peer-id ping) (seq (:id peer)))) nil
+      (not (= (:peer-id ping) (:id peer))) nil
       (:transport ping) (send-pong-own! peer remote-peer ping)
       :else (send-pong-using! peer remote-peer ping))))
 
@@ -299,7 +317,7 @@
 
 (defn emit-continuation!
   [peer transport remote-peer encoded-address result]
-  (if result
+  (when result
     (let [addresses ((deref (:transport-addresses-agent remote-peer))
                       (:name transport))
           address (addresses encoded-address)]