Support :mbox-type for pinned dispatchers

[svetlyak40wt]

Also I've added an ability to pass a function instead of class-name. This is useful, when you want to create an message box instance with arguments other than supported by base implementation.

Usage example:

(defclass deduplication-message-box (sento.messageb:message-box/bt)
  ((deduplicatep :initarg :deduplicatep
                 :type function
                 :reader message-box-deduplicate-p)))


(defgeneric search-in-queue (queue predicate)
  (:method ((queue queue-bounded) predicate)
    (error "Search in cl-speedy-queue:queue is not supported yet."))

  (:method ((queue queue-unbounded) predicate)
    (let ((inner-queue (slot-value queue 'sento.queue::queue))
          (lock (slot-value queue 'sento.queue::lock)))
      (check-type inner-queue sento.queue::queue)

      (bt2:with-lock-held (lock)
        (let ((head (sento.queue::queue-head inner-queue))
              (tail (sento.queue::queue-tail inner-queue)))

          (or (find-if predicate tail
                       :key #'sento.messageb::message-item/bt-message)
              (find-if predicate (reverse head)
                       :key #'sento.messageb::message-item/bt-message)))))))

(defmethod submit :around ((self deduplication-message-box) message withreply-p time-out handler-fun-args)
  (cond
    (withreply-p
     ;; We can't deduplicate messages for which caller is waiting for the reply:
     (call-next-method))
    ((null (search-in-queue (slot-value self 'sento.messageb::queue)
                            (alexandria:curry (message-box-deduplicate-p self)
                                              message)))
     (call-next-method))
    (t
     (log:warn "Message was deduplicated" message))))
     
     
;;; Then, when creating an actor:

(defun deduplicate-message-with-site-config (message-to-check message-from-queue)
  "Returns T if message-to-check is equal to message-from-queue"
  (and (listp message-to-check)
       (listp message-from-queue)
       (eql (first message-to-check)
            (first message-from-queue))
       (equal (bot/config::chat-id (second message-to-check))
              (bot/config::chat-id (second message-from-queue)))))


(defun create-mbox-with-site-actions-deduplication (&key (max-queue-size nil) &allow-other-keys)
  (when max-queue-size
    (error "Deduplication queue does not work with bounded queues yet."))

  (make-instance 'deduplication-message-box
                 :deduplicatep #'deduplicate-message-with-site-config))


(defun create-site-updater (actors-system)
  (sento.actor-context:actor-of actors-system
                                :receive #'do-update
                                :dispatcher :pinned
                                :name "site-updater"
                                :mbox-type #'create-mbox-with-site-actions-deduplication))

[mdbergmann]

This is cool.
Can you please add a test case similar to actor-of--custom-dispatcher-with-custom-mbox-type in actor-context-test.lisp showing that the wiring works as it should.

[mdbergmann]

I think it would be good to check if the generated instance is at least message-box-base.

[mdbergmann]

OK, looks look. Thanks.

[mdbergmann]

That's a good point:

;; We can't deduplicate messages for which caller is waiting for the reply

Actually, withreply-p is just used for the synchronous reply.
Any async reply, should there be multiple requests, will probably also leave one requester waiting.
But that's up to the user of custom mailbox.

[svetlyak40wt]

I don't understand whether something else needs to be changed in this regard, or if everything is already fine as is?

[mdbergmann]

I don't understand whether something else needs to be changed in this regard, or if everything is already fine as is?

It's fine.
I'll merge...