Merge pull request #7 from nilenso/feature/move_ticket_queue_creation…

…_to_component

Initialize message queue in a component instead of it being a globally defined value

swift-ticketing/src/swift_ticketing/app.clj

@@ -8,7 +8,7 @@
    [compojure.api.sweet :refer :all]
    [swift-ticketing.app :as app]))
 
-(defn init-routes [db-spec]
+(defn init-routes [db-spec message-queue]
   (defroutes app-routes
     (GET "/event" request
       (handlers/get-events-handler db-spec request))
@@ -19,21 +19,21 @@
     (POST "/event/:event-id/ticket" request
       (handlers/create-tickets-handler db-spec request))
     (POST "/event/:event-id/booking" request
-      (handlers/reserve-ticket-handler db-spec request))
+      (handlers/reserve-ticket-handler db-spec message-queue request))
     (GET "/ticket" request
       (handlers/get-tickets-handler db-spec request))
     (GET "/booking/:booking-id/status" request
       (handlers/get-booking-status-handler db-spec request))
     (POST "/booking/:booking-id/payment" request
-      (handlers/post-payment-handler db-spec request))
+      (handlers/post-payment-handler db-spec message-queue request))
     (POST "/booking/:booking-id/cancel" request
-      (handlers/cancel-booking-handler db-spec request))
+      (handlers/cancel-booking-handler db-spec message-queue request))
     (GET "/booking/:booking-id/ticket" request
       (handlers/get-tickets-by-booking-id-handler db-spec request))
     (route/not-found "Not Found")))
 
-(defn swift-ticketing-app [db-spec]
-  (-> (init-routes db-spec)
+(defn swift-ticketing-app [db-spec message-queue]
+  (-> (init-routes db-spec message-queue)
       (wrap-defaults (assoc-in site-defaults [:security :anti-forgery] false))
       wrap-json-response
       (wrap-json-body {:keywords? true

swift-ticketing/src/swift_ticketing/components/http_server.clj

@@ -3,14 +3,15 @@
             [ring.adapter.jetty :refer [run-jetty]]
             [swift-ticketing.app :as app]))
 
-(defrecord HTTPServer [port database]
+(defrecord HTTPServer [port database worker]
   component/Lifecycle
 
   (start [component]
     (println ";; Starting API Server")
     (let [connection (:connection database)
+          message-queue (:message-queue worker)
           server (run-jetty
-                  (app/swift-ticketing-app connection)
+                  (app/swift-ticketing-app connection message-queue)
                   {:port port
                    :join? true})]
       (assoc component :http-server server)))

swift-ticketing/src/swift_ticketing/components/worker.clj

@@ -1,21 +1,23 @@
 (ns swift-ticketing.components.worker
   (:require [com.stuartsierra.component :as component]
-            [swift-ticketing.worker :as w]))
+            [swift-ticketing.worker :as w]
+            [clojure.core.async :as async]))
 
 (defrecord Worker [total-workers redis-opts database]
   component/Lifecycle
 
   (start [component]
     (println ";; Spawn workers")
-    (let [connection (:connection database)]
+    (let [connection (:connection database)
+          message-queue (async/chan)]
       (dotimes [i total-workers]
-        (w/process-ticket-requests i connection redis-opts)))
-    component)
+        (w/process-ticket-requests i message-queue connection redis-opts))
+      (assoc component :message-queue message-queue)))
 
   (stop [component]
     (println ";; Stop workers")
-    ; (async/close! w/ticket-queue)
-    component))
+    (async/close! (:message-queue component))
+    (dissoc component :message-queue)))
 
 (defn new-worker [total-workers redis-opts]
   (map->Worker {:total-workers total-workers

swift-ticketing/src/swift_ticketing/core.clj

@@ -19,7 +19,7 @@
      :database (new-database (:database config))
      :app (component/using
            (new-http-server (get-in config [:server :port]))
-           [:database])
+           [:database :worker])
      :worker (component/using
               (new-worker 5 redis-opts)
               [:database]))))

swift-ticketing/src/swift_ticketing/handlers.clj

@@ -93,48 +93,50 @@
      (s/valid? ::specs/event-id event-id)
      (validate-req body ::specs/create-tickets-params on-success))))
 
