clojure async interceptors demo

1. deps.edn file

{:paths       ["src/clj" "src/cljs" "test" "resources" "target"]
 :extra-paths ["resources" "resources/public"]              ;; fix cambada packaging
 :deps        {org.clojure/clojure          {:mvn/version "1.10.0-beta6"}
               org.clojure/core.async       {:mvn/version "0.4.474"}
               org.clojure/spec.alpha       {:mvn/version "0.2.176"}
               org.clojure/core.specs.alpha {:mvn/version "0.2.44"}
               http-kit/http-kit            {:mvn/version "2.3.0"}
               metosin/sieppari             {:mvn/version "0.0.0-alpha6"}
               metosin/reitit               {:mvn/version "0.2.7"}
               fipp                         {:mvn/version "0.6.14"}}
 :aliases     {:repl {:extra-deps  {nrepl/nrepl                {:mvn/version "0.4.5"}
                                    com.bhauman/rebel-readline {:mvn/version "0.1.4"}
                                    cider/piggieback           {:mvn/version "0.3.8"}
                                    com.bhauman/figwheel-main  {:mvn/version "0.1.9"}}
                      :extra-paths ["src" "env/development/clj"]
                      :main-opts   ["--main" "nrepl.cmdline" "--port" "7888" "--middleware" "[cider.piggieback/wrap-cljs-repl]"]}}}

2. Source code

(ns test01.core  (:gen-class)
  (:require [org.httpkit.server :as kit]
            [org.httpkit.client :as kit-client]
            [sieppari.core :as sieppari]
            [sieppari.context]
            [muuntaja.interceptor]
            [clojure.core.async :as a :refer [go chan <! <!! >! >!! timeout close! put! take! alts! alts!!]]
            [reitit.ring :as ring]
            [reitit.http :as http]
            [reitit.core :as r]
            [fipp.edn :refer (pprint)]
            [reitit.interceptor.sieppari]))


