feat: transaction batcher module

executions/transaction-batcher/build.gradle.kts

@@ -0,0 +1,19 @@
+description = "Transaction Batcher"
+
+val reactiveStreamsVersion: String by project
+val junitVersion: String by project
+val slf4jVersion: String by project
+val mockkVersion: String by project
+val reactorVersion: String by project
+val reactorExtensionsVersion: String by project
+
+dependencies {
+    implementation("org.reactivestreams:reactive-streams:$reactiveStreamsVersion")
+    implementation("org.slf4j:slf4j-api:$slf4jVersion")
+    testImplementation("org.junit.jupiter:junit-jupiter-api:$junitVersion")
+    testImplementation("org.junit.jupiter:junit-jupiter-engine:$junitVersion")
+    testImplementation("io.mockk:mockk:$mockkVersion")
+    testImplementation("io.projectreactor:reactor-core:$reactorVersion")
+    testImplementation("io.projectreactor.kotlin:reactor-kotlin-extensions:$reactorExtensionsVersion")
+    testImplementation("io.projectreactor:reactor-test:$reactorVersion")
+}

executions/transaction-batcher/src/main/kotlin/com/expediagroup/graphql/transactionbatcher/publisher/TriggeredPublisher.kt

@@ -0,0 +1,74 @@
+package com.expediagroup.graphql.transactionbatcher.publisher
+
+import com.expediagroup.graphql.transactionbatcher.transaction.BatcheableTransaction
+import com.expediagroup.graphql.transactionbatcher.transaction.TransactionBatcherCacheRepository
+import org.reactivestreams.Publisher
+import org.reactivestreams.Subscriber
+import org.reactivestreams.Subscription
+
+@Suppress(
+    "ReactiveStreamsSubscriberImplementation",
+    "UNCHECKED_CAST"
+)
+interface TriggeredPublisher<TInput, TOutput> {
+    fun produce(input: List<TInput>): Publisher<TOutput>
+
+    /**
+     * attempt to collect values from cache resolving say [1, null, 3, null, 5, 6]
+     * so we will only need to produce values for index 1 and 3
+     * when onComplete complete futures from either values from cache or from produce
+     * order is important
+     */
+    fun trigger(
+        batcheableTransactions: List<BatcheableTransaction<TInput, TOutput>>,
+        cacheRepository: TransactionBatcherCacheRepository
+    ) {
+
+        val values = batcheableTransactions.map { batcheableTransaction ->
+            cacheRepository.get(batcheableTransaction.key)
+        }
+
+        val transactionsNotInCache = values.mapIndexedNotNull { index, value ->
+            when (value) {
+                null -> batcheableTransactions.getOrNull(index)
+                else -> null
+            }
+        }
+
+        produce(
+            transactionsNotInCache.map(BatcheableTransaction<TInput, TOutput>::input)
+        ).subscribe(
+            object : Subscriber<TOutput> {
+                private lateinit var subscription: Subscription
+                private val results = mutableListOf<TOutput>()
+
+                override fun onSubscribe(subscription: Subscription) {
+                    this.subscription = subscription
+                    this.subscription.request(1)
+                }
+
+                override fun onNext(result: TOutput) {
+                    results += result
+                    this.subscription.request(1)
+                }
+
+                override fun onError(throwable: Throwable) {
+                    throwable.printStackTrace()
+                }
+
+                override fun onComplete() {
+                    var resultsCounter = 0
+                    values.forEachIndexed { index, value ->
+                        value?.let {
+                            batcheableTransactions[index].future.complete(value as TOutput)
+                        } ?: run {
+                            val result = results[resultsCounter++]
+                            cacheRepository.set(batcheableTransactions[index].key, result as Any)
+                            batcheableTransactions[index].future.complete(result)
+                        }
+                    }
+                }
+            }
+        )
+    }
+}

executions/transaction-batcher/src/main/kotlin/com/expediagroup/graphql/transactionbatcher/transaction/BatcheableTransaction.kt

