feat(sandside): #224 refactor OPML loader, improve architecture

avoid use Spring DataBuffer in domain fix PipedOutputstream close and unit tests
Marthym · Sep 29, 2024 · f893ced · f893ced
1 parent 428373a
commit f893ced
Show file tree

Hide file tree

Showing 5 changed files with 177 additions and 22 deletions.
diff --git a/sandside/src/main/java/fr/ght1pc9kc/baywatch/opml/api/OpmlService.java b/sandside/src/main/java/fr/ght1pc9kc/baywatch/opml/api/OpmlService.java
@@ -1,13 +1,12 @@
 package fr.ght1pc9kc.baywatch.opml.api;
 
-import org.springframework.core.io.buffer.DataBuffer;
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.io.InputStream;
+import java.util.function.Supplier;
 
 public interface OpmlService {
     Mono<InputStream> opmlExport();
 
-    Mono<Void> opmlImport(Flux<DataBuffer> data);
+    Mono<Void> opmlImport(Supplier<InputStream> inputSupplier);
 }
diff --git a/sandside/src/main/java/fr/ght1pc9kc/baywatch/opml/domain/OpmlServiceImpl.java b/sandside/src/main/java/fr/ght1pc9kc/baywatch/opml/domain/OpmlServiceImpl.java
@@ -1,25 +1,24 @@
 package fr.ght1pc9kc.baywatch.opml.domain;
 
 import com.machinezoo.noexception.Exceptions;
+import fr.ght1pc9kc.baywatch.common.domain.QueryContext;
 import fr.ght1pc9kc.baywatch.opml.api.OpmlService;
 import fr.ght1pc9kc.baywatch.security.api.AuthenticationFacade;
 import fr.ght1pc9kc.baywatch.security.api.model.User;
 import fr.ght1pc9kc.baywatch.security.domain.exceptions.UnauthenticatedUser;
 import fr.ght1pc9kc.baywatch.techwatch.api.model.WebFeed;
-import fr.ght1pc9kc.baywatch.common.domain.QueryContext;
 import fr.ght1pc9kc.baywatch.techwatch.infra.persistence.FeedRepository;
 import fr.ght1pc9kc.entity.api.Entity;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.DataBufferUtils;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.PipedInputStream;
 import java.io.PipedOutputStream;
+import java.util.function.Supplier;
 
 @Slf4j
 @RequiredArgsConstructor
@@ -47,25 +46,17 @@ public Mono<InputStream> opmlExport() {
     }
 
     @Override
