mbaechler commented on a change in pull request #808:
URL: https://github.com/apache/james-project/pull/808#discussion_r774923294
##########
File path:
server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/FilterStage.scala
##########
@@ -0,0 +1,63 @@
+package org.apache.james.queue.pulsar
+
+import akka.actor.{Actor, ActorLogging, Props}
+import com.sksamuel.pulsar4s.akka.streams.CommittableMessage
+import com.sksamuel.pulsar4s.{ConsumerMessage, SequenceId}
+
+import scala.math.Ordered.orderingToOrdered
+
+private[pulsar] class FilterStage extends Actor with ActorLogging {
+ private var filters = Set.empty[Filter]
+ private val name = self.path.name
+
+ def receive: Receive = {
+
+ case filter: Filter =>
+ registerFilter(filter)
+ log.debug(s"new filter registered, active filters : $filters ")
+
+ // processing mail
+ case (metadata: MailMetadata, cm: CommittableMessage[_]) =>
+ val sequenceId = cm.message.sequenceId
+ log.debug(s"filtering mail with active filters : $filters , sequence:
${sequenceId} metadata : $metadata")
+ if (shouldBeFiltered(metadata, sequenceId)) {
+ log.debug(s"message filtered : sequence: ${sequenceId} metadata :
$metadata")
+ sender() ! (None, cm)
+ } else {
+ log.debug(s"message not filtered : sequence: ${sequenceId} metadata :
$metadata")
+ sender() ! (Some(metadata), cm)
+ }
Review comment:
I guess it's what debug is for: helping debug complex stuff. We had to
put them in place and decided this code is hard enough to grasp to let them in
place, the next developers will probably need them (:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
