Merge pull request #15 from mahdibohloul/kpring-mediatr/feature/notif…

…ication-exception-handler

Notification error handling

README.md

@@ -54,7 +54,7 @@ class MediatorConfiguration(
 
 ## Features
 
-**Kpring Mediator** offers three types of features:
+**Kpring Mediator** offers four types of features:
 
 #### Request:
 
@@ -70,6 +70,18 @@ Like [Request](#Request:), you send a command to anyone who can handle the comma
 You publish a notification and the relevant handlers receive the notification and do something about it. It is useful when something special is happening in your system, and you want to do separate tasks in parallel afterwards. For example, when a shipment is cancelled, you may want to send an email to the order owner and the driver and vendor of the shipment that is not related to your main process, to do this you can publish a notification.
 Also, you can specify a coroutine dispatcher to run the notification handler on. it will become handy when you have many notification handlers, and you want to run some of them on another thread.
 
+#### Notification Exception Handler:
+
+Catch thrown exceptions in your notification handlers and handle them in a special way. For example, you may want to log the exception and email to the system administrator.
+You can also specify a coroutine dispatcher to run the notification exception handler on.
+To enable this feature, you need to add the following method to your factory configuration:
+```kotlin
+@Bean
+fun factory(): Factory {
+    return DefaultFactory(applicationContext).enableNotificationExceptionHandling()
+}
+```
+
 ## Usage
 
 > ###### Do not forget to use the **component** annotation with your handler. it helps the mediator to register your handlers.
@@ -136,6 +148,20 @@ class SendEmailToVendorWhenOrderCancelledNotificationHandler : NotificationHandl
 }
 ```
 
+#### Notification Exception Handler usage
+```kotlin
+@Component
+class OrderCancellationSendingEmailExceptionHandler: NotificationExceptionHandler<OrderCancellationNotification, EmailServiceException> {
+    override suspend fun handle(notification: OrderCancellationNotification, exception: EmailServiceException) {
+        //Handle exception
+    }
+
+    override fun getCoroutineDispatcher(): CoroutineDispatcher {
+        return Dispatchers.IO
+    }
+}
+```
+
 ### Use in reactive style
 
 Add the [Kotlinx coroutines reactor](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactor/index.html) to your project.
@@ -160,9 +186,8 @@ class ProductService(
 
 There are 5 different exceptions: **NoRequestHandlerException**, **DuplicateRequestHandlerRegistrationException**, **NoNotificationHandlersException**, **NoCommandHandlerException**, **DuplicateCommandHandlerRegistrationException**, all of each are inherited from **KpringMediatorException**.
 
-Exceptions that occur in *request handler*s and *command handler*s were propagated in the parent and canceled the process, but if an exception occurs in one of the *notification handler*s, it is ignored and the other *notification handler*s continue to operate.
-
-> I am currently working on a system for logging exceptions that occur in specific notification handlers, and I hope to improve this in later versions.
+Exceptions that occur in *request handler*s and *command handler*s were propagated in the parent and canceled the process, but if an exception occurs in one of the *notification handler*s, it is ignored if there is no *notification exception handler*s found and the other *notification handler*s continue to operate.
 
+## Contribution
 ***If you can improve this project, do not hesitate to contribute with me. I'm waiting for your merge requests with open arms.***
 

build.gradle.kts

@@ -9,7 +9,7 @@ plugins {
 }
 
 group = "io.github.mahdibohloul"
-version = "1.0.1"
+version = "1.1.0"
 java.sourceCompatibility = JavaVersion.VERSION_11
 
 repositories {
@@ -22,9 +22,9 @@ dependencies {
   implementation(group = "javax.validation", name = "validation-api", version = "2.0.1.Final")
   implementation("org.jetbrains.kotlin:kotlin-reflect")
   implementation("org.jetbrains.kotlin:kotlin-stdlib-jdk8")
-  implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.5.2-native-mt")
+  implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.6.0-native-mt")
   implementation("junit:junit:4.13.2")
-  testImplementation("org.springframework.boot:spring-boot-starter-test:2.5.5")
+  testImplementation("org.springframework.boot:spring-boot-starter-test:2.6.1")
   testImplementation("org.jetbrains.kotlin:kotlin-test:1.5.31")
   testImplementation(group = "org.mockito", name = "mockito-core", version = "2.23.4")
 }

src/main/kotlin/io/mahdibohloul/kpringmediator/core/Factory.kt

@@ -1,5 +1,7 @@
 package io.mahdibohloul.kpringmediator.core
 
+import kotlin.reflect.KClass
+
 /**
  * A factory for handlers for messages that can be sent or published in Kpring MediatR
  *
@@ -15,8 +17,8 @@ interface Factory {
      * @return The [RequestHandler] for the request
      * @throws NoRequestHandlerException When there is not a RequestHandler available for the request
      */
-    fun <TRequest : Request<TResponse>, TResponse> get(requestClass: Class<out TRequest>):
-            RequestHandler<TRequest, TResponse>
+    fun <TRequest : Request<TResponse>, TResponse> getRequestHandler(requestClass: KClass<out TRequest>):
+        RequestHandler<TRequest, TResponse>
 
     /**
      * Retrieves all the NotificationHandlers for the provided notification type. If no NotificationHandlers are
@@ -26,7 +28,7 @@ interface Factory {
      * @return Set of [NotificationHandler]s for the notificationClass
      * @throws NoNotificationHandlersException When there are no EventHandlers available
      */
-    fun <TNotification : Notification> get(notificationClass: Class<out TNotification>): Set<NotificationHandler<TNotification>>
+    fun <TNotification : Notification> getNotificationHandlers(notificationClass: KClass<out TNotification>): Set<NotificationHandler<TNotification>>
 
     /**
      * Retrieves a CommandHandler for the provided type. If no CommandHandler
@@ -36,5 +38,19 @@ interface Factory {
      * @return The [CommandHandler] for the command
      * @throws NoCommandHandlerException When there isn't a CommandHandler available
      */
-    fun <TCommand : Command> get(commandClass: Class<out TCommand>): CommandHandler<TCommand>
+    fun <TCommand : Command> getCommandHandler(commandClass: KClass<out TCommand>): CommandHandler<TCommand>
+
+    /**
+     * Retrieves all the NotificationExceptionHandlers for the provided notification and exception type. If no NotificationExceptionHandlers are
+     * registered to handle the notification and exception type provided an empty set will be returned.
+     *
+     * @author Mahdi Bohloul
+     * @param notificationClass The type of the event
+     * @param exceptionClass The type of the exception
+     * @return Set of [NotificationExceptionHandler]s for the notificationClass and exceptionClass
+     */
+    fun <TNotification : Notification, TException : Exception> getNotificationExceptionHandlers(
+        notificationClass: KClass<out TNotification>, exceptionClass: KClass<out TException>
+    ):
+        Set<NotificationExceptionHandler<TNotification, TException>>
 }

src/main/kotlin/io/mahdibohloul/kpringmediator/core/FactoryExtensions.kt

@@ -0,0 +1,8 @@
+package io.mahdibohloul.kpringmediator.core
+
+import io.mahdibohloul.kpringmediator.infrastructure.DefaultFactory
+
+fun DefaultFactory.enableNotificationExceptionHandling(): DefaultFactory {
+    this.handleNotificationExceptions = true
+    return this
+}

src/main/kotlin/io/mahdibohloul/kpringmediator/core/NotificationExceptionHandler.kt

@@ -0,0 +1,20 @@
+package io.mahdibohloul.kpringmediator.core
+
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.Dispatchers
+
+/**
+ * This class is responsible for handling exceptions thrown during the handling published notification.
+ * @author Mahdi Bohloul
+ * @param TException the type of the exception that will be handled
+ * @param TNotification the type of the notification that will be handled
+ */
+interface NotificationExceptionHandler<
+    in TNotification : Notification,
+    in TException : Exception> {
+    suspend fun handle(notification: TNotification, exception: TException)
+
+    fun getCoroutineDispatcher(): CoroutineDispatcher {
+        return Dispatchers.Default
+    }
+}

src/main/kotlin/io/mahdibohloul/kpringmediator/infrastructure/DefaultFactory.kt

@@ -1,6 +1,7 @@
 package io.mahdibohloul.kpringmediator.infrastructure
 
 import io.mahdibohloul.kpringmediator.core.*
+import kotlin.reflect.KClass
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import org.springframework.context.ApplicationContext
@@ -13,47 +14,78 @@ class DefaultFactory(
     private val applicationContext: ApplicationContext
 ) : Factory {
 
-    private val registeredRequestHandlers: MutableMap<Class<out Request<*>>, RequestHandlerProvider<*>> = HashMap()
-    private val registeredNotificationHandlers: MutableMap<Class<out Notification>, MutableSet<NotificationHandlerProvider<*>>> =
+    private val registeredRequestHandlers: MutableMap<KClass<out Request<*>>, RequestHandlerProvider<*>> = HashMap()
+    private val registeredNotificationHandlers: MutableMap<KClass<out Notification>, MutableSet<NotificationHandlerProvider<*>>> =
         HashMap()
-    private val registeredCommandHandler: MutableMap<Class<out Command>, CommandHandlerProvider<*>> = HashMap()
+    private val registeredNotificationExceptionHandlers:
+        MutableMap<
+            KClass<out Notification>,
+            MutableMap<KClass<out Exception>,
+                MutableSet<NotificationExceptionHandlerProvider<*>>>> =
+        HashMap()
+    private val registeredCommandHandler: MutableMap<KClass<out Command>, CommandHandlerProvider<*>> = HashMap()
     private var initialized: Boolean = false
 
+    internal var handleNotificationExceptions: Boolean = false
 
-    override fun <TRequest : Request<TResponse>, TResponse> get(requestClass: Class<out TRequest>): RequestHandler<TRequest, TResponse> {
+    override fun <TRequest : Request<TResponse>, TResponse> getRequestHandler(
+        requestClass: KClass<out TRequest>
+    ):
+        RequestHandler<TRequest, TResponse> {
         if (!initialized) {
             initializeHandlers()
         }
         registeredRequestHandlers[requestClass]?.let {
             return it.handler as RequestHandler<TRequest, TResponse>
         }
-            ?: throw NoRequestHandlerException("No RequestHandler is registered to handle request of type ${requestClass.canonicalName}")
+            ?: throw NoRequestHandlerException("No RequestHandler is registered to handle request of type ${requestClass.simpleName}")
     }
 
-    override fun <TNotification : Notification> get(notificationClass: Class<out TNotification>): Set<NotificationHandler<TNotification>> {
+    override fun <TNotification : Notification> getNotificationHandlers(notificationClass: KClass<out TNotification>): Set<NotificationHandler<TNotification>> {
         if (!initialized) {
             initializeHandlers()
         }
         val handlers = mutableSetOf<NotificationHandler<TNotification>>()
         registeredNotificationHandlers[notificationClass]?.let {
-            for (provider in it) {
+            it.forEach { provider ->
                 val handler = provider.handler as NotificationHandler<TNotification>
                 handlers.add(handler)
             }
         }
-            ?: throw NoNotificationHandlersException("No NotificationHandlers are registered to receive notification of type ${notificationClass.canonicalName}")
+            ?: throw NoNotificationHandlersException("No NotificationHandlers are registered to receive notification of type ${notificationClass.simpleName}")
         return handlers
     }
 
-    override fun <TCommand : Command> get(commandClass: Class<out TCommand>): CommandHandler<TCommand> {
+    override fun <TCommand : Command> getCommandHandler(commandClass: KClass<out TCommand>): CommandHandler<TCommand> {
         if (!initialized) {
             initializeHandlers()
         }
         registeredCommandHandler[commandClass]?.let { provider ->
             return provider.handler as CommandHandler<TCommand>
         }
-            ?: throw NoCommandHandlerException("No CommandHandler is registered to handle request of type ${commandClass.canonicalName}")
+            ?: throw NoCommandHandlerException("No CommandHandler is registered to handle request of type ${commandClass.simpleName}")
+    }
 
+    override fun <TNotification : Notification, TNotificationException : Exception> getNotificationExceptionHandlers(
+        notificationClass: KClass<out TNotification>,
+        exceptionClass: KClass<out TNotificationException>
+    ): Set<NotificationExceptionHandler<TNotification, TNotificationException>> {
+        if (!handleNotificationExceptions) {
+            logger.warn("Notification exception handling is disabled")
+            return emptySet()
+        }
+        if (!initialized) {
+            initializeHandlers()
+        }
+        val handlers = mutableSetOf<NotificationExceptionHandler<TNotification, TNotificationException>>()
+        registeredNotificationExceptionHandlers[notificationClass]?.get(exceptionClass)?.let {
+            it.forEach { provider ->
+                val handler = provider.handler as NotificationExceptionHandler<TNotification, TNotificationException>
+                handlers.add(handler)
+            }
+        }
+            ?: logger.warn("No NotificationExceptionHandlers are registered to receive notification exception of type ${notificationClass.simpleName}")
+        return handlers
     }
 
     private fun initializeHandlers() {
@@ -65,6 +97,9 @@ class DefaultFactory(
                     .forEach { registerNotificationHandler(it) }
                 applicationContext.getBeanNamesForType(CommandHandler::class.java)
                     .forEach { registerCommandHandler(it) }
+                if (handleNotificationExceptions)
+                    applicationContext.getBeanNamesForType(NotificationExceptionHandler::class.java)
+                        .forEach { registerNotificationExceptionHandler(it) }
                 initialized = true
             }
         }
@@ -75,10 +110,10 @@ class DefaultFactory(
         val handler: RequestHandler<*, *> = applicationContext.getBean(requestHandlerName) as RequestHandler<*, *>
         val generics = GenericTypeResolver.resolveTypeArguments(handler::class.java, RequestHandler::class.java)
         generics?.let {
-            val requestType = it[0] as Class<out Request<*>>
+            val requestType = (it[0] as Class<out Request<*>>).kotlin
             if (registeredRequestHandlers.contains(requestType)) {
                 throw DuplicateRequestHandlerRegistrationException(
-                    "${requestType.canonicalName} already has a registered handler. Each request must have a single request handler"
+                    "${requestType.simpleName} already has a registered handler. Each request must have a single request handler"
                 )
             }
 
@@ -95,7 +130,7 @@ class DefaultFactory(
         val generics =
             GenericTypeResolver.resolveTypeArguments(notificationHandler::class.java, NotificationHandler::class.java)
         generics?.let {
-            val notificationType = it[0] as Class<out Notification>
+            val notificationType = (it[0] as Class<out Notification>).kotlin
             val eventProvider = NotificationHandlerProvider(applicationContext, notificationHandler::class)
             registeredNotificationHandlers[notificationType]?.add(eventProvider) ?: kotlin.run {
                 registeredNotificationHandlers[notificationType] = mutableSetOf(eventProvider)
@@ -104,15 +139,40 @@ class DefaultFactory(
         }
     }
 
+    private fun registerNotificationExceptionHandler(notificationExceptionHandlerName: String) {
+        logger.debug("Registering NotificationExceptionHandler with name $notificationExceptionHandlerName")
+        val notificationExceptionHandler: NotificationExceptionHandler<*, *> =
+            applicationContext.getBean(notificationExceptionHandlerName) as NotificationExceptionHandler<*, *>
+        val generics =
+            GenericTypeResolver.resolveTypeArguments(
+                notificationExceptionHandler::class.java,
+                NotificationExceptionHandler::class.java
+            )
+        generics?.let {
+            val notificationType = (it[0] as Class<out Notification>).kotlin
+            val exceptionType = (it[1] as Class<out Exception>).kotlin
+            val eventProvider =
+                NotificationExceptionHandlerProvider(applicationContext, notificationExceptionHandler::class)
+            registeredNotificationExceptionHandlers[notificationType]?.let { exceptionTypes ->
+                exceptionTypes[exceptionType]?.add(eventProvider) ?: kotlin.run {
+                    exceptionTypes[exceptionType] = mutableSetOf(eventProvider)
+                }
+            } ?: kotlin.run {
+                registeredNotificationExceptionHandlers[notificationType] =
+                    mutableMapOf(exceptionType to mutableSetOf(eventProvider))
+            }
+        }
+    }
+
     private fun registerCommandHandler(commandHandlerName: String) {
         logger.debug("Registering CommandHandler with name $commandHandlerName")
         val handler: CommandHandler<*> = applicationContext.getBean(commandHandlerName) as CommandHandler<*>
         val generics = GenericTypeResolver.resolveTypeArguments(handler::class.java, CommandHandler::class.java)
         generics?.let {
-            val requestType = it[0] as Class<out Command>
+            val requestType = (it[0] as Class<out Command>).kotlin
             if (registeredCommandHandler.contains(requestType)) {
                 throw DuplicateCommandHandlerRegistrationException(
-                    "${requestType.canonicalName} already has a registered handler. Each request must have a single request handler"
+                    "${requestType.simpleName} already has a registered handler. Each request must have a single request handler"
                 )
             }