@@ -0,0 +1,9 @@
+package com.expediagroup.graphql.transactionbatcher.transaction
+
+import java.util.concurrent.CompletableFuture
+
+data class BatcheableTransaction<TInput, TOutput>(
+    val input: TInput,
+    val future: CompletableFuture<TOutput>,
+    val key: String
+)

executions/transaction-batcher/src/main/kotlin/com/expediagroup/graphql/transactionbatcher/transaction/TransactionBatcher.kt

@@ -0,0 +1,58 @@
+package com.expediagroup.graphql.transactionbatcher.transaction
+
+import com.expediagroup.graphql.transactionbatcher.publisher.TriggeredPublisher
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.ConcurrentHashMap
+
+class TransactionBatcher {
+
+    private val queue = ConcurrentHashMap<
+        Class<out TriggeredPublisher<Any, Any>>,
+        TransactionBatcherQueueValue
+        >()
+
+    private val cacheRepository = TransactionBatcherCacheRepository()
+
+    /**
+     * deduplication
+     * if there is a BatcheableTransaction with the same input (which could be defined by client),
+     * return that future
+     * else create a new BatcheableTransaction
+     */
+    @Suppress("UNCHECKED_CAST")
+    fun <TInput : Any, TOutput : Any> enqueue(
+        input: TInput,
+        triggeredPublisher: TriggeredPublisher<TInput, TOutput>,
+        key: String = input.toString()
+    ): CompletableFuture<TOutput> {
+        val queueKey = (triggeredPublisher as TriggeredPublisher<Any, Any>)::class.java
+        return queue[queueKey]?.let { (_, batcheableTransactions) ->
+            batcheableTransactions
+                .find { transaction -> transaction.key == key }
+                ?.let { match -> match.future as CompletableFuture<TOutput> }
+                ?: run {
+                    val future = CompletableFuture<TOutput>()
+                    batcheableTransactions.add(
+                        BatcheableTransaction(input, future as CompletableFuture<Any>, key)
+                    )
+                    future
+                }
+        } ?: run {
+            val future = CompletableFuture<TOutput>()
+            queue[queueKey] = TransactionBatcherQueueValue(
+                triggeredPublisher,
+                mutableListOf(
+                    BatcheableTransaction(input, future as CompletableFuture<Any>, key)
+                )
+            )
+            future
+        }
+    }
+
+    fun dispatch() {
+        queue.values.forEach { (triggeredPublisher, transactions) ->
+            triggeredPublisher.trigger(transactions, cacheRepository)
+        }
+        queue.clear()
+    }
+}

executions/transaction-batcher/src/main/kotlin/com/expediagroup/graphql/transactionbatcher/transaction/TransactionBatcherCacheRepository.kt

@@ -0,0 +1,11 @@
+package com.expediagroup.graphql.transactionbatcher.transaction
+
+import java.util.concurrent.ConcurrentHashMap
+
+class TransactionBatcherCacheRepository {
+    private val cache = ConcurrentHashMap<String, Any>()
+
+    fun set(key: String, value: Any) = cache.put(key, value)
+
+    fun get(key: String): Any? = cache[key]
+}

executions/transaction-batcher/src/main/kotlin/com/expediagroup/graphql/transactionbatcher/transaction/TransactionBatcherQueueValue.kt

@@ -0,0 +1,8 @@
+package com.expediagroup.graphql.transactionbatcher.transaction
+
+import com.expediagroup.graphql.transactionbatcher.publisher.TriggeredPublisher
+
+internal data class TransactionBatcherQueueValue(
+    val triggeredPublisher: TriggeredPublisher<Any, Any>,
+    val transactions: MutableList<BatcheableTransaction<Any, Any>>
+)

executions/transaction-batcher/src/test/kotlin/com/expediagroup/graphql/transactionbatcher/publisher/AstronautService.kt

