Rename NotificationConsumer

This commit is contained in:
2021-02-03 01:49:12 +01:00
parent 3b3a71f6eb
commit 3580c33891
4 changed files with 62 additions and 64 deletions

View File

@@ -33,7 +33,7 @@ import fr.dcproject.component.vote.routes.installVoteRoutes
import fr.dcproject.component.vote.voteKoinModule import fr.dcproject.component.vote.voteKoinModule
import fr.dcproject.component.workgroup.routes.installWorkgroupRoutes import fr.dcproject.component.workgroup.routes.installWorkgroupRoutes
import fr.dcproject.component.workgroup.workgroupKoinModule import fr.dcproject.component.workgroup.workgroupKoinModule
import fr.dcproject.notification.EventNotification import fr.dcproject.notification.NotificationConsumer
import fr.dcproject.routes.definition import fr.dcproject.routes.definition
import fr.dcproject.routes.notificationArticle import fr.dcproject.routes.notificationArticle
import fr.dcproject.security.AccessDeniedException import fr.dcproject.security.AccessDeniedException
@@ -122,7 +122,7 @@ fun Application.module(env: Env = PROD) {
masking = false masking = false
} }
EventNotification(get(), get(), get(), get(), get(), Configuration.exchangeNotificationName).config() NotificationConsumer(get(), get(), get(), get(), get(), Configuration.exchangeNotificationName).config()
install(Authentication, jwtInstallation(get())) install(Authentication, jwtInstallation(get()))

View File

@@ -0,0 +1,51 @@
package fr.dcproject.notification
import com.fasterxml.jackson.databind.DeserializationFeature
import com.fasterxml.jackson.databind.PropertyNamingStrategies
import com.fasterxml.jackson.databind.SerializationFeature
import com.fasterxml.jackson.databind.module.SimpleModule
import com.fasterxml.jackson.datatype.joda.JodaModule
import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
import com.fasterxml.jackson.module.kotlin.readValue
import fr.dcproject.component.article.ArticleForView
import fr.postgresjson.entity.UuidEntity
import org.joda.time.DateTime
import kotlin.random.Random
open class Notification(
val type: String,
val createdAt: DateTime = DateTime.now()
) {
val id: Double = randId(createdAt.millis)
private fun randId(time: Long): Double {
return (time.toString() + Random.nextInt(1000, 9999).toString()).toDouble()
}
override fun toString(): String = mapper.writeValueAsString(this) ?: error("Unable to serialize notification")
fun toByteArray() = toString().toByteArray()
companion object {
val mapper = jacksonObjectMapper().apply {
registerModule(SimpleModule())
propertyNamingStrategy = PropertyNamingStrategies.SNAKE_CASE
registerModule(JodaModule())
disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)
configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
}
inline fun <reified T : Notification> fromString(raw: String): T = mapper.readValue(raw)
}
}
open class EntityNotification(
val target: UuidEntity,
type: String,
val action: String
) : Notification(type)
class ArticleUpdateNotification(
target: ArticleForView
) : EntityNotification(target, "article", "update")

View File

