jbrisbin

import org.apache.commons.lang.time.StopWatch;
import reactor.Environment;
import reactor.fn.BiFunction;
import reactor.rx.Streams;

import java.util.concurrent.TimeUnit;

/**
 * Demo class for calculating Pi using Reactor.
 */

jbrisbin

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.mycompany</groupId>
  <artifactId>my-project</artifactId>
  <version>1.0.0.BUILD-SNAPSHOT</version>

jbrisbin

static {
  // Only done once, statically, and shared across this classloader
  Environment.initialize();
}

// Create a Stream subclass we can sink values into
Broadcaster<String> b = Broadcaster.create();

b
  // dispatch onto a Thread other than 'main'

jbrisbin

public interface PersonStream extends ReactiveStream<Person> {

  @MapAction("@personRepo.merge($1, $2)")
  ReactivePromise<Person> merge(String id, Person p);

}

@ReactiveRestController
public class MyController {

jbrisbin

  @Bean
  public Action<Chain> handlers(PersonRepository persons,
                                HotStream<Person> personStream,
                                ObjectMapper jsonMapper,
                                ModelMapper beanMapper) {
    return (chain) -> {
      chain.handler("person", ctx ->
          ctx.byMethod(spec ->
             spec
                 .get(c -> {

jbrisbin

	@Bean
	public Renderer<Stream> personRenderer() {
		return new RendererSupport<Stream>() {
			@Override
			public void render(Context ctx, Stream s) throws Exception {
				ctx.promise(f -> s.consume(f::success))
				   .then(o -> ctx.render(json(o)));
			}
		};
	}

jbrisbin

def stream1 = AmqpStreams.fromQueue(Queue.lookup('test')
                                         .durable(true)
                                         .bind('test'))

def stream2 = AmqpStreams.fromQueue('test')

Streams.merge(stream1, stream2)
       .dispatchOn(environment)

jbrisbin

	@Test
	public void workerOrchestrator() throws InterruptedException {
		Environment env = new Environment();

		Reactor reactor = Reactors.reactor(env, Environment.THREAD_POOL);

		CountDownLatch latch = new CountDownLatch(2);

		reactor.on(Selectors.$("worker"), new Consumer() {
			@Override

jbrisbin

routing {
  include SpringConfig

  http {
    route("/path/to/{resourceId}") to CustomResource
  }
}

interface Context extends AbstractMutableMap<String, Object> {

jbrisbin

  _  _ _____ _____ ___ ___      _____ _   ____
 | || |_   _|_   _| _ \ \ \    / / __(_) / / /
 | __ | | |   | | |  _/ |\ \/\/ /\__ \_ / / /
 |_||_| |_|   |_| |_| | | \_/\_/ |___(_)_/_/
 =====================|_|=====================
 HTTP and WebSocket interactive shell - version 1.0

(http://localhost:8080/):> get --headers "X-Custom: custom header value"
> GET http://localhost:8080/ HTTP/1.1
>