(def  #(a/go %))
(def  identity)
(def  #(a/thread %))


(defn prn-tap [v]
  (println v))

(add-tap prn-tap)

(defn interceptor [f x]
  {:enter (fn [ctx] (f                      (update-in ctx [:request :via] (fnil conj []) {:enter  x                                                                     :thread (.getName (Thread/currentThread))})))
   :leave (fn [ctx] (f                      (update-in ctx [:response :body] conj {:leave  x                                                             :thread (.getName (Thread/currentThread))})))})

(defn handler [f data]
  (fn [{:keys [via] :as request}]
    (f      (do        (pprint (dissoc request :reitit.core/match :reitit.core/router))
        {:status 200,
         :body   (conj via (str {:handler data                                 :thread  (.getName (Thread/currentThread))}))}))))



(def routes (http/router [["/" {:get {:handler (handler  "root")}}]
                          ["/test1"                           ["" {:name         ::test1                                :interceptors [(interceptor  :test1)]
                                :get          {:handler (handler  "test1")}}]
                           ["/test2" {:interceptors [(interceptor  :test2-level)]
                                      :get          {:interceptors [(interceptor  :test2-get)]
                                                     :handler      (handler  "test2")}}]]]))

;;(r/routes routes)
(def app (http/ring-handler routes                            (ring/create-default-handler)
                            {:executor     reitit.interceptor.sieppari/executor                             :interceptors []}))

(app {:uri            "/test1/test2"      :request-method :get})


(def params {:host "0.0.0.0" :port 8080})

(defn start-server  [handler params]
  (kit/run-server handler params))


(defn -main "entry point"  [& args]
  (start-server app params)
  (println "server is started: " params))



;; ----------- client and repl zone---------------------------------------
(defn handle-response  [resp]
  (println (:body resp)))

(comment  (def stop-fn (start-server app params))
  (time (dotimes [_ 10]
          (kit-client/get "http://localhost:8080" {:timeout 20000} handle-response)))
  (time (slurp "http://localhost:8080/test1"))
  (stop-fn))

clojure reitit router

all you need to know is here

(def routes (http/router [["/" {:interceptors [(interceptor  :async) (interceptor2 :root-2)]
                                :get          {:interceptors [(interceptor  :get) (interceptor2 :get-2)]
                                               :handler      (handler  "root")}}]
                          ["/test1" {:interceptors [(interceptor  :level-test-1-1) (interceptor2 :level-test-1-2)]}
                           ["" {:name ::test1                                :interceptors [(interceptor  :test1-hanler-level-int1) (interceptor2 :test1-hanler-level-int1)]
                                :get  {:handler (handler  "test1")}}]
                           ["/test2" {:interceptors [(interceptor  :level-test-2-1) (interceptor2 :level-test-2-2)]
                                :get  {:interceptors [(interceptor  :test2-hanler-level-int1) (interceptor2 :test2-hanler-level-int2)]
                                       :handler (handler  "test2")}}]]]))

clojure interceptors

1. dep.edn

{:paths       ["src/clj" "src/cljs" "test" "resources" "target"]
 :extra-paths ["resources" "resources/public"]              ;; fix cambada packaging
 :deps        {org.clojure/clojure          {:mvn/version "1.10.0-RC1"}
               org.clojure/core.async       {:mvn/version "0.4.474"}
               org.clojure/spec.alpha       {:mvn/version "0.2.176"}
               org.clojure/core.specs.alpha {:mvn/version "0.2.44"}
               http-kit/http-kit            {:mvn/version "2.3.0"}
               metosin/sieppari             {:mvn/version "0.0.0-alpha5"}
               fipp                         {:mvn/version "0.6.13"}}}

2. source code

(ns test01.core  (:gen-class)
  (:require [org.httpkit.server :as kit]
            [org.httpkit.client :as kit-client]
            [sieppari.core :as sieppari]
            [sieppari.context]
            [clojure.core.async :as a :refer [go chan <! <!! >! >!! timeout close! put! take! alts! alts!!]]))


(defn prn-tap [v]
  (println v))

(add-tap prn-tap)

;;(require '[fipp.edn :refer (pprint) :rename {pprint fipp}])
(def int-1  {:name :int-1   :enter (fn [ctx]
            (tap> :int-1)
            ctx)
   :leave (fn [ctx]
            (tap> :int-1-leave)
            ctx)})


(def int-2  {:name :int-2   :enter (fn [ctx]
            (go              (<! (a/timeout 1000))
              (tap> :int-2)
              ctx))})

(def int-3  {:name :int-3   :enter (fn [ctx]
            (sieppari.context/terminate ctx {:status  200                                             :headers {"Content-Type" "text/plain"}
                                             :body    "yo! error!"}))})

(def int-4  {:name :int-4   :enter (fn [ctx]
            (/ 1 0)
            ctx)
   :error (fn [ctx]
            (tap> ["got error:" (-> ctx :error .getMessage)])
            ;;(fipp ctx)            (assoc ctx :error nil                       :response {:status  200                                  :headers {"Content-Type" "text/plain"}
                                  :body   ["got error: " (-> ctx :error .getMessage)] }))})

(defn handler [request]
  {:status  200   :headers {"Content-Type" "text/plain"}
   :body    (:remote-addr request)})

(def pipeline [int-1 int-2 #_int-3 #_int-4 handler])

(defn exec-request [request]
  (a/thread    (tap> ["exec  " (.getName (Thread/currentThread))])
    (sieppari/execute pipeline request)))


(defn drain [c]
  (go (while (not (nil? (<! c)))
        nil)))

(defn async-kit-app  [request]
  (tap> ["async " (.getName (Thread/currentThread))])
  (kit/with-channel request channel                    (do                      (take! (exec-request request) #(kit/send! channel %))
                      (tap> ["async2 " (.getName (Thread/currentThread))]))))

(def params {:host "0.0.0.0" :port 8080})

(defn start-kit  [handler]
  (kit/run-server handler params))


(defn -main "entry point"  [& args]
  (start-kit async-kit-app)
  (println "started: " params))

(defn handle-response  [resp]
  (println (:body resp)))

(comment  (def stop-fn (start-kit async-kit-app))
  (time (dotimes [_ 1000]
     (kit-client/get "http://localhost:8080" {:timeout 20000} handle-response)))
  (slurp "http://localhost:8080")
  (stop-fn))

(ns test01.ret  (:gen-class)
  (:require [org.httpkit.server :as kit]
            [org.httpkit.client :as kit-client]
            [sieppari.core :as sieppari]
            [sieppari.context]
            [muuntaja.interceptor]
            [clojure.core.async :as a :refer [go chan <! <!! >! >!! timeout close! put! take! alts! alts!!]]
            [reitit.ring :as ring]
            [reitit.http :as http]
            [reitit.interceptor.sieppari]))



(defn interceptor [f x]
  {:enter (fn [ctx] (f (update-in ctx [:request :via] (fnil conj []) {:enter x})))
   :leave (fn [ctx] (f (update-in ctx [:response :body] conj {:leave x})))})

(defn interceptor2 [this-name]
  {
   :enter (fn [ctx] (update-in ctx [:request :via] (fnil conj []) {:enter this-name}))
   :leave (fn [ctx] (update-in ctx [:response :body] conj {:leave this-name}))})

(defn handler [f]
  (fn [{:keys [via]}]
    (f {:status 200,
        :body   (conj via :handler)})))


(def  #(a/go %))
(def  identity)

(def routes (http/router [["/" {:interceptors [(interceptor  :async) (interceptor2 :root-2)]
                                :get          {:interceptors [(interceptor  :get) (interceptor2 :get-2)]
                                               :handler      (handler )}}]
                          ]))

(def app (http/ring-handler routes                            (ring/create-default-handler)
                            {:executor     reitit.interceptor.sieppari/executor                             :interceptors [(muuntaja.interceptor/format-interceptor)]}))


(def params {:host "0.0.0.0" :port 8080})

(defn start-server  [handler params]
  (kit/run-server handler params))


(defn -main "entry point"  [& args]
  (start-server app params)
  (println "server is started: " params))



;; ----------- client and repl zone---------------------------------------(defn handle-response  [resp]
  (println (:body resp)))

(comment  (def stop-fn (start-server app params))
  (time (dotimes [_ 10]
          (kit-client/get "http://localhost:8080" {:timeout 20000} handle-response)))
  (time (slurp "http://localhost:8080"))
  (stop-fn))

clojure websocket client example

This is my deps.edn
https://github.com/middlesphere/spacemacs-clojure-cheatsheet/blob/master/.clojure/deps.edn

1) Start REPL with necessary deps.

clojure -R:1.10:repl:add-lib:test:debug-tools:rebel:xml-bind -m nrepl.cmdline -p 7888 -i -m "[refactor-nrepl.middleware/wrap-refactor,cider.nrepl/cider-middleware]"
nREPL server started on port 7888 on host 0:0:0:0:0:0:0:0 - nrepl://0:0:0:0:0:0:0:0:7888
nREPL 0.4.5
Clojure 1.10.0-alpha8
Java HotSpot(TM) 64-Bit Server VM 10.0.2+13
user=> (use 'clojure.tools.deps.alpha.repl)
nil
user=> (add-lib 'stylefruits/gniazdo {:mvn/version "1.0.1"})
true
user=> (add-lib 'http-kit {:mvn/version "2.2.0"})
true

2) Create empty file (e.g a.clj), connect to REPL (port 7888)  and put following code there:

;; ------------server code --------------------

(use 'org.httpkit.server)

(defn async-handler [ring-request]
  ;; unified API for WebSocket and HTTP long polling/streaming
  (with-channel ring-request channel    ; get the channel
    (if (websocket? channel)            ; if you want to distinguish them
      (on-receive channel (fn [data]     ; two way communication
                            (send! channel data)))
      (send! channel {:status  200
                      :headers {"Content-Type" "text/plain"}
                      :body    "Long polling?"}))))

(run-server async-handler {:port 8080})

;; ------------client code --------------------

(require '[gniazdo.core :as ws])

(def socket (ws/connect "ws://localhost:8080/" :on-receive #(prn 'received %)))
(ws/send-msg socket "hello")
(ws/send-msg socket "world")

(ws/close socket)

Run lein with cider from command line

Headless mode

lein update-in :dependencies conj \[org.clojure/tools.nrepl\ \"0.2.13\"\ \:exclusions\ \[org.clojure/clojure\]\] -- update-in :plugins conj \[refactor-nrepl\ \"2.4.0\"\] -- update-in :plugins conj \[cider/cider-nrepl\ \"0.18.0\"\] -- with-profile +mike-local repl :headless :host ::

Interactive mode

lein update-in :dependencies conj \[org.clojure/tools.nrepl\ \"0.2.13\"\ \:exclusions\ \[org.clojure/clojure\]\] -- update-in :plugins conj \[refactor-nrepl\ \"2.4.0\"\] -- update-in :plugins conj \[cider/cider-nrepl\ \"0.18.0\"\] -- with-profile +mike-local repl :start :host ::

where mike-local is my profile

Also, :port parameter can be used to run REPL on custom port.

clojure cli repl rebel

First way

In order to run REPL (nREPL) server via clojure cli tools with beautiful rebel capabilities just add following section into deps.edn file:

:aliases {
                :repl {:extra-deps {cider/cider-nrepl                       {:mvn/version "0.18.0"}
                                               nrepl                                        {:mvn/version "0.4.5"}
                                               refactor-nrepl                          {:mvn/version "2.4.0"}
                                               com.bhauman/rebel-readline  {:mvn/version "0.1.4"}}
                  :main-opts  ["-e" "(require,(quote,cider-nrepl.main)),"
                               "-e" "(cider-nrepl.main/init,[\"refactor-nrepl.middleware/wrap-refactor\",\"cider.nrepl/cider-middleware\"])"
                               "-e" "(use,'rebel-readline.main),(rebel-readline.main/-main)"]}

}

Then just run with command: clojure -A:repl

Second way

:aliases {
 :repl {:extra-deps {cider/cider-nrepl          {:mvn/version "0.18.0"}
                               nrepl                      {:mvn/version "0.4.5"}
                               refactor-nrepl             {:mvn/version "2.4.0"}
                               com.bhauman/rebel-readline {:mvn/version "0.1.4"}}
                  }
}

clj -R:repl -m nrepl.cmdline -p 7888 -i -m "[refactor-nrepl.middleware/wrap-refactor,cider.nrepl/cider-middleware]"

This command runs repl without rebel capabilities. In order to enable rebel just call it:

user=> (use 'rebel-readline.main)
nil
user=> (rebel-readline.main/-main)
[Rebel readline] Type :repl/help for online help info
user=>


Also, read the docs https://nrepl.readthedocs.io/en/latest/usage/

socks proxy on centos 7

0. yum install gcc pam-devel tcp_wrappers-devel

1. yum install http://mirror.ghettoforge.org/distributions/gf/gf-release-latest.gf.el7.noarch.rpm

2. yum --enablerepo=gf-plus install dante-server

3. systemctl enable sockd.service

4. mkdir /var/run/sockd

5. cp /etc/sockd.conf /etc/sockd.conf.orig

6. vi /etc/sockd.conf

# the server will log both via syslog, to stdout and to /var/log/sockd.log                                                                                                                                

logoutput: syslog stdout /var/log/sockd.log

logoutput: stderr

# The server will bind to the address 10.1.1.1, port 1080 and will only                                                                                                                                   

# accept connections going to that address.                                                                                                                                                               

#internal: 10.1.1.1 port = 1080                                                                                                                                                                           

# Alternatively, the interface name can be used instead of the address.                                                                                                                                   

internal: eth0 port = 80

# all outgoing connections from the server will use the IP address                                                                                                                                        

# 195.168.1.1                                                                                                                                                                                             

#external: 192.168.1.1                                                                                                                                                                                    

external: eth0

# list over acceptable authentication methods, order of preference.                                                                                                                                       

# An authentication method not set here will never be selected.                                                                                                                                           

#                                                                                                                                                                                                         

# If the socksmethod field is not set in a rule, the global                                                                                                                                               

# socksmethod is filled in for that rule.                                                                                                                                                                 

#                                                                                                                                                                                                         

# methods for socks-rules.                                                                                                                                                                                

socksmethod: username

#                                                                                                                                                                                                         

# User identities, an important section.                                                                                                                                                                  

#                                                                                                                                                                                                         

# when doing something that can require privilege, it will use the                                                                                                                                        

# userid "sockd".                                                                                                                                                                                         

user.privileged: root

# when running as usual, it will use the unprivileged userid of "sockd".                                                                                                                                  

user.unprivileged: nobody

client pass {

        from: 0.0.0.0/0 to: 0.0.0.0/0

        log: error connect disconnect

}

client block {

        from: 0.0.0.0/0 to: 0.0.0.0/0

        log: connect error

}

socks pass {

        from: 0.0.0.0/0 to: 0.0.0.0/0

        log: error connect disconnect

}

socks block {

        from: 0.0.0.0/0 to: 0.0.0.0/0

        log: connect error

}

7. sudo useradd -s /bin/false proxyuser

8. sudo passwd proxyuser

9. systemctl start sockd.service

core.async tip from Tim Baldridge

If I'm understand the problem I think this is by design. Closing a channel results in a logical termination value being put into the channel.

This is so that, by default, core.async won't loose data.

Most of the time when I encounter this problem I fix it by draining the channel:

(defn drain [c]
  (go (while (not (nil? (<! c)))
          nil)))


And then:

(close! c)
(drain c)

clojure ring async example

;;requirements

[ring "1.6.3"]
[ring/ring-jetty-adapter "1.6.3"]
[org.clojure/core.async "0.4.474"]
[org.immutant/web "2.1.10"]

;;Jetty

(ns test01.core
  (:gen-class)
  (:require [ring.adapter.jetty :as jetty]
            [clojure.core.async :refer [go chan <! <!! >! >!! timeout close!]]))


(defn what-is-my-ip [request respond raise]
  (respond {:status  200
            :headers {"Content-Type" "text/plain"}
            :body    (:remote-addr request)}))



(defn what-is-my-ip2 [request respond raise]
  (let [ch (chan)]
    (go
      (respond (<! ch)))
    (go
      ;;(<! (timeout 2000))
      (>! ch {:status  200
              :headers {"Content-Type" "text/plain"}
              :body    (:remote-addr request)}))))

(defn start-server
  [handler]
  (jetty/run-jetty handler {:host "localhost" :async? true :port 8080 :path "/" :join? false}))


(defn -main
  "entry point"
  [& args]
  (println "Hello, world."))

;; (def s (start-server what-is-my-ip))
;; (.stop s)
;;(def s (start-server what-is-my-ip2))

;;Immutant

(ns test01.imm
  (:require [immutant.web :as web]
            [immutant.web.async :as async]))


(defn app [request]
  (async/as-channel request
                    {:on-open (fn [stream]
                                (async/send! stream {:status  200
                                                     :headers {"Content-Type" "text/plain"}
                                                     :body    (:remote-addr request)}
                                             {:close? true}))}))


(defn start-server [h]
  (web/run h {:host "localhost" :port 8080}))

(def s (start-server app))

(web/stop s)