@@ -0,0 +1,53 @@
+package com.expediagroup.graphql.transactionbatcher.publisher
+
+import com.expediagroup.graphql.transactionbatcher.transaction.TransactionBatcher
+import org.reactivestreams.Publisher
+import reactor.core.publisher.Mono
+import reactor.kotlin.core.publisher.toFlux
+import reactor.kotlin.core.publisher.toMono
+import java.time.Duration
+import java.util.concurrent.atomic.AtomicInteger
+
+data class AstronautServiceRequest(val id: Int)
+data class Astronaut(val id: Int, val name: String)
+
+class AstronautService(
+    private val transactionBatcher: TransactionBatcher
+) {
+
+    val produceArguments: MutableList<List<AstronautServiceRequest>> = mutableListOf()
+    val getAstronautCallCount: AtomicInteger = AtomicInteger(0)
+
+    companion object {
+        private val astronauts = mapOf(
+            1 to Pair(Astronaut(1, "Buzz Aldrin"), Duration.ofMillis(300)),
+            2 to Pair(Astronaut(2, "William Anders"), Duration.ofMillis(600)),
+            3 to Pair(Astronaut(3, "Neil Armstrong"), Duration.ofMillis(200))
+        )
+    }
+
+    fun getAstronaut(request: AstronautServiceRequest): Mono<Astronaut> {
+        getAstronautCallCount.incrementAndGet()
+        val self = this
+        val future = this.transactionBatcher.enqueue(
+            request,
+            object : TriggeredPublisher<AstronautServiceRequest, Astronaut> {
+                override fun produce(input: List<AstronautServiceRequest>): Publisher<Astronaut> {
+                    produceArguments.add(input)
+                    return self.getAstronauts(input)
+                }
+            }
+        )
+        return future.toMono()
+    }
+
+    fun getAstronauts(input: List<AstronautServiceRequest>): Publisher<Astronaut> =
+        input.toFlux()
+            .flatMapSequential { request ->
+                { astronauts[request.id] }
+                    .toMono()
+                    .flatMap { (astronaut, delay) ->
+                        astronaut.toMono().delayElement(delay)
+                    }
+            }
+}

executions/transaction-batcher/src/test/kotlin/com/expediagroup/graphql/transactionbatcher/publisher/MissionService.kt

@@ -0,0 +1,57 @@
+package com.expediagroup.graphql.transactionbatcher.publisher
+
+import com.expediagroup.graphql.transactionbatcher.transaction.TransactionBatcher
+import org.reactivestreams.Publisher
+import reactor.core.publisher.Mono
+import reactor.kotlin.core.publisher.toFlux
+import reactor.kotlin.core.publisher.toMono
+import java.time.Duration
+import java.util.concurrent.atomic.AtomicInteger
+
+data class MissionServiceRequest(val id: Int)
+data class Mission(
+    val id: Int,
+    val designation: String,
+    val crew: List<Int>
+)
+
+class MissionService(
+    private val transactionBatcher: TransactionBatcher
+) {
+
+    val produceArguments: MutableList<List<MissionServiceRequest>> = mutableListOf()
+    val getMissionCallCount: AtomicInteger = AtomicInteger(0)
+
+    companion object {
+        private val missions = mapOf(
+            2 to Pair(Mission(2, "Apollo 4", listOf(14, 30, 7)), Duration.ofMillis(100)),
+            3 to Pair(Mission(3, "Apollo 5", listOf(23, 10, 12)), Duration.ofMillis(400)),
+            4 to Pair(Mission(4, "Apollo 6", listOf(1, 28, 31, 6)), Duration.ofMillis(300))
+        )
+    }
+
+    fun getMission(request: MissionServiceRequest): Mono<Mission> {
+        getMissionCallCount.incrementAndGet()
+        val self = this
+        val future = this.transactionBatcher.enqueue(
+            request,
+            object : TriggeredPublisher<MissionServiceRequest, Mission> {
+                override fun produce(input: List<MissionServiceRequest>): Publisher<Mission> {
+                    produceArguments.add(input)
+                    return self.getMissions(input)
+                }
+            }
+        )
+        return future.toMono()
+    }
+
+    fun getMissions(input: List<MissionServiceRequest>): Publisher<Mission> =
+        input.toFlux()
+            .flatMapSequential { request ->
+                { missions[request.id] }
+                    .toMono()
+                    .flatMap { (astronaut, delay) ->
+                        astronaut.toMono().delayElement(delay)
+                    }
+            }
+}