@@ -1,76 +1,27 @@
package fr.dcproject.notification package fr.dcproject.notification
import com.fasterxml.jackson.databind.DeserializationFeature import com.rabbitmq.client.AMQP.BasicProperties
import com.fasterxml.jackson.databind.PropertyNamingStrategies
import com.fasterxml.jackson.databind.SerializationFeature
import com.fasterxml.jackson.databind.module.SimpleModule
import com.fasterxml.jackson.datatype.joda.JodaModule
import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
import com.fasterxml.jackson.module.kotlin.readValue
import com.rabbitmq.client.AMQP
import com.rabbitmq.client.BuiltinExchangeType.DIRECT import com.rabbitmq.client.BuiltinExchangeType.DIRECT
import com.rabbitmq.client.ConnectionFactory import com.rabbitmq.client.ConnectionFactory
import com.rabbitmq.client.Consumer import com.rabbitmq.client.Consumer
import com.rabbitmq.client.DefaultConsumer import com.rabbitmq.client.DefaultConsumer
import com.rabbitmq.client.Envelope import com.rabbitmq.client.Envelope
import fr.dcproject.common.entity.TargetRef import fr.dcproject.common.entity.TargetRef
import fr.dcproject.component.article.ArticleForView
import fr.dcproject.component.citizen.CitizenRef import fr.dcproject.component.citizen.CitizenRef
import fr.dcproject.component.follow.FollowArticleRepository import fr.dcproject.component.follow.FollowArticleRepository
import fr.dcproject.component.follow.FollowConstitutionRepository import fr.dcproject.component.follow.FollowConstitutionRepository
import fr.dcproject.component.follow.FollowSimple import fr.dcproject.component.follow.FollowSimple
import fr.dcproject.notification.publisher.Publisher
import fr.dcproject.messages.NotificationEmailSender import fr.dcproject.messages.NotificationEmailSender
import fr.postgresjson.entity.UuidEntity import fr.dcproject.notification.publisher.Publisher
import io.ktor.utils.io.errors.IOException import io.ktor.utils.io.errors.IOException
import io.lettuce.core.RedisClient import io.lettuce.core.RedisClient
import io.lettuce.core.api.async.RedisAsyncCommands import io.lettuce.core.api.async.RedisAsyncCommands
import kotlinx.coroutines.flow.collect import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.runBlocking import kotlinx.coroutines.runBlocking
import org.joda.time.DateTime
import org.slf4j.Logger import org.slf4j.Logger
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import kotlin.random.Random
open class Notification( class NotificationConsumer(
val type: String,
val createdAt: DateTime = DateTime.now()
) {
val id: Double = randId(createdAt.millis)
private fun randId(time: Long): Double {
return (time.toString() + Random.nextInt(1000, 9999).toString()).toDouble()
}
override fun toString(): String = mapper.writeValueAsString(this) ?: error("Unable to serialize notification")
fun toByteArray() = toString().toByteArray()
companion object {
val mapper = jacksonObjectMapper().apply {
registerModule(SimpleModule())
propertyNamingStrategy = PropertyNamingStrategies.SNAKE_CASE
registerModule(JodaModule())
disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)
configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
}
inline fun <reified T : Notification> fromString(raw: String): T = mapper.readValue(raw)
}
}
open class EntityNotification(
val target: UuidEntity,
type: String,
val action: String
) : Notification(type)
class ArticleUpdateNotification(
target: ArticleForView
) : EntityNotification(target, "article", "update")
class EventNotification(
private val rabbitFactory: ConnectionFactory, private val rabbitFactory: ConnectionFactory,
private val redisClient: RedisClient, private val redisClient: RedisClient,
private val followConstitutionRepo: FollowConstitutionRepository, private val followConstitutionRepo: FollowConstitutionRepository,
@@ -101,7 +52,7 @@ class EventNotification(
override fun handleDelivery( override fun handleDelivery(
consumerTag: String, consumerTag: String,
envelope: Envelope, envelope: Envelope,
properties: AMQP.BasicProperties, properties: BasicProperties,
body: ByteArray body: ByteArray
) = runBlocking { ) = runBlocking {
followersFromMessage(body) { followersFromMessage(body) {
@@ -121,7 +72,7 @@ class EventNotification(
override fun handleDelivery( override fun handleDelivery(
consumerTag: String, consumerTag: String,
envelope: Envelope, envelope: Envelope,
properties: AMQP.BasicProperties, properties: BasicProperties,
body: ByteArray body: ByteArray
) { ) {
runBlocking { runBlocking {

View File

@@ -8,30 +8,26 @@ import fr.dcproject.component.citizen.CitizenRef
import fr.dcproject.component.follow.FollowArticleRepository import fr.dcproject.component.follow.FollowArticleRepository
import fr.dcproject.component.follow.FollowSimple import fr.dcproject.component.follow.FollowSimple
import fr.dcproject.notification.ArticleUpdateNotification import fr.dcproject.notification.ArticleUpdateNotification
import fr.dcproject.notification.EventNotification import fr.dcproject.notification.NotificationConsumer
import fr.dcproject.notification.publisher.Publisher import fr.dcproject.notification.publisher.Publisher
import fr.dcproject.messages.NotificationEmailSender import fr.dcproject.messages.NotificationEmailSender
import io.ktor.locations.KtorExperimentalLocationsAPI import io.ktor.locations.KtorExperimentalLocationsAPI
import io.ktor.util.KtorExperimentalAPI import io.ktor.util.KtorExperimentalAPI
import io.lettuce.core.RedisClient import io.lettuce.core.RedisClient
import io.lettuce.core.api.async.RedisAsyncCommands
import io.mockk.every import io.mockk.every
import io.mockk.mockk import io.mockk.mockk
import io.mockk.spyk import io.mockk.spyk
import io.mockk.verify import io.mockk.verify
import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.InternalCoroutinesApi import kotlinx.coroutines.InternalCoroutinesApi
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking import kotlinx.coroutines.runBlocking
import org.junit.jupiter.api.Tag import org.junit.jupiter.api.Tag
import org.junit.jupiter.api.Test import org.junit.jupiter.api.Test
import org.junit.jupiter.api.TestInstance import org.junit.jupiter.api.TestInstance
import org.koin.test.AutoCloseKoinTest
import org.koin.test.KoinTest
@TestInstance(TestInstance.Lifecycle.PER_CLASS) @TestInstance(TestInstance.Lifecycle.PER_CLASS)
class EventNotificationTest { class NotificationConsumerTest {
@InternalCoroutinesApi @InternalCoroutinesApi
@KtorExperimentalLocationsAPI @KtorExperimentalLocationsAPI
@KtorExperimentalAPI @KtorExperimentalAPI
@@ -68,7 +64,7 @@ class EventNotificationTest {
} }
/* Config consumer */ /* Config consumer */
EventNotification( NotificationConsumer(
rabbitFactory = rabbitFactory, rabbitFactory = rabbitFactory,
redisClient = redisClient, redisClient = redisClient,
followArticleRepo = followArticleRepo, followArticleRepo = followArticleRepo,