diff options
| -rw-r--r-- | NOTES | 2 | ||||
| -rw-r--r-- | src/org/gnu/clojure/gnunet/inet.clj | 6 | ||||
| -rw-r--r-- | src/org/gnu/clojure/gnunet/tcp.clj | 12 | ||||
| -rw-r--r-- | src/org/gnu/clojure/gnunet/transport.clj | 80 | ||||
| -rw-r--r-- | src/org/gnu/clojure/gnunet/util.clj | 8 |
5 files changed, 70 insertions, 38 deletions
@@ -6,4 +6,4 @@ (activate-tcp! my-peer 5678) (configure-inet-addresses! my-peer "tcp" (get-local-addresses) 5678) (activate-filesharing! my-peer) -(download-hostlist! (partial admit-hello! my-peer) "http://192.168.8.2:58080") +(download-hostlist! (partial admit-hello! my-peer) "http://localhost:58080") diff --git a/src/org/gnu/clojure/gnunet/inet.clj b/src/org/gnu/clojure/gnunet/inet.clj index 376f73d..3ab6851 100644 --- a/src/org/gnu/clojure/gnunet/inet.clj +++ b/src/org/gnu/clojure/gnunet/inet.clj @@ -6,9 +6,9 @@ (defn encode-address [inet-socket-address] - (concat - (.getAddress (.getAddress inet-socket-address)) - (encode-int16 (.getPort inet-socket-address)))) + (vec (concat + (.getAddress (.getAddress inet-socket-address)) + (encode-int16 (.getPort inet-socket-address))))) (def parse-address (match-one diff --git a/src/org/gnu/clojure/gnunet/tcp.clj b/src/org/gnu/clojure/gnunet/tcp.clj index e261688..fac38e4 100644 --- a/src/org/gnu/clojure/gnunet/tcp.clj +++ b/src/org/gnu/clojure/gnunet/tcp.clj @@ -27,6 +27,8 @@ (defn handle-disconnect! [peer transport encoded-address selection-key] ;; This is always called from inside the selector thread + (assert-args handle-disconnect! + (vector? encoded-address) "encoded-address as vector") (.cancel selection-key) (.close (.channel selection-key)) (send-do-exception-m! (:sessions-agent transport) @@ -54,6 +56,8 @@ (defn admit-tcp-message! [peer transport encoded-address selection-key message] + (assert-args admit-tcp-message! + (vector? encoded-address) "encoded-address as vector") (if (== message-type-tcp-welcome (:message-type message)) (send-do-exception-m! (:sessions-agent transport) [:when-let [welcome (first (parse-welcome (:bytes message)))] @@ -93,6 +97,8 @@ (defn handle-socket-channel-readable! [peer transport encoded-address selection-key] + (assert-args handle-socket-channel-readable! + (vector? encoded-address) "encoded-address as vector") (let [socket-channel (.channel selection-key) socket (.socket socket-channel) buffer-length (.getReceiveBufferSize socket) @@ -120,6 +126,8 @@ ;; selector-continuations-queue because we want to make sure we set the ;; interest ops on the selection-key after any other continuations that might ;; be setting OP_WRITE. + (assert-args handle-socket-channel-writable! + (vector? encoded-address) "encoded-address as vector") (.add (:selector-continuations-queue peer) #(let [sessions (deref (:sessions-agent transport))] (when-let [connection (sessions [:connection encoded-address])] @@ -164,6 +172,8 @@ (defn set-connection-writable-or-connect! [peer transport remote-peer encoded-address] + (assert-args set-connection-writable-or-connect! + (vector? encoded-address) "encoded-address as vector") (if-let [connection ((deref (:sessions-agent transport)) [:connection encoded-address])] (set-connection-writable! peer remote-peer connection) @@ -201,6 +211,8 @@ (defn emit-messages-tcp! [peer transport remote-peer encoded-address continuation! messages] + (assert-args emit-tcp-message! + (vector? encoded-address) "encoded-address as vector") (send-do-exception-m! (:sessions-agent transport) [:let [continuation! #(do (emit-continuation! peer transport remote-peer encoded-address %) diff --git a/src/org/gnu/clojure/gnunet/transport.clj b/src/org/gnu/clojure/gnunet/transport.clj index 127fc80..651aa06 100644 --- a/src/org/gnu/clojure/gnunet/transport.clj +++ b/src/org/gnu/clojure/gnunet/transport.clj @@ -229,19 +229,23 @@ (:transport ping)) ;; XXX: Here we're looking for an exact match, gnunet allows transport ;; plugins to do inexact matches. - :when (contains? transport-addresses (:encoded-address ping))] - (let [pong {:challenge (:challenge ping) - :signature-purpose signature-purpose-pong-own - :expiration (pong-expiration) - :peer-id (:id peer) - :transport (:transport ping) - :encoded-address (:encoded-address ping)} - 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. + :when (contains? transport-addresses (:encoded-address ping)) + pong {:challenge (:challenge ping) + :signature-purpose signature-purpose-pong-own + :expiration (pong-expiration) + :peer-id (:id peer) + :transport (:transport ping) + :encoded-address (:encoded-address ping)} + 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) + state (deref (:state-agent remote-peer))] + (if (:is-connected state) + (let [transport (:connected-transport state) + encoded-address (:connected-address state)] + ((:emit-messages! transport) transport remote-peer encoded-address + nil [{:message-type message-type-pong :bytes encoded-pong}])) (doseq [transports (deref (:transport-addresses-agent remote-peer)) address (val transports)] (when-let [transport ((deref (:transports-agent peer)) @@ -251,34 +255,42 @@ [{:message-type message-type-pong :bytes encoded-pong}])))))))) (defn send-pong-using! - [peer remote-peer ping] + [peer remote-peer address ping] (.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}]))))))) + (let + [state (deref (:state-agent remote-peer)) + pong {:challenge (:challenge ping) + :signature-purpose signature-purpose-pong-using + :expiration (pong-expiration) + :peer-id (:id remote-peer) + :transport (:transport address) + :encoded-address (:encoded-address address)} + 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) + state (deref (:state-agent remote-peer))] + (if (:is-connected state) + (let [transport (:connected-transport state) + encoded-address (:connected-address state)] + ((:emit-messages! transport) transport remote-peer encoded-address + nil [{:message-type message-type-pong :bytes encoded-pong}])) + (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] + [peer remote-peer address message] (when-let [ping (first (parse-ping (:bytes message)))] (cond (not (= (:peer-id ping) (:id peer))) nil (:transport ping) (send-pong-own! peer remote-peer ping) - :else (send-pong-using! peer remote-peer ping)))) + :else (send-pong-using! peer remote-peer address ping)))) (defn handle-pong-own! [peer remote-peer pong] @@ -338,7 +350,7 @@ remote-peer (remote-peers sender-id)] (condp = (:message-type message) message-type-hello (handle-hello! peer message) - message-type-ping (handle-ping! peer remote-peer message) + message-type-ping (handle-ping! peer remote-peer address message) message-type-pong (handle-pong! peer message) (handle-receive! peer remote-peer message)) remote-peers)))) diff --git a/src/org/gnu/clojure/gnunet/util.clj b/src/org/gnu/clojure/gnunet/util.clj index d4de23f..329df62 100644 --- a/src/org/gnu/clojure/gnunet/util.clj +++ b/src/org/gnu/clojure/gnunet/util.clj @@ -1,5 +1,13 @@ (ns org.gnu.clojure.gnunet.util) +(defmacro assert-args [fnname & pairs] + `(do (when-not ~(first pairs) + (throw (IllegalArgumentException. + ~(str fnname " requires " (second pairs))))) + ~(let [more (nnext pairs)] + (when more + (list* `assert-args fnname more))))) + (defn queue-seq! "Consume a queue and present it as a sequence." [queue] |