-(defn- reserve-ticket [db-spec uid event-id booking-req]
+(defn- reserve-ticket [db-spec message-queue uid event-id booking-req]
   (let [booking-id (java.util.UUID/randomUUID)
         req-ticket-ids (:ticket_ids booking-req)
         ticket-ids (when-not (nil? req-ticket-ids)
                      (map #(java.util.UUID/fromString %) req-ticket-ids))]
     (booking/insert-booking db-spec uid booking-id)
-    (worker/add-reserve-ticket-request-to-queue {:booking-id booking-id
-                                                 :ticket-type-id (:ticket_type_id booking-req)
-                                                 :ticket-ids ticket-ids
-                                                 :quantity (:quantity booking-req)})
+    (worker/add-reserve-ticket-request-to-queue
+     message-queue
+     {:booking-id booking-id
+      :ticket-type-id (:ticket_type_id booking-req)
+      :ticket-ids ticket-ids
+      :quantity (:quantity booking-req)})
     booking-id))
 
-(defn reserve-ticket-handler [db-spec request]
+(defn reserve-ticket-handler [db-spec message-queue request]
   (let [{:keys [cookies body route-params]} request
         uid (get-uid cookies)
         event-id (:event-id route-params)
         on-success (fn []
                      (respond-201
                       {"booking_id"
-                       (reserve-ticket db-spec uid event-id body)}))]
+                       (reserve-ticket db-spec message-queue uid event-id body)}))]
     (and
      (s/valid? ::specs/event-id event-id)
      (validate-req body ::specs/reserve-tickets-params on-success))))
 
-(defn- post-payment [db-spec booking-id]
-  (worker/add-book-ticket-request-to-queue {:booking-id booking-id}))
+(defn- post-payment [db-spec message-queue booking-id]
+  (worker/add-book-ticket-request-to-queue message-queue {:booking-id booking-id}))
 
-(defn post-payment-handler [db-spec request]
+(defn post-payment-handler [db-spec message-queue request]
   (let [booking-id (get-in request [:route-params :booking-id])
         on-success (fn []
-                     (post-payment db-spec booking-id)
+                     (post-payment db-spec message-queue booking-id)
                      (respond-200
                       {"booking_id" booking-id}))]
     (validate-req booking-id ::specs/booking-id on-success)))
 
-(defn- cancel-booking [db-spec booking-id]
-  (worker/add-cancel-ticket-request-to-queue {:booking-id booking-id}))
+(defn- cancel-booking [db-spec message-queue booking-id]
+  (worker/add-cancel-ticket-request-to-queue message-queue {:booking-id booking-id}))
 
-(defn cancel-booking-handler [db-spec request]
+(defn cancel-booking-handler [db-spec message-queue request]
   (let [booking-id (get-in request [:route-params :booking-id])
         on-success (fn []
-                     (cancel-booking db-spec booking-id)
+                     (cancel-booking db-spec message-queue booking-id)
                      (respond-200 {"booking_id" booking-id}))]
     (validate-req booking-id ::specs/booking-id on-success)))
 
@@ -164,5 +166,5 @@
 (defn get-tickets-by-booking-id-handler [db-spec request]
   (let [booking-id (get-in request [:route-params :booking-id])
         on-success (fn []
-            (respond-200 (get-tickets-by-booking-id db-spec booking-id)))]
+                     (respond-200 (get-tickets-by-booking-id db-spec booking-id)))]
     (validate-req booking-id ::specs/booking-id on-success)))

swift-ticketing/src/swift_ticketing/worker.clj

@@ -7,38 +7,42 @@
             [swift-ticketing.redis :as redis])
   (:import [java.time Instant Duration]))
 
-(def ticket-queue (async/chan))
-
 (def reserve-event "RESERVE")
 (def book-event "BOOK")
 (def cancel-event "CANCEL")
 
-(defn add-ticket-request-to-queue [request]
-  (async/go (async/>! ticket-queue request)))
+(defn add-ticket-request-to-queue [message-queue request]
+  (async/go (async/>! message-queue request)))
 
-(defn add-reserve-ticket-request-to-queue [request]
+(defn add-reserve-ticket-request-to-queue [message-queue request]
   (let [event-type reserve-event
         booking-id (:booking-id request)
         ticket-ids (:ticket-ids request)
         ticket-type-id (:ticket-type-id request)
         quantity (:quantity request)]
-    (add-ticket-request-to-queue {:event event-type
-                                  :data {:booking-id booking-id
-                                         :ticket-ids ticket-ids
-                                         :ticket-type-id ticket-type-id
-                                         :quantity quantity}})))
+    (add-ticket-request-to-queue
+     message-queue
+     {:event event-type
+      :data {:booking-id booking-id
+             :ticket-ids ticket-ids
+             :ticket-type-id ticket-type-id
+             :quantity quantity}})))
 
-(defn add-book-ticket-request-to-queue [request]
+(defn add-book-ticket-request-to-queue [message-queue request]
   (let [event-type book-event
         booking-id (:booking-id request)]
-    (add-ticket-request-to-queue {:event event-type
-                                  :data {:booking-id booking-id}})))
+    (add-ticket-request-to-queue
+     message-queue
+     {:event event-type
+      :data {:booking-id booking-id}})))
 
-(defn add-cancel-ticket-request-to-queue [request]
+(defn add-cancel-ticket-request-to-queue [message-queue request]
   (let [event-type cancel-event
         booking-id (:booking-id request)]
-    (add-ticket-request-to-queue {:event event-type
-                                  :data {:booking-id booking-id}})))
+    (add-ticket-request-to-queue
+     message-queue
+     {:event event-type
+      :data {:booking-id booking-id}})))
 
 (defn- select-unbooked-tickets-with-db-lock [tx ticket-ids ticket-type-id ticket-quantity]
   (let [selected-rows (ticket/lock-unbooked-tickets tx ticket-ids ticket-type-id ticket-quantity)
@@ -117,11 +121,11 @@
       (ticket/cancel-tickets tx selected-ticket-ids)
       (booking/update-booking-status tx booking-id booking/CANCELED))))
 
-(defn process-ticket-requests [worker-id db-spec redis-opts]
+(defn process-ticket-requests [worker-id message-queue db-spec redis-opts]
   (async/go
     (while true
       (try
-        (let [request (async/<! ticket-queue)
+        (let [request (async/<! message-queue)
               event-type (:event request)]
           (println "Got Message:")
           (println request)