Mutiny - How to group items to send request by blocks

Question

I'm using Mutiny extension (for Quarkus) and I don't know how to manage this problem.

I want to send many request in an async way so I've read about Mutiny extension. But the server closes the connection because it receives thousand of them.

So I need:

• Send the request by blocks
• After all request are sent, do things.

I've been using Uni object to combine all the responses as this:

Uni<Map<Integer, String>> uniAll = Uni.combine()
   .all()
   .unis(list)
   .combinedWith(...);

And then:

uniAll.subscribe()
      .with(...);

This code, send all the request in paralell so the server closes the connection.

I'm using group of Multi objects, but I don't know how to use it (in Mutiny docs I can't found any example).

This is the way I'm doing now:

          //Launch 1000 request
          for (int i=0;i<1000;i++) {
            multi = client.getAbs("https://api.*********.io/jokes/random")
                      .as(BodyCodec.jsonObject())
                      .send()
                      .onItem().transformToMulti(
                              array -> Multi.createFrom()
                                       .item(array.body().getString("value")))
                                       .group()
                                       .intoLists()
                                       .of(100)
                                       .subscribe()
                                       .with(a->{
                                         System.out.println("Value: "+a);
                                        });
          }

I think that the subscription doesn't execute until there are "100" groups of items, but I guess this is not the way because it doesn't work.

Does anybody know how to launch 1000 of async requests in blocks of 100?

Thanks in advance.

UPDATED 2021-04-19

I've tried with this approach:

    List<Uni<String>> listOfUnis = new ArrayList<>();
    for (int i=0;i<1000;i++) {
        listOfUnis.add(client
                       .getAbs("https://api.*******.io/jokes/random")
                       .as(BodyCodec.jsonObject())
                       .send()
                       .onItem()
                       .transform(item -> item
                                        .body()
                                        .getString("value")));
      }


    Multi<Uni<String>> multiFormUnis = Multi.createFrom()
                                            .iterable(listOfUnis);

    List<String> listOfResponses = new ArrayList<>();

    List<String> listOfValues = multiFormUnis.group()
                 .intoLists()
                 .of(100)
                 .onItem()
                 .transformToMultiAndConcatenate(listOfOneHundred -> 
                 {
                     System.out.println("Size: "+listOfOneHundred.size());
                     
                     for (int index=0;index<listOfOneHundred.size();index++) {
                         listOfResponses.add(listOfOneHundred.get(index)
                                                             .await()
                                                             .indefinitely());                          
                     }                       
                
                return Multi.createFrom()
                            .iterable(listOfResponses);
                 })
                 .collectItems()
                 .asList()
                 .await()
                 .indefinitely();

    for (String value : listOfValues) {
        System.out.println(value);
    }

When I put this line:

 listOfResponses.add(listOfOneHundred.get(index)
                                     .await()
                                     .indefinitely()); 

The responses are printed one after each other, and when the first 100s group of items ends, it prints the next group. The problem? There are sequential requests and it takes so much time

I think I am close to the solution, but I need to know, how to send the parallel request only in group of 100s, because if I put:

subscribe().with()

All the request are sent in parallel (and not in group of 100s)

Answer 1

So in short you want to batch parallel calls to the server, without hitting it with everything at once.

Could this work for you? It uses merge. In my example, it has a parallelism of 2.

Multi.createFrom().range(1, 10)
            .onItem()
            .transformToUni(integer -> {
                return <<my long operation Uni>>
            })
            .merge(2) //this is the concurrency 
            .collect()
            .asList();

I'm not sure if merge was added later this year, but this seems to do what you want. In my example, the "long operation producing Uni" is actually a call to the Microprofile Rest Client which produces a Uni, and returns a string. After the merge you can put another onItem to perform something with the response (it's a plain Multi after the merge), instead of collecting everything as list.

Answer 2

I think you create the multy wrong, it would be much easier to use this:

Multi<String> multiOfJokes = Multi.createFrom().emitter(multiEmitter -> {
        for (int i=0;i<1000;i++) {
            multiEmitter.emit(i);
        }
        multiEmitter.complete();
    }).onItem().transformToUniAndMerge(index -> {
        return Uni.createFrom().item("String" + index);
    })

With this approach it should mace the call parallel. Now is the question of how to make it to a list.

The grouping works fine I run it with this code:

Random random = new Random();
    Multi<Integer> multiOfInteger = Multi.createFrom().emitter(multiEmitter -> {
        for (Integer i=0;i<1000;i++) {
            multiEmitter.emit(i);
        }
        multiEmitter.complete();
    });
    Multi<String> multiOfJokes = multiOfInteger.onItem().transformToUniAndMerge(index -> {
        if (index % 10 == 0 ) {
            Duration delay = Duration.ofMillis(random.nextInt(100) + 1);
            return Uni.createFrom().item("String " + index  + " delayed").onItem()
                    .delayIt().by(delay);
        }
        return Uni.createFrom().item("String" + index);
    }).onCompletion().invoke(() -> System.out.println("Completed"));
    Multi<List<String>> multiListJokes = multiOfJokes
            .group().intoLists().of(100)
            .onCompletion().invoke(() -> System.out.println("Completed"))
            .onItem().invoke(strings -> System.out.println(strings));
    multiListJokes.collect().asList().await().indefinitely();

You will get a list of your string.

I don't know, how you intend to send the list to backend. But you can either to it with:

• call (executed asynchronously)
• write own subscriber (implements Subscriber) the methods are straight forward. As you need for your bulk request.

I hope you understand it better afterward.

PS: link to guide where I learned all of it: https://smallrye.io/smallrye-mutiny/guides