Reaktor 3.x (Java): für das Abstreifen des Netzes

Question

Reaktor noob hier.

Dies ist eher eine HowTo Frage.

Angenommen, ich habe eine Website, die ich mit einem seitenweisen Suchergebnis auskratzen möchte. Die Anzahl der Suchergebnisseiten ist unbekannt. Jede Suchseite hat einen Link zur nächsten Seite. Ich möchte alle Suchergebnisse von allen Seiten scrappen und jedes Suchergebnis verarbeiten.

Wie kann ich Reactor (Mono/Flux) verwenden, um dies in Java zu erreichen?

Ich möchte dies als "reaktiv" wie möglich tun.

Grundsätzlich ist der Reaktor (3.x) Version des folgenden Imperativ Pseudo-Code:

String url = "http://example.com/search/1"; 
    Optional<Document> docOp = getNextPage(url); (1) 
    while (docOp.isPresent()) { 
     Document doc = docOp.get(); 
     processDoc(doc);       (2) 
     docOp = getNextPage(getNextUrl(doc));  (3) 
    } 

    // (1) Get the first page of search results 
    // (2) Process all the search results on this page asynchronously 
    // (3) Find the next page URL, and get that page 

Answer 1

Mit etwas Hilfe von https://gitter.im/reactor/reactor kam ich an dieser Lösung. Es ist vielleicht nicht ideal. Ich würde gerne Feedback zu allen Problemen bekommen, die irgendjemand sehen könnte.

public void scrape() { 

    Try<Document> firstDocTry = this.getSearchResultsPage(Option.<Document>none().toTry()); (1) 

    // Generate a flux where each element in the flux is created using the current element 
    Flux.<Try<Document>, Try<Document>>generate(() -> firstDocTry, (docTry, sink) -> {   (2) 
      docTry = this.getSearchResultsPage(docTry); 
      docTry.isFailure() ? sink.complete() : sink.next(docTry); 
      return docTry; 
     }) 
     .flatMap(docTry -> this.transformToScrapedLoads(docTry))        (3) 
     .log() 
     .subscribe(scrapedLoad -> 
      scrapedLoadRepo.save(scrapedLoad)             (4) 
     ); 
} 

protected Try<Document> getSearchResultsPage(Try<Document> docTry) { 
    ... 
} 

protected Flux<ScrapedLoad> transformToScrapedLoads(Try<Document> docTry) { 
    ... 
} 

(1) Verwenden von Javaslangs monadischen Try und Option hier. 'firstDocTry' Seeds den Generator. getSearchResultsPage() weiß zu Start auf der ersten Seite der Suche, wenn kein Dokument zur Verfügung gestellt.

(2) Hier einen Generator verwenden. Jedes Element in dem Flux veröffentlicht wird durch das vorherige Element bestimmt

(3) das Transformationsverfahren wandelt jeden doc an ein Flux, das kombiniert werden, und gesendet als einzelnes Flux abonnieren

(4) Der Teilnehmer betreibt auf jedes vom Flux produzierte Element. In diesem Fall persistent sie.