core:service: use async:task for concurrent / parallel processing

config.lisp

@@ -44,7 +44,7 @@
              (dotenv-val (if dotenv-data (gethash key dotenv-data))))
         (if env-val
           (setf (gethash sl data) env-val)
-          (progn 
+          (when dotenv-val 
             (setf (uiop:getenv key) dotenv-val)
             (setf (gethash sl data) dotenv-val)))))))
 

core/core.lisp

@@ -74,14 +74,13 @@
    (services :reader services :initform (make-hash-table))))
 
 (defclass service (context)
-  ((task :accessor task :initform nil)
-   (mailbox :reader mailbox :initform (lpq:make-queue))))
+  ((task :accessor task :initform nil)))
 
-(defgeneric do-start (ctx)
-  (:method ((ctx context)))
-  (:method ((ctx service)) 
-    (let ((ch (lp:make-channel))) 
-      (lp:submit-task ch (do-listen ctx)))))
+(defgeneric do-start (ctx &key wait)
+  (:method ((ctx context) &key wait))
+  (:method ((ctx service) &key wait) 
+    (setf (task ctx) (async:make-task :handle-message #'handle-message))
+    (async:start (task ctx :wait wait))))
 
 (defgeneric do-listen (ctx)
   (:method ((ctx service))
@@ -97,14 +96,11 @@
   (:method ((rcvr base-context) msg)
     (handle-message rcvr msg))
   (:method ((rcvr service) msg)
-    (lpq:push-queue msg (mailbox rcvr))))
+    (async:send (task rcvr) msg)))
 
  (defun default-setup (cfg &optional (cls 'context) &rest args &key &allow-other-keys)
   (apply #'make-instance cls :config cfg :name (config:name cfg) args))
 
-;;; setup-service
-        ;(setf (task child) (async:make-task :handle-message #'handle-message))
-
 (defun find-service (name &optional (parent *root*))
   (with-slots (services) parent
     (when services
@@ -117,6 +113,8 @@
     (add-service *root* c)))
 
 (defun shutdown ()
+  (if (task *root*) 
+    (async:stop (task *root*)))
   (dolist (ctx (alx:hash-table-values (services *root*)))
     (funcall (config:shutdown (config ctx)) ctx)))
  
@@ -124,7 +122,8 @@
    (unwind-protect
      (progn
        (setup-services cfg)
-       (do-listen *root*))
+       ;(do-listen *root*)
+       (do-start *root* :wait t))
      (shutdown)))
 
 (defun add-action (ctx pat hdlr)

util/async.lisp

@@ -52,12 +52,14 @@
       (setf (mailbox tsk) (lpq:make-queue)))
     tsk))
 
-(defun start (tsk)
+(defun start (tsk &key (wait nil))
   (when (eq (status tsk) :running)
     (util:lgw "task already running" (taskid tsk))
     (return-from start))
   (setf (status tsk) :running)
-  (lp:submit-task (channel tsk) (job tsk)))
+  (if wait 
+    (funcall (job tsk))
+    (lp:submit-task (channel tsk) (job tsk))))
 
 (defun stop (tsk &key (wait t))
   (when (mailbox tsk)