ReacxtorMergeTest

package com.instana;

import static org.junit.Assert.assertTrue;

import java.time.Duration;
import java.util.ArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import reactor.core.publisher.DirectProcessor;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

public class RxJavaMergeTest {

  private final ExecutorService executor = Executors.newFixedThreadPool(1);

  private final Scheduler scheduler = Schedulers.newSingle("test");

  private final MessageProducer messageProducer = new MessageProducer(40);
  private final MessageConsumer messageConsumer = new MessageConsumer();

  @Before
  public void setup() {
    messageProducer.start();
  }

  @After
  public void stop() {
    messageProducer.stop();
    executor.shutdown();
  }

  @Test
  public void test() throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(10);

    messageProducer
        .getObservable()
        .onBackpressureDrop()
        .publishOn(Schedulers.elastic())
        .doOnRequest(x -> System.out.println("requested: " + x))
        .flatMapIterable(x -> {
          ArrayList<Object> result = new ArrayList<>();
          for (int i = 0; i < 50; i++) {
            result.add(x);
          }
          return result;
        })
        .mergeWith(
            Flux
                .interval(Duration.ofMillis(300), scheduler)
                .<Object>map(x -> MessageConsumer.SLOWDOWN_MARKER))
        .doOnNext(x -> {
          if (x == MessageConsumer.SLOWDOWN_MARKER) {
            messageProducer.start();
            messageConsumer.setWaitMillis(1);
            System.out.println("Slowing down...");
            LockSupport.parkNanos(50 * 1_000_000);
          }
        })
        .subscribe(messageConsumer::consume);

    Flux
        .interval(Duration.ofMillis(50), scheduler)
        .subscribe(x -> {
          System.out.println("Countdown: " + x);
          latch.countDown();
        });

    assertTrue(latch.await(20, TimeUnit.SECONDS));
  }

  public static class MessageProducer {

    private Thread thread;
    private DirectProcessor subject;
    private volatile boolean stopped;
    private int intervalMillis;

    public MessageProducer(final int intervalMillis) {
      this.subject = DirectProcessor.create();
      this.intervalMillis = intervalMillis;
      this.thread = new Thread() {
        @Override
        public void run() {
          int i = 0;
          while (true) {
            subject.onNext(i++);
            LockSupport.parkNanos(MessageProducer.this.intervalMillis * 1_000_000);
            if (stopped || Thread.interrupted())
              return;
          }
        }
      };
      thread.setDaemon(true);
    }

    public Flux getObservable() {
      return subject;
    }

    public void start() {
      if (!thread.isAlive()) {
        thread.start();
      }
    }

    public void stop() {
      thread.interrupt();
      stopped = true;
    }
  }

  public static class MessageConsumer {
    public static final Object SLOWDOWN_MARKER = new Object();

    private volatile int waitMillis = 0;

    public MessageConsumer() {
    }

    public void setWaitMillis(int waitMillis) {
      this.waitMillis = waitMillis;
    }

    public void consume(Object x) {
      if (waitMillis > 0) {
        LockSupport.parkNanos(waitMillis * 1_000_000);
      }
    }
  }

}