Switch to reactor and fix tests (#343)

* WIP: Switch to reactor * Use Micronaut 3.0.0-SNAPSHOT * Add back micronaut-reactor because of the converters * remove snapshot repository * pin micrometer version to 4.0.0-RC2 * build: set micronaut version to 3.0.0-M4 * return Publisher and SingleResult Co-authored-by: Sergio del Amo <[email protected]>
micronaut-projects · Jul 22, 2021 · ea5e7fd · ea5e7fd
1 parent 6cdcf76
commit ea5e7fd
Show file tree

Hide file tree

Showing 21 changed files with 166 additions and 144 deletions.
diff --git a/build.gradle b/build.gradle
@@ -9,9 +9,6 @@ buildscript {
 }
 
 subprojects { Project subproject ->
-    repositories {
-        maven { url "https://s01.oss.sonatype.org/content/repositories/snapshots/" }
-    }
     group "io.micronaut.cache"
 
     if (subproject.name != 'cache-bom') {

diff --git a/cache-caffeine/build.gradle b/cache-caffeine/build.gradle
@@ -1,18 +1,16 @@
 dependencies {
     api project(":cache-core")
     api "com.github.ben-manes.caffeine:caffeine:${caffeineVersion}"
-    implementation "io.micronaut.reactor:micronaut-reactor"
     compileOnly "ch.qos.logback:logback-classic:1.2.3"
-    compileOnly "io.micronaut.micrometer:micronaut-micrometer-core"
+    implementation "io.projectreactor:reactor-core"
+    compileOnly "io.micronaut.micrometer:micronaut-micrometer-core:$micronautMicrometerVersion"
 
     testImplementation "ch.qos.logback:logback-classic:1.2.3"
 
     testImplementation 'org.jsr107.ri:cache-ri-impl:1.1.1'
     testImplementation "io.micronaut:micronaut-inject-groovy"
-    testImplementation 'io.micronaut.reactor:micronaut-reactor-http-client:1.2.0'
     testImplementation "io.micronaut:micronaut-management"
-    testImplementation "io.micronaut.reactor:micronaut-reactor-http-client"
+    testImplementation "io.micronaut:micronaut-http-client"
     testImplementation "io.micronaut:micronaut-http-server-netty"
-
-    testImplementation "io.micronaut.micrometer:micronaut-micrometer-core"
-}
+    testImplementation "io.micronaut.micrometer:micronaut-micrometer-core:$micronautMicrometerVersion"
+}
diff --git a/cache-caffeine/src/test/groovy/io/micronaut/cache/SyncCacheSpec.groovy b/cache-caffeine/src/test/groovy/io/micronaut/cache/SyncCacheSpec.groovy
@@ -26,6 +26,7 @@ import io.micronaut.cache.annotation.PutOperations
 import io.micronaut.context.ApplicationContext
 import io.micronaut.inject.qualifiers.Qualifiers
 import jakarta.inject.Singleton
+import org.reactivestreams.Publisher
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import spock.lang.Retry
@@ -58,30 +59,30 @@ class SyncCacheSpec extends Specification {
         CounterService counterService = applicationContext.getBean(CounterService)
 
         then:
-        counterService.fluxValue("test").blockFirst() == 0
-        counterService.monoValue("test").block() == 0
+        Flux.from(counterService.fluxValue("test")).blockFirst() == 0
+        Mono.from(counterService.monoValue("test")).block() == 0
 
         when:
         counterService.reset()
         def result =counterService.increment("test")
 
         then:
         result == 1
-        counterService.fluxValue("test").blockFirst() == 1
+        Flux.from(counterService.fluxValue("test")).blockFirst() == 1
         counterService.futureValue("test").get() == 1
         counterService.stageValue("test").toCompletableFuture().get() == 1
-        counterService.monoValue("test").block() == 1
+        Mono.from(counterService.monoValue("test")).block() == 1
         counterService.getValue("test") == 1
 
         when:
         result = counterService.incrementNoCache("test")
 
         then:
         result == 2
-        counterService.fluxValue("test").blockFirst() == 1
+        Flux.from(counterService.fluxValue("test")).blockFirst() == 1
         counterService.futureValue("test").get() == 1
         counterService.stageValue("test").toCompletableFuture().get() == 1
-        counterService.monoValue("test").block() == 1
+        Mono.from(counterService.monoValue("test")).block() == 1
         counterService.getValue("test") == 1
 
         when:
@@ -162,25 +163,25 @@ class SyncCacheSpec extends Specification {
         publisherService.callCount.get() == 0
 
         when:
-        publisherService.fluxValue("abc").blockFirst()
+        Flux.from(publisherService.fluxValue("abc")).blockFirst()
 
         then:
         publisherService.callCount.get() == 1
 
         when:
-        publisherService.fluxValue("abc").blockFirst()
+        Flux.from(publisherService.fluxValue("abc")).blockFirst()
 
         then:
         publisherService.callCount.get() == 1
 
         when:
-        publisherService.monoValue("abcd").block()
+        Mono.from(publisherService.monoValue("abcd")).block()
 
         then:
         publisherService.callCount.get() == 2
 
         when:
-        publisherService.monoValue("abcd").block()
+        Mono.from(publisherService.monoValue("abcd")).block()
 
         then:
         publisherService.callCount.get() == 2
@@ -291,13 +292,13 @@ class SyncCacheSpec extends Specification {
 
         @Cacheable
         @SingleResult
-        Flux<Integer> fluxValue(String name) {
+        Publisher<Integer> fluxValue(String name) {
             callCount.incrementAndGet()
             return Flux.just(0)
         }
 
         @Cacheable
-        Mono<Integer> monoValue(String name) {
+        Publisher<Integer> monoValue(String name) {
             callCount.incrementAndGet()
             return Mono.just(0)
         }
@@ -346,12 +347,13 @@ class SyncCacheSpec extends Specification {
 
         @Cacheable
         @SingleResult
-        Flux<Integer> fluxValue(String name) {
+        Publisher<Integer> fluxValue(String name) {
             return Flux.just(counters.computeIfAbsent(name, { 0 }))
         }
 
         @Cacheable
-        Mono<Integer> monoValue(String name) {
+        @SingleResult
+        Publisher<Integer> monoValue(String name) {
             return Mono.just(counters.computeIfAbsent(name, { 0 }))
         }
 

diff --git a/cache-caffeine/src/test/groovy/io/micronaut/cache/client/CachingReactorCrudSpec.groovy b/cache-caffeine/src/test/groovy/io/micronaut/cache/client/CachingReactorCrudSpec.groovy
@@ -19,6 +19,7 @@ import io.micronaut.cache.annotation.CacheInvalidate
 import io.micronaut.cache.annotation.CachePut
 import io.micronaut.cache.annotation.Cacheable
 import io.micronaut.context.ApplicationContext
+import io.micronaut.core.async.annotation.SingleResult
 import io.micronaut.http.HttpResponse
 import io.micronaut.http.HttpStatus
 import io.micronaut.http.annotation.Controller
@@ -28,6 +29,7 @@ import io.micronaut.http.annotation.Patch
 import io.micronaut.http.annotation.Post
 import io.micronaut.http.client.annotation.Client
 import io.micronaut.runtime.server.EmbeddedServer
+import org.reactivestreams.Publisher
 import reactor.core.publisher.Mono
 import reactor.core.publisher.MonoSink
 import spock.lang.AutoCleanup
@@ -55,25 +57,27 @@ class CachingReactorCrudSpec extends Specification {
         given:
         BookClient client = context.getBean(BookClient)
         BookController bookController = context.getBean(BookController)
-
+        
         when:
-        Book book = client.get(99).block()
-        List<Book> books = client.list().block()
+        Optional<Book> bookOptional = Mono.from(client.get(99))
+                .onErrorResume(t -> Mono.empty())
+                .blockOptional()
+        List<Book> books = Mono.from(client.list()).block()
 
         then:
-        book == null
+        !bookOptional.isPresent()
         books.size() == 0
 
         when:
-        book = client.save("The Stand").block()
+        Book book = Mono.from(client.save("The Stand")).block()
 
         then:
         book != null
         book.title == "The Stand"
         book.id == 1
 
         when:
-        book = client.get(book.id).block()
+        book = Mono.from(client.get(book.id)).block()
 
         then:
         book != null
@@ -83,7 +87,7 @@ class CachingReactorCrudSpec extends Specification {
 
 
         when:
-        book = client.get(book.id).block()
+        book = Mono.from(client.get(book.id)).block()
 
         then:
         book != null
@@ -92,22 +96,22 @@ class CachingReactorCrudSpec extends Specification {
         bookController.getInvocationCount.get() == 2
 
         when:'the full response is resolved'
-        HttpResponse<Book> bookAndResponse = client.getResponse(book.id).block()
+        HttpResponse<Book> bookAndResponse = Mono.from(client.getResponse(book.id)).block()
 
         then:"The response is valid"
         bookAndResponse.status() == HttpStatus.OK
         bookAndResponse.body().title == "The Stand"
 
         when:
-        book = client.update(book.id, "The Shining").block()
+        book = Mono.from(client.update(book.id, "The Shining")).block()
 
         then:
         book != null
         book.title == "The Shining"
         book.id == 1
 
         when:
-        book = client.get(book.id).block()
+        book = Mono.from(client.get(book.id)).block()
 
         then:
         book != null
@@ -116,16 +120,16 @@ class CachingReactorCrudSpec extends Specification {
         bookController.getInvocationCount.get() == 3
 
         when:
-        book = client.delete(book.id).block()
+        book = Mono.from(client.delete(book.id)).block()
 
         then:
         book != null
 
         when:
-        book = client.get(book.id).block()
+        bookOptional = Mono.from(client.get(book.id)).onErrorResume(t -> Mono.empty()).blockOptional()
 
         then:
-        book == null
+        !bookOptional.isPresent()
     }
 
 
@@ -141,23 +145,17 @@ class CachingReactorCrudSpec extends Specification {
         AtomicInteger getInvocationCount = new AtomicInteger()
 
         @Override
-        Mono<Book> get(Long id) {
-            Mono.create(new Consumer<MonoSink<Book>>() {
-                @Override
-                void accept(MonoSink<Book> monoSink) {
-                    getInvocationCount.incrementAndGet()
-                    Book book = books.get(id)
-                    if(book) {
-                        monoSink.success(book)
-                    } else {
-                        monoSink.success()
-                    }
-                }
-            })
+        @SingleResult
+        Publisher<Book> get(Long id) {
+            getInvocationCount.incrementAndGet()
+            Book book = books.get(id)
+            Mono<Book> mono = book != null ? Mono.just(book) : Mono.empty()
+            return mono
         }
 
         @Override
-        Mono<HttpResponse<Book>> getResponse(Long id) {
+        @SingleResult
+        Publisher<HttpResponse<Book>> getResponse(Long id) {
             Book book = books.get(id)
             if(book) {
                 return Mono.just(HttpResponse.ok(book))
@@ -166,12 +164,14 @@ class CachingReactorCrudSpec extends Specification {
         }
 
         @Override
-        Mono<List<Book>> list() {
+        @SingleResult
+        Publisher<List<Book>> list() {
             return Mono.just(books.values().toList())
         }
 
         @Override
-        Mono<Book> delete(Long id) {
+        @SingleResult
+        Publisher<Book> delete(Long id) {
             Book book = books.remove(id)
             if(book) {
                 return Mono.just(book)
@@ -180,14 +180,16 @@ class CachingReactorCrudSpec extends Specification {
         }
 
         @Override
-        Mono<Book> save(String title) {
+        @SingleResult
+        Publisher<Book> save(String title) {
             Book book = new Book(title: title, id:currentId.incrementAndGet())
             books[book.id] = book
             return Mono.just(book)
         }
 
         @Override
-        Mono<Book> update(Long id, String title) {
+        @SingleResult
+        Publisher<Book> update(Long id, String title) {
             Book book = books[id]
             if(book != null) {
                 book.title = title
@@ -203,27 +205,33 @@ class CachingReactorCrudSpec extends Specification {
 
         @Get("/{id}")
         @Cacheable("books")
-        Mono<Book> get(Long id)
+        @SingleResult
+        Publisher<Book> get(Long id)
 
         @Get("/res/{id}")
         @Cacheable("books")
-        Mono<HttpResponse<Book>> getResponse(Long id)
+        @SingleResult
+        Publisher<HttpResponse<Book>> getResponse(Long id)
 
         @Get
         @Cacheable("book-list")
-        Mono<List<Book>> list()
+        @SingleResult
+        Publisher<List<Book>> list()
 
         @Delete("/{id}")
         @CacheInvalidate("books")
-        Mono<Book> delete(Long id)
+        @SingleResult
+        Publisher<Book> delete(Long id)
 
         @Post
         @CachePut("books")
-        Mono<Book> save(String title)
+        @SingleResult
+        Publisher<Book> save(String title)
 
         @Patch("/{id}")
         @CacheInvalidate("books")
-        Mono<Book> update(Long id, String title)
+        @SingleResult
+        Publisher<Book> update(Long id, String title)
     }
 
 

diff --git a/cache-core/build.gradle b/cache-core/build.gradle
@@ -3,14 +3,14 @@ dependencies {
     api "io.micronaut:micronaut-inject"
     api "io.micronaut:micronaut-aop"
     api "io.micronaut:micronaut-runtime"
-    implementation "io.micronaut.reactor:micronaut-reactor"
+    implementation "io.projectreactor:reactor-core"
     compileOnly "javax.cache:cache-api"
     compileOnly "io.micronaut:micronaut-management"
-    compileOnly "io.micronaut.micrometer:micronaut-micrometer-core"
+    compileOnly "io.micronaut.micrometer:micronaut-micrometer-core:$micronautMicrometerVersion"
 
     compileOnly "ch.qos.logback:logback-classic:1.2.3"
     testImplementation "ch.qos.logback:logback-classic:1.2.3"
-
+    testImplementation "io.micronaut:micronaut-http-client"
     testImplementation 'org.jsr107.ri:cache-ri-impl:1.1.1'
     testImplementation "javax.cache:cache-api"
     testImplementation "io.micronaut:micronaut-inject-groovy"