-    public Mono<Void> opmlImport(Flux<DataBuffer> data) {
-        log.debug("Start importing...");
+    public Mono<Void> opmlImport(Supplier<InputStream> inputSupplier) {
         return authFacade.getConnectedUser()
                 .switchIfEmpty(Mono.error(new UnauthenticatedUser("Authentication not found !")))
                 .flatMapMany(Exceptions.wrap().function(owner -> {
-                    PipedOutputStream pos = new PipedOutputStream();
-                    PipedInputStream pis = new PipedInputStream(pos);
-                    Flux<Entity<WebFeed>> feeds = readOpml(pis);
-                    Mono<Entity<WebFeed>> db = DataBufferUtils.write(data, pos)
-                            .map(DataBufferUtils::release)
-                            .doOnTerminate(Exceptions.wrap().runnable(() -> {
-                                pos.flush();
-                                pos.close();
-                            }))
-                            .then(Mono.empty());
-                    return Flux.merge(db, feeds)
+                    InputStream is = inputSupplier.get();
+                    return readOpml(is)
                             .buffer(100)
                             .flatMap(f -> feedRepository.persist(f).collectList())
-                            .flatMap(f -> feedRepository.persistUserRelation(owner.id(), f));
+                            .flatMap(f -> feedRepository.persistUserRelation(owner.id(), f))
+                            .doOnTerminate(Exceptions.wrap().runnable(is::close));
+
                 })).then();
     }
 

diff --git a/sandside/src/main/java/fr/ght1pc9kc/baywatch/opml/infra/OpmlController.java b/sandside/src/main/java/fr/ght1pc9kc/baywatch/opml/infra/OpmlController.java
@@ -1,11 +1,13 @@
 package fr.ght1pc9kc.baywatch.opml.infra;
 
+import com.machinezoo.noexception.Exceptions;
 import fr.ght1pc9kc.baywatch.opml.api.OpmlService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.core.io.InputStreamResource;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
@@ -20,9 +22,13 @@
 import org.springframework.web.bind.annotation.RestController;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
 
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
 import java.nio.charset.StandardCharsets;
 import java.time.LocalDateTime;
 
@@ -34,6 +40,7 @@
 public class OpmlController {
 
     private final OpmlService opmlService;
+    private final Scheduler uploadReader = Schedulers.boundedElastic();
 
     @ResponseBody
     @GetMapping("/export/baywatch.opml")
@@ -51,8 +58,19 @@ public Mono<ResponseEntity<Resource>> exportOpml() {
     }
 
     @PostMapping("/import")
+    @SuppressWarnings("CallingSubscribeInNonBlockingScope")
     public Mono<Void> importOpml(@RequestPart("opml") Mono<FilePart> opmlFilePart) {
         Flux<DataBuffer> data = opmlFilePart.flatMapMany(Part::content);
-        return opmlService.opmlImport(data);
+
+        PipedOutputStream pos = new PipedOutputStream();
+        DataBufferUtils.write(data, pos)
+                .doOnTerminate(Exceptions.wrap().runnable(pos::close))
+                .subscribe(
+                        DataBufferUtils.releaseConsumer(),
+                        t -> log.atError().log("STACKTRACE", t)
+                );
+
+        return opmlService.opmlImport(Exceptions.wrap().supplier(() -> new PipedInputStream(pos)))
+                .subscribeOn(uploadReader);
     }
-}
+}
diff --git a/sandside/src/test/java/fr/ght1pc9kc/baywatch/opml/domain/OpmlServiceImplTest.java b/sandside/src/test/java/fr/ght1pc9kc/baywatch/opml/domain/OpmlServiceImplTest.java
@@ -0,0 +1,75 @@
+package fr.ght1pc9kc.baywatch.opml.domain;
+
+import fr.ght1pc9kc.baywatch.common.domain.QueryContext;
+import fr.ght1pc9kc.baywatch.opml.api.OpmlService;
+import fr.ght1pc9kc.baywatch.security.api.AuthenticationFacade;
+import fr.ght1pc9kc.baywatch.security.infra.adapters.SpringAuthenticationContext;
+import fr.ght1pc9kc.baywatch.techwatch.api.model.WebFeed;
+import fr.ght1pc9kc.baywatch.techwatch.infra.persistence.FeedRepository;
+import fr.ght1pc9kc.baywatch.tests.samples.FeedSamples;
+import fr.ght1pc9kc.baywatch.tests.samples.UserSamples;
+import fr.ght1pc9kc.entity.api.Entity;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Collection;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+class OpmlServiceImplTest {
+
+    private OpmlService tested;
+
+    private AuthenticationFacade authenticationFacadeMock;
+    private FeedRepository feedRepositoryMock;
+
+    @BeforeEach
+    void setUp() {
+        feedRepositoryMock = mock(FeedRepository.class);
+        doReturn(Flux.just(FeedSamples.JEDI)).when(feedRepositoryMock).list(any(QueryContext.class));
+        doAnswer(answer -> Flux.fromIterable(answer.getArgument(0))).when(feedRepositoryMock).persist(any());
+        doAnswer(answer -> Flux.fromIterable(answer.getArgument(1))).when(feedRepositoryMock).persistUserRelation(anyString(), any());
+
+        authenticationFacadeMock = spy(new SpringAuthenticationContext());
+        tested = new OpmlServiceImpl(feedRepositoryMock, authenticationFacadeMock, OpmlWriter::new, OpmlReader::new);
+    }
+
+    @Test
+    void should_export_opml() {
+        doReturn(Mono.just(UserSamples.OBIWAN)).when(authenticationFacadeMock).getConnectedUser();
+
+        StepVerifier.create(tested.opmlExport())
+                .assertNext(actual -> Assertions.assertThat(actual).asString(StandardCharsets.UTF_8)
+                        .contains("<ownerName>Obiwan Kenobi</ownerName>")
+                        .contains("<title>Baywatch OPML export</title>")
+                        .containsIgnoringNewLines("""
+                                <outline text="Jedi Feed" type="rss" xmlUrl="https://www.jedi.com/" title="Jedi Feed" category=""/>
+                                """))
+                .verifyComplete();
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    void should_import_opml() {
+        doReturn(Mono.just(UserSamples.OBIWAN)).when(authenticationFacadeMock).getConnectedUser();
+
+        StepVerifier.create(tested.opmlImport(() -> OpmlServiceImplTest.class.getResourceAsStream("okenobi.xml")))
+                .verifyComplete();
+
+        ArgumentCaptor<Collection<Entity<WebFeed>>> captor = ArgumentCaptor.forClass(Collection.class);
+        verify(feedRepositoryMock).persist(captor.capture());
+        Assertions.assertThat(captor.getValue().size()).isEqualTo(30);
+    }
+}
diff --git a/sandside/src/test/resources/fr/ght1pc9kc/baywatch/opml/domain/okenobi.xml b/sandside/src/test/resources/fr/ght1pc9kc/baywatch/opml/domain/okenobi.xml
@@ -0,0 +1,72 @@
+<?xml version='1.0' encoding='UTF-8'?>
+<opml version="2.0">
+    <head>
+        <title>Baywatch OPML export</title>
+        <dateCreated>Sun, 28 Apr 2024 17:47:47 GMT</dateCreated>
+        <ownerName>Fred</ownerName>
+        <ownerEmail>marthym@gmail.com</ownerEmail>
+    </head>
+    <body>
+        <outline text="A Java Geek" type="rss" xmlUrl="https://blog.frankel.ch/feed.xml" title="A Java Geek"
+                 category="coding,java,techno"/>
+        <outline text="Clubic.com" type="rss" xmlUrl="https://www.clubic.com/feed/news-pro.rss" title="Clubic.com"
+                 category="culture,high-tech"/>
+        <outline text="Clubic.com" type="rss" xmlUrl="https://www.clubic.com/feed/news.rss" title="Clubic.com"
+                 category="culture,high-tech"/>
+        <outline text="Bits from Debian" type="rss" xmlUrl="https://bits.debian.org/feeds/feed.rss"
+                 title="Bits from Debian" category="linux,techno"/>
+        <outline text="Developpez.com" type="rss" xmlUrl="https://www.developpez.com/index/rss" title="Developpez.com"
+                 category="high-tech,coding"/>
+        <outline text="ght1kp9kc.fr" type="rss" xmlUrl="https://blog.ght1pc9kc.fr/index.xml" title="ght1kp9kc.fr"
+                 category="techno,culture"/>
+        <outline text="Human Coders News" type="rss" xmlUrl="https://news.humancoders.com/items/feed.rss"
+                 title="Human Coders News" category="coding,java,techno"/>
+        <outline text="Journal du hacker" type="rss" xmlUrl="https://www.journalduhacker.net/rss"
+                 title="Journal du hacker" category="culture,techno"/>
+        <outline text="Le blog de Seboss666" type="rss" xmlUrl="https://blog.seboss666.info/feed/"
+                 title="Le blog de Seboss666" category="culture,techno"/>
+        <outline text="Le Monde.fr - Actualités et Infos en France et dans le monde" type="rss"
+                 xmlUrl="https://www.lemonde.fr/rss/une.xml"
+                 title="Le Monde.fr - Actualités et Infos en France et dans le monde" category="culture,general"/>
+        <outline text="Les Cast Codeurs" type="rss" xmlUrl="https://lescastcodeurs.libsyn.com/rss"
+                 title="Les Cast Codeurs" category="culture,java,techno"/>
+        <outline text="OCTO Talks !" type="rss" xmlUrl="https://blog.octo.com/feed/" title="OCTO Talks !"
+                 category="culture,techno"/>
+        <outline text="Spring" type="rss" xmlUrl="https://spring.io/blog.atom" title="Spring" category="java,techno"/>
+        <outline text="Lyft Engineering - Medium" type="rss" xmlUrl="https://eng.lyft.com/feed"
+                 title="Lyft Engineering - Medium" category="techno"/>
+        <outline text="Netflix TechBlog - Medium" type="rss" xmlUrl="https://netflixtechblog.com/feed"
+                 title="Netflix TechBlog - Medium" category="techno"/>
+        <outline text="The Airbnb Tech Blog - Medium" type="rss" xmlUrl="https://medium.com/feed/airbnb-engineering"
+                 title="The Airbnb Tech Blog - Medium" category="techno"/>
+        <outline text="The Pragmatic Engineer" type="rss" xmlUrl="https://feeds.feedburner.com/ThePragmaticEngineer"
+                 title="The Pragmatic Engineer" category="techno,management"/>
+        <outline text="Solutions Numeriques" type="rss" xmlUrl="https://www.solutions-numeriques.com/dossiers/feed/"
+                 title="Solutions Numeriques" category="management,high-tech"/>
+        <outline text="Uber Blog" type="rss" xmlUrl="https://www.uber.com/blog/engineering/rss/" title="Uber Blog"
+                 category="techno,coding"/>
+        <outline text="ByteByteGo Newsletter" type="rss" xmlUrl="https://blog.bytebytego.com/feed"
+                 title="ByteByteGo Newsletter" category="techno,management"/>
+        <outline text="HowToDoInJava" type="rss" xmlUrl="https://howtodoinjava.com/feed/" title="HowToDoInJava"
+                 category="coding,java,techno"/>
+        <outline text="patkua.com" type="rss" xmlUrl="https://www.patkua.com/feed/" title="patkua.com"
+                 category="techno,management"/>
+        <outline text="Une tasse de café" type="rss" xmlUrl="https://une-tasse-de.cafe/index.xml"
+                 title="Une tasse de café" category="sysadmin,techno"/>
+        <outline text="Tailwind CSS Blog" type="rss" xmlUrl="https://tailwindcss.com/feeds/feed.xml"
+                 title="Tailwind CSS Blog" category=""/>
+        <outline text="foojay" type="rss" xmlUrl="https://foojay.io/feed/" title="foojay"
+                 category="coding,java,techno"/>
+        <outline text="Java Newsletter" type="rss" xmlUrl="https://javabulletin.substack.com/feed"
+                 title="Java Newsletter" category="coding,java,techno"/>
+        <outline text="Guillaume Laforge" type="rss" xmlUrl="https://glaforge.dev/index.xml" title="Guillaume Laforge"
+                 category="coding,high-tech"/>
+        <outline text="Frontend Masters Boost" type="rss" xmlUrl="https://frontendmasters.com/blog/feed/"
+                 title="Frontend Masters Boost" category="coding,frontend"/>
+        <outline text="top scoring links : java" type="rss" xmlUrl="https://www.reddit.com/r/java/top/.rss?sort=new"
+                 title="top scoring links : java" category="coding,java,techno"/>
+        <outline text="top scoring links : programming" type="rss"
+                 xmlUrl="https://www.reddit.com/r/programming/top/.rss?sort=new" title="top scoring links : programming"
+                 category="coding,techno"/>
+    </body>
+</opml>