jbrisbin

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

def stream2 = AmqpStreams.fromQueue('test')

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

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

  @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

public interface PersonStream extends ReactiveStream<Person> {

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

}

@ReactiveRestController
public class MyController {

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

<?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

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

interface Channel<IN, OUT> extends Publisher<IN> {
	
	Publisher<Boolean> write(Publisher<? extends OUT> data);

}

interface ClientSocketOptions {
	
	SocketAddress connectAddress();

jbrisbin

public interface TcpConnection<R, W> {

  Reader<R> reader();

  Writer<W> writer();

  public interface Reader<R> extends Publisher<R> {
    <NEWR> Reader<NEWR> intercept(Interceptor<R, NEWR> interceptor);
  }

jbrisbin

ReactorTcpServer.listen(3000, ByteBuf.class)
                .log("connection")
                .consume(conn -> conn.out(conn.in().log("in"))
                                     .log("confirmation")
                                     .consume(buf -> LOG.info("write confirmed: {}", buf)));