How could I merge active reactive streams?

I just started exploring project reactor and my current understanding of it's main purpose is to get information from several sources in asynchronous manner. For example, I have one source, that produces data fast and one, that produces data pretty slow. I wanna merge these two sources and return information from both of them as soon, as i got it.
I've tried code like this:

@RestController
public class ReactiveController {

@RequestMapping("/info")
public Flux<String> getInfoFromServices() {
    return Flux.merge(getDataFromSlowSource(), getDataFromFastSource()).take(10);
}

private Flux<String> getDataFromFastSource() {
    return Flux.generate(sink -> {
        sink.next("data from fast source\n");
        try {
            Thread.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
}

private Flux<String> getDataFromSlowSource() {
    return Flux.generate(sink -> {
        sink.next("data from slow source\n");
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
}

I expected something like this in answer from server:

data from fast source
data from fast source
data from slow source
data from fast source
data from fast source
data from slow source
data from fast source
data from fast source
data from slow source

But I've got:

data from slow source
data from slow source
data from slow source
data from slow source
data from slow source
data from slow source
data from slow source
data from slow source
data from slow source
data from slow source

So, could I somehow take data from both sources at the same time as soon as source produces it?

Your problem is Thread.sleep which blocks the current thread. That is why you see that behavior. Instead of using Thread.sleep, use delayElements in the fast and slow sources, you would see the expected behavior.

private Flux<Object> getDataFromFastSource() {
    return Flux.generate(sink -> {
        sink.next("data from fast source\n");
    }).delayElements(Duration.ofSeconds(1));
}

private Flux<Object> getDataFromSlowSource() {
    return Flux.generate(sink -> {
        sink.next("data from slow source\n");
    }).delayElements(Duration.ofSeconds(2));
}

Note:

Always make the whole reactor chain non-blocking and use appropriate schedulers to do that. More info is here.

http://www.vinsguru.com/reactive-programming-schedulers/

Your problem is that merge operation, slow source and fast source all run on the same thread. Thus there is no race between the two sources.

If you modify your code like this, so that slow and fast sources run on separate threads (schedulers), you will see the expected result:

@RequestMapping("/info")
public Flux<String> getInfoFromServices() {
    return Flux.merge(
      getDataFromSlowSource().subscribeOn(Schedulers.boundedElastic(),
      getDataFromFastSource().subscribeOn(Schedulers.boundedElastic()
    ).take(10);
}

Regarding vins's answer: delayElements(duration) method also employs a scheduler.