Merge pull request #88 from cl-rabbit/cl-bunny

michaelklishin · michaelklishin · commit b6054156b0ad · 2016-03-02T17:03:30.000+03:00
Common Lisp tutorials
diff --git a/.gitignore b/.gitignore
@@ -17,4 +17,5 @@ venv/*
 ruby*/rubygems*
 
 java*/.idea/workspace.xml
-java*/.idea/encodings.xml
+java*/.idea/encodings.xml
+*~
diff --git a/common-lisp/README.md b/common-lisp/README.md
@@ -0,0 +1,50 @@
+# Common Lisp code for RabbitMQ tutorials
+
+Here you can find Common Lisp code examples from
+[RabbitMQ tutorials](http://cl-rabbit.io/cl-bunny/tutorials/).
+
+## Requirements
+
+To run this code you need [Cl-Bunny](http://cl-rabbit.io/cl-bunny).
+
+You can install it via Quicklisp:
+
+    (ql:quickload :cl-bunny)
+
+Note: Cl-Bunny developed and tested using x64 sbcl on GNU/Linux
+
+All our examples are in fact executable sbcl scripts. You can run them from command line
+
+## Code
+
+[Tutorial one: "Hello World!"](http://cl-rabbit.io/cl-bunny/tutorials/tutorial-one-cl.html):
+
+    ./send.lisp
+    ./receive.lisp
+
+[Tutorial two: Work Queues](http://cl-rabbit.io/cl-bunny/tutorials/tutorial-two-cl.html):
+
+    ./new-task.lisp
+    ./worker.lisp
+
+[Tutorial three: Publish/Subscribe](http://cl-rabbit.io/cl-bunny/tutorials/tutorial-three-cl.html)
+
+    ./receive-logs.lisp
+    ./emit-log.lisp
+
+[Tutorial four: Routing](http://cl-rabbit.io/cl-bunny/tutorials/tutorial-four-cl.html)
+
+    ./receive-logs-direct.lisp
+    ./emit-log-direct.lisp
+
+[Tutorial five: Topics](http://cl-rabbit.io/cl-bunny/tutorials/tutorial-five-cl.html)
+
+    ./receive-logs-topic.lisp
+    ./emit-log-topic.lisp
+
+[Tutorial six: RPC](http://cl-rabbit.io/cl-bunny/tutorials/tutorial-six-cl.html)
+
+    ./rpc-server.lisp
+    ./rpc-client.lisp
+
+To learn more, visit [Cl-Bunny documentation](http://cl-rabbit.io/cl-bunny) site.
diff --git a/common-lisp/emit-log-direct.lisp b/common-lisp/emit-log-direct.lisp
@@ -0,0 +1,18 @@
+#!/bin/sh
+
+sbcl --noinform --noprint $@ <<EOF
+
+(ql:quickload :cl-bunny.examples)
+
+(in-package :cl-bunny.examples)
+
+(with-connection ()
+  (with-channel ()
+    (let* ((args (rest sb-ext:*posix-argv*))
+           (severity (or (first args) "info"))
+           (msg (format nil "~{~a~^ ~}" (rest args)))
+           (x (exchange.direct "direct_logs")))
+      (publish x msg :routing-key severity)
+      (format t " [x] Sent '~a'~%" msg))))
+
+EOF
diff --git a/common-lisp/emit-log-topic.lisp b/common-lisp/emit-log-topic.lisp
@@ -0,0 +1,18 @@
+#!/bin/sh
+
+sbcl --noinform --noprint $@ <<EOF
+
+(ql:quickload :cl-bunny.examples)
+
+(in-package :cl-bunny.examples)
+
+(with-connection ()
+  (with-channel ()
+    (let* ((args (rest sb-ext:*posix-argv*))
+           (severity (or (first args) "info"))
+           (msg (format nil "~{~a~^ ~}" (rest args)))
+           (x (exchange.topic "topic_logs")))
+      (publish x msg :routing-key severity)
+      (format t " [x] Sent '~a'~%" msg))))
+
+EOF
diff --git a/common-lisp/emit-log.lisp b/common-lisp/emit-log.lisp
@@ -0,0 +1,16 @@
+#!/bin/sh
+
+sbcl --noinform --noprint $@ <<EOF
+
+(ql:quickload :cl-bunny.examples)
+
+(in-package :cl-bunny.examples)
+
+(with-connection ("amqp://")
+  (with-channel ()
+    (let ((msg (format nil "~{~a~^ ~}" (cdr sb-ext:*posix-argv*)))
+          (x (exchange.fanout "logs")))
+      (publish x msg)
+      (format t " [x] Sent '~a'~%" msg))))
+
+EOF
diff --git a/common-lisp/new-task.lisp b/common-lisp/new-task.lisp
@@ -0,0 +1,17 @@
+#!/bin/sh
+
+sbcl --noinform --noprint $@ <<EOF
+
+(ql:quickload :cl-bunny.examples)
+
+(in-package :cl-bunny.examples)
+
+(with-connection ()
+  (with-channel ()
+    (let ((x (exchange.default))
+          (msg (format nil "~{~a~^ ~}" (cdr sb-ext:*posix-argv*))))
+      (publish x msg :routing-key "task_queue"
+                     :properties '(:persistent t))
+      (format t " [x] Sent '~a'~%" msg))))
+
+EOF
diff --git a/common-lisp/receive-logs-direct.lisp b/common-lisp/receive-logs-direct.lisp
@@ -0,0 +1,27 @@
+#!/bin/sh
+
+sbcl --noinform --noprint $@ <<EOF
+
+(ql:quickload :cl-bunny.examples)
+
+(in-package :cl-bunny.examples)
+
+(if-let ((args (rest sb-ext:*posix-argv*)))
+  (with-connection ()
+    (with-channel ()
+      (let ((q (queue.declare :auto-delete t))
+            (x (exchange.direct "direct_logs")))
+        (loop for severity in args do
+                 (queue.bind q x :routing-key severity))
+        (format t " [*] Waiting for logs. To exit press CTRL+C~%")
+        (handler-case
+            (progn
+              (subscribe q (lambda (message)
+                              (format t " [x] #~a:~a~%" (message-routing-key message) (message-body-string message)))
+                         :type :sync)
+              (consume))
+          (sb-sys:interactive-interrupt ()
+            (sb-ext:exit))))))
+  (format t "Usage: $0 [info] [warning] [error]~%"))
+
+EOF
diff --git a/common-lisp/receive-logs-topic.lisp b/common-lisp/receive-logs-topic.lisp
@@ -0,0 +1,27 @@
+#!/bin/sh
+
+sbcl --noinform --noprint $@ <<EOF
+
+(ql:quickload :cl-bunny.examples)
+
+(in-package :cl-bunny.examples)
+
+(if-let ((args (rest sb-ext:*posix-argv*)))
+  (with-connection ()
+    (with-channel ()
+      (let ((q (queue.declare :auto-delete t))
+            (x (exchange.topic "topic_logs")))
+        (loop for severity in args do
+                 (queue.bind q x :routing-key severity))
+        (format t " [*] Waiting for logs. To exit press CTRL+C~%")
+        (handler-case
+            (progn
+              (subscribe q (lambda (message)
+                              (format t " [x] #~a:~a~%" (message-routing-key message) (message-body-string message)))
+                         :type :sync)
+              (consume))
+          (sb-sys:interactive-interrupt ()
+            (sb-ext:exit))))))
+  (format t "Usage: $0 [binding key]~%"))
+
+EOF
diff --git a/common-lisp/receive-logs.lisp b/common-lisp/receive-logs.lisp
@@ -0,0 +1,23 @@
+#!/bin/sh
+
+sbcl --noinform --noprint <<EOF
+
+(ql:quickload :cl-bunny.examples)
+
+(in-package :cl-bunny.examples)
+
+(with-connection ("amqp://")
+  (with-channel ()
+    (let ((q (queue.declare-temp)))
+      (queue.bind q "logs")
+      (format t " [*] Waiting for logs. To exit press CTRL+C~%")
+      (handler-case
+          (progn
+            (subscribe q (lambda (message)
+                            (format t " [x] #~a~%" (message-body-string message)))
+                       :type :sync)
+            (consume))
+        (sb-sys:interactive-interrupt ()
+          (sb-ext:exit))))))
+
+EOF
diff --git a/common-lisp/receive.lisp b/common-lisp/receive.lisp
@@ -0,0 +1,23 @@
+#!/bin/sh
+
+sbcl --noinform --noprint <<EOF
+
+(ql:quickload :cl-bunny.examples)
+
+(in-package :cl-bunny.examples)
+
+(with-connection ("amqp://")
+  (with-channel ()
+    (let ((q (queue.declare :name "hello")))
+      (format t " [*] Waiting for messages in queue 'hello'. To exit press CTRL+C~%")
+      (handler-case
+          (progn
+            (subscribe q (lambda (message)
+                            (format t " [x] Received ~a~%" (message-body-string message)))
+                       :type :sync
+                       :no-ack t)
+            (consume))
+        (sb-sys:interactive-interrupt ()
+          (sb-ext:exit))))))
+
+EOF
diff --git a/common-lisp/rpc-client.lisp b/common-lisp/rpc-client.lisp
@@ -0,0 +1,46 @@
+#!/bin/sh
+
+sbcl --noinform --noprint <<EOF
+
+(ql:quickload :cl-bunny.examples)
+(ql:quickload :nibbles)
+
+(in-package :cl-bunny.examples)
+
+(defun int64-to-octets(val)
+  (let ((obuffer (fast-io:make-output-buffer)))
+    (fast-io:write64-be val obuffer)
+    (fast-io:finish-output-buffer obuffer)))
+
+(defun start-client (n)
+  (with-connection ("amqp://")
+    (with-channel ()
+      (let ((x (exchange.default))
+            (server-queue "rpc_queue")
+            (reply-queue (queue.declare :auto-delete t))
+            (lock (bt:make-lock))
+            (condition (bt:make-condition-variable))
+            (result nil))
+        (format t " [x] Requesting fib(~a)~%" n)
+        (bt:with-lock-held (lock)
+          (subscribe reply-queue (lambda (message)
+                                    (bt:with-lock-held (lock)
+                                      (setf result (nibbles:sb64ref/be (message-body message) 0))
+                                      (bt:condition-notify condition))))
+          (publish x
+                   (int64-to-octets n)
+                   :routing-key server-queue
+                   :properties (list :correlation-id (format nil "~a~a~a" (random 100) (random 100) (random 100))
+                                     :reply-to reply-queue))
+          (bt:condition-wait condition lock)
+          (format t " [.] Got ~a~%" result)
+          result)))))
+
+(start-client 0)
+(start-client 1)
+(start-client 22)
+(start-client 33)
+(start-client 44)
+(start-client 55)
+
+EOF
diff --git a/common-lisp/rpc-server.lisp b/common-lisp/rpc-server.lisp
@@ -0,0 +1,54 @@
+#!/bin/sh
+
+sbcl --noinform --noprint <<EOF
+
+(ql:quickload :cl-bunny.examples)
+(ql:quickload :nibbles)
+
+(in-package :cl-bunny.examples)
+
+(defun int64-to-octets(val)
+  (let ((obuffer (fast-io:make-output-buffer)))
+    (fast-io:write64-be val obuffer)
+    (fast-io:finish-output-buffer obuffer)))
+
+;; http://www.cliki.net/fibonacci
+(defun fibonacci (n)
+  "Successive squaring method from SICP"
+  (check-type n (integer 0 *))
+  (labels ((fib-aux (a b p q count)
+             (cond ((= count 0) b)
+                   ((evenp count)
+                    (fib-aux a
+                             b
+                             (+ (* p p) (* q q))
+                             (+ (* q q) (* 2 p q))
+                             (/ count 2)))
+                   (t (fib-aux (+ (* b q) (* a q) (* a p))
+                               (+ (* b p) (* a q))
+                               p
+                               q
+                               (- count 1))))))
+    (fib-aux 1 0 0 1 n)))
+
+(with-connection ()
+  (with-channel ()
+    (let ((x (exchange.default))
+          (q (queue.declare :name "rpc_queue" :auto-delete t)))
+      (format t " [x] Awaiting RPC requests~%")
+      (handler-case
+          (progn
+            (subscribe q (lambda (message)
+                            (let* ((n (nibbles:sb64ref/be (message-body message) 0))
+                                   (r (fibonacci n)))
+                              (format t " [.] fib(~a)~%" r)
+                              (publish x
+                                       (int64-to-octets r)
+                                       :routing-key (message-reply-to message)
+                                       :properties (list :correlation-id (message-correlation-id message)))))
+                       :type :sync)
+            (consume))
+        (sb-sys:interactive-interrupt ()
+          (sb-ext:exit))))))
+
+EOF
diff --git a/common-lisp/send.lisp b/common-lisp/send.lisp
@@ -0,0 +1,14 @@
+#!/bin/sh
+
+sbcl --noinform --noprint <<EOF
+
+(ql:quickload :cl-bunny.examples)
+
+(in-package :cl-bunny.examples)
+
+(with-connection ("amqp://")
+  (with-channel ()
+    (publish (exchange.default) "Hello world!" :routing-key "hello")
+    (format t " [x] Sent 'Hello World!'~%")))
+
+EOF
diff --git a/common-lisp/worker.lisp b/common-lisp/worker.lisp
@@ -0,0 +1,28 @@
+#!/bin/sh
+
+sbcl --noinform --noprint <<EOF
+
+(ql:quickload :cl-bunny.examples)
+
+(in-package :cl-bunny.examples)
+
+(with-connection ()
+  (with-channel ()
+    (let ((q (queue.declare :name "task_queue" :durable t)))
+      (format t " [*] Waiting for messages in queue 'task_queue'. To exit press CTRL+C~%")
+      (qos :prefetch-count 1)
+      (handler-case
+          (progn
+            (subscribe q (lambda (message)
+                            (let ((body (message-body-string message)))
+                              (format t " [x] Received '~a'~%" body)
+                              ;; imitate some work
+                              (sleep (count #\. body))
+                              (message.ack message)
+                              (format t " [x] Done~%")))
+                       :type :sync)
+            (consume))
+        (sb-sys:interactive-interrupt ()
+          (sb-ext:exit))))))
+
+EOF
