reduce creation of canonical probes in servers

src/aleph/http/client.clj

@@ -78,20 +78,23 @@
 		 identity
 		 options)]
     (run-pipeline client
+      :error-handler (fn [_])
       (fn [ch]
 	(splice
 	  (wrap-response-stream options ch)
 	  (siphon->> (wrap-request-stream options) ch))))))
 
 (defn- http-client- [client-fn options]
-  (let [options (merge
+  (let [options (process-options options)
+	options (merge
 		  {:name
 		   (str "http-client:" (:server-name options) ":" (gensym ""))
 		   :description
 		   (str (:scheme options) "://" (:server-name options) ":" (:server-port options))}
-		  (process-options options))
+		  options)
 	client (client-fn
-		 #(http-connection options))
+		 #(http-connection options)
+		 options)
 	f (fn [request timeout]
 	    (if (map? request)
 	      (client (assoc (merge options request)

src/aleph/netty.clj

@@ -161,6 +161,30 @@
 	(.close ch)
 	(.sendUpstream ctx evt)))))
 
+(def pipeline-handlers
+  (memoize
+    (fn [pipeline-name]
+      (let [error-probe (canonical-probe [pipeline-name :errors])
+	    error-handler (fn [evt]
+			    (when-let [ex ^ExceptionEvent (exception-event evt)]
+			      (when-not (instance? ClosedChannelException ex)
+				(when-not (trace error-probe
+					    {:exception ex
+					     :address (-> ex .getChannel channel-origin)})
+				  (log/error nil ex)))
+			      nil))
+	    traffic-handler (fn [probe-suffix]
+			      (let [traffic-probe (canonical-probe [pipeline-name :traffic probe-suffix])]
+				(fn [evt]
+				  (when-let [msg (message-event evt)]
+				    (trace traffic-probe
+				      {:address (-> evt channel-event channel-origin)
+				       :bytes (.readableBytes ^ChannelBuffer msg)}))
+				  nil)))]
+	{:error error-handler
+	 :in (traffic-handler :in)
+	 :out (traffic-handler :out)}))))
+
 (defn create-netty-pipeline
   "Creates a pipeline.  Each stage must have a name.
 
@@ -170,34 +194,17 @@
      :stage-b b)"
   [pipeline-name & stages]
   (let [netty-pipeline (Channels/pipeline)
-	error-probe (canonical-probe [pipeline-name :errors])
-	_ (register-probe error-probe)
-	error-handler (fn [evt]
-			(when-let [ex ^ExceptionEvent (exception-event evt)]
-			  (when-not (instance? ClosedChannelException ex)
-			    (when-not (trace error-probe
-					{:exception ex
-					 :address (-> ex .getChannel channel-origin)})
-			      (log/error nil ex)))
-			  nil))
-	traffic-handler (fn [probe-suffix]
-			  (let [traffic-probe (canonical-probe [pipeline-name :traffic probe-suffix])]
-			    (fn [evt]
-			      (when-let [msg (message-event evt)]
-				(trace traffic-probe
-				  {:address (-> evt channel-event channel-origin)
-				   :bytes (.readableBytes ^ChannelBuffer msg)}))
-			      nil)))]
+	{:keys [error in out]} (pipeline-handlers pipeline-name)]
     (doseq [[id stage] (partition 2 stages)]
       (.addLast netty-pipeline (name id) stage))
     (.addFirst netty-pipeline "incoming-traffic"
-      (upstream-stage (traffic-handler :in)))
+      (upstream-stage in))
     (.addFirst netty-pipeline "outgoing-traffic"
-      (downstream-stage (traffic-handler :out)))
+      (downstream-stage out))
     (.addLast netty-pipeline "outgoing-error"
-      (downstream-stage error-handler))
+      (downstream-stage error))
     (.addFirst netty-pipeline "incoming-error"
-      (upstream-stage error-handler))
+      (upstream-stage error))
     netty-pipeline))
 
 ;;;
@@ -436,6 +443,7 @@
 
     ;; intialize client
     (run-pipeline (.connect client (InetSocketAddress. ^String host (int port)))
+      :error-handler (fn [_])
       wrap-netty-channel-future
       (fn [^Channel netty-channel]
 

test/aleph/benchmark/hello_world.clj

@@ -7,8 +7,8 @@
 
 #_(def stop-server
   (start-http-server
-    (fn [ch request]
-      (enqueue ch
+    (wrap-ring-handler
+      (fn [request]
 	{:status 200
 	 :content-type "text/plain"
 	 :body (str (swap! counter inc))}))