$ java -Dreactor.profiles.default=production -jar reactor-app-1.0.jar

# change to the eventLoop for a default Dispatcher
reactor.dispatchers.default = eventLoop

 
   
 
    
Tuple2<String,Float> tuple = Tuple.of("Hello World!",Float.valueOf(1.0));
reactor.notify("topic",Event.wrap(tuple));
 
   
 
  

AStreamis a stateless event processor that offers a reactive alternative to callback spaghetti programming. Values passing through a Stream can be transformed,filtered,and consumed by calling the appropriate method. Entire graphs of operations can be constructed from a Stream since many methods transparently return new Streams which start new flows of reactions.

Streams aren’t created directly since aStreamis a consumer-side concern. The producer side is called aDeferred. TheDeferredis the object that ties the producer to the consumer and it’s this object that data is passed into. ADeferredis also aConsumer,so to pass data into aStream,call theaccept(T)method.

Generally the two halves of the Deferred/Stream combination are not in the same scope. If you’re creating an API for your application,you would likely create aDeferredinside a method call and retain a reference to that Deferred so you can publish values to it asynchronously. You would then return aStream<T>to the caller so it could interact with the values.

The real power of the Stream API lies in the composition functions for transforming and filtering data.

ThefilteredStream now contains only lowercased values that start with "nsprefix:".

Streams are by default unbounded,which means they have no beginning and no end. Streams also have a method calledbatch(int size)which creates a newStreamthat works with values in batches. In batch mode,Streammethods likefirst()andlast()make it easy to work with the beginning and ends of batches.

Thelast()methods provide Streams whose values are only the first of the Stream,in the case of an unbounded Stream,or the first value of each batch,if the Stream is in batch mode. The Stream returned fromlast()contains values from the end of a Stream (in unbounded mode,that means whenenverflush()is called) or the last values in each batch if the Stream is in batch mode.

Values can be collected into aList<T>by callingStream.collect(). This new Stream is populated by accumulated values wheneverflush()is called or the configured batch size is reached.

Streams can also be reduced. Provided with aSupplier<T>which returns a new instance of the accumulator at the beginning of the Stream or the beginning of each batch. ATuple2is passed into the providedFunctioncontaining the current accumulator value and the next data element passing through the Stream. The value returned from the providedFunctionwill be sent along again as the accumulator value the next time a data value passes through the Stream.

APromiseis a stateful event processor that accepts either a single value or an error. Once either of these values has been accepted,monospace; font-size:13.6000003814697px; padding:0.2em 0px; margin:0px">Promiseis considered complete.

When aPromiseis populated with a value,monospace; font-size:13.6000003814697px; padding:0.2em 0px; margin:0px">onSuccessConsumers are invoked. When populated with an error,monospace; font-size:13.6000003814697px; padding:0.2em 0px; margin:0px">onErrorConsumers are invoked. When either a success or an error is set,monospace; font-size:13.6000003814697px; padding:0.2em 0px; margin:0px">onCompleteConsumers are invoked.

Promises aren’t created directly. They’re meant to handle future values when they are available. Promises only provide methods for setting Consumers—you can’t publish a value to a Promise. Instead you use aDeferred,which produces aPromisefrom itscompose()method. The Deferred has anaccept()method on it which is used to populate the Promise.

Promises also need to be aware of aDispatcherto use when publishing events. There are static helper methods on thePromisesclass for creating them.

Promises can also be created in a success or error state right away if the value is already known. This is useful when a Promise should be returned from a method but the value (or error) is immediately available.

A key feature of Promises is that they are composable. Themap(Function<T,V> fn)function can be used to assign aFunction<T,V>that will transform the result valueTinto aVand populate a newPromise<V>with the return value.

A Promise can also be filtered. A call to thefilter(Predicate<T> p)method will return a newPromise<T>that will,depending on the Predicate check,either be populated with the result value or be in error if the Predicate test fails.

Since Promises are designed to be stateful,they also provide methods for getting the value of the Promise. Promises areSuppliers,so a call to theget()method will return the current value of the Promise,whether it is complete or not (Promises also provide methods for checking whether the Promise is complete,a success,or in error).

Promises can block the calling thread waiting for a result. Call one of theawait()methods to block until a value is available or the Promise is notified of an error.

Note that the zero-argawait()method doesnotblock indefinitely. It will only block for a period of milliseconds as defined in the System propertyreactor.await.defaultTimeout. This value is 30 seconds by default. It can be changed by setting this property in theMETA-INF/reactor/default.properties(or any other activeprofileName.properties).

TheProcessoris a thin abstraction around theLMAX Disruptor RingBufferdesigned for high performance. Unlike aProcessorhas no awareness of dispatching or Selectors and doesn’t directly support dynamic Consumer assignment (that can be achieved by using theDelegatingConsumer). The primary goal of theProcessoris to expose the power of the Disruptor RingBuffer as closely to the core as possible,without introducing unnecessary overhead.

Creation of aProcessorhappens through aProcessorSpec:

TheProcessorcan also handle operations in batches of any size (though we’ve found that smaller batches generally offer higher throughput) using thebatch(int size,Consumer<Frame> mutator)method. Here’s an example of creating a batch the same size as theList<Message> eventsof incoming messages:

There’s no need to callcommit()for each operation in batch mode. TheConsumer<T>passed into the batch method is a mutator whose purpose is to set the values of the data object before it is implicitly committed (published).

Reactor provides out-of-the-box support for Spring ApplicationContexts by providing aBeanPostProcessorimplementation that finds annotated beans and wires them into Reactors using SpEL and also provides some helperFactoryBeanimplementations for creating ReactorEnvironmentandReactorinstances.

Since creating the initialEnvironmentis a standard part of using Reactor in any application,monospace; font-size:13.6000003814697px; padding:0.2em 0px; margin:0px">reactor-springprovides a JavaConfig annotation that you put on your@Configurationbean to implicitly create anEnvironmentbased on the default properties file bootstrapping mechanism. You don’t have to create anEnvironmentbean explicitly if you use@EnableReactor.

Using the@EnableReactorannotation also configures theBeanPostProcessorto post-process your beans that have methods annotated with the@Selectorannotation. Here’s an example of a POJO bean definition that uses annotations to consume events published to aReactorbean defined in the sameApplicationContext:

Any other components who also have the sameReactorinjected into them can publish events to it,while the POJO handler beans can handle the events.

If you’re using annotated handler beans as Consumers using the@Selectorannotation,your method can also serve as a request/reply handler by returning a value. To tell the BeanPostProcessor where to send the return value,monospace; font-size:13.6000003814697px; padding:0.2em 0px; margin:0px">@ReplyTo("topic")annotation on your handler method.

Environment env = new Environment();

// create a spec using the Netty-based client
TcpClient<String,String> client = new TcpClientSpec<String,String>(NettyTcpClient.class)
    .env(env)
    .codec(StandardCodecs.LINE_FEED_CODEC)
    .connect("localhost",8080)
    .get();

client.open().consume(new Consumer<TcpConnection<String,String>>() {
  public void accept(TcpConnection<String,String> conn) {

    conn.in().consume(new Consumer<String>() {
      public void accept(String line) {
        // handle lines of incoming data
      }
    });

    // write data to the server
    conn.send("Hello World!");

  }
});

Environment env = new Environment();

TcpClient<String,8080)
    .get();

Stream<TcpConnection<String,String>> connections = client.open(new Reconnect() {
  public Tuple2<InetSocketAddress,Long> reconnect(InetSocketAddress currentAddress,int attempt) {

    // try reconnecting 3 times
    switch (attempt) {
      case 1:
        return Tuple.of(currentAddress,100L);
      case 2:
        return Tuple.of(currentAddress,500L);
      case 3:
        return Tuple.of(currentAddress,1000L);
      default:
        // try to connect somewhere else
        return Tuple.of(nextHost(currentAddress),0L);
    }

  }
});

connections.consume(new Consumer<TcpConnection<String,String> conn) {
    // handle each connection,including reconnects
  }
});

connections.consume(new Consumer<TcpConnection<String,String>>() {
  public void accept(final TcpConnection<String,String> conn) {

    // handle each connection,including reconnects
    conn.on()
      .close(new Runnable() {
        public void run() {
          // handle connection close event
        }
      })
      .readIdle(5000,new Runnable() {
        public void run() {
          // no data read for 5 seconds,ping client
          conn.send("ping");
        }
      })
      .writeIdle(5000,new Runnable() {
        public void run() {
          // no data written for 5 seconds,ping client
          conn.send("ping");
        }
      })

  }
});

Environment env = new Environment();

// create a spec using the Netty-based server
TcpServer<String,String> server = new TcpServerSpec<String,String>(NettyTcpServer.class)
  .env(env)
  .codec(StandardCodecs.LINE_FEED_CODEC)
  .consume(new Consumer<NetChannel<String,String>>() {
    public void accept(NetChannel<String,String> conn) {

      // for each connection,process incoming data
      conn.in().consume(new Consumer<String>() {
        public void accept(String line) {
          // handle line feed data
        }
      });

    }
  })
  .get();

server.start();

package reactor.tcp.encoding;

import reactor.function.Consumer;
import reactor.function.Function;
import reactor.io.Buffer;

import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CharsetEncoder;

public class StringCodec implements Codec<Buffer,String,String> {

  private final Charset utf8 = Charset.forName("UTF-8");

  @Override
  public Function<Buffer,String> decoder(Consumer<String> next) {
    return new StringDecoder(next);
  }

  @Override
  public Function<String,Buffer> encoder() {
    return new StringEncoder();
  }

  private class StringDecoder implements Function<Buffer,String> {
    private final Consumer<String> next;
    private final CharsetDecoder decoder = utf8.newDecoder();

    private StringDecoder(Consumer<String> next) {
      this.next = next;
    }

    @Override
    public String apply(Buffer bytes) {
      try {
        String s = decoder.decode(bytes.byteBuffer()).toString();
        if (null != next) {
          next.accept(s);
          return null;
        } else {
          return s;
        }
      } catch (CharacterCodingException e) {
        throw new IllegalStateException(e);
      }
    }
  }

  private class StringEncoder implements Function<String,Buffer> {
    private final CharsetEncoder encoder = utf8.newEncoder();

    @Override
    public Buffer apply(String s) {
      try {
        ByteBuffer bb = encoder.encode(CharBuffer.wrap(s));
        return new Buffer(bb);
      } catch (CharacterCodingException e) {
        throw new IllegalStateException(e);
      }
    }
  }

}

TcpServer<String,String>()
  .env(env)
  .codec(StandardCodecs.LINE_FEED_CODEC)
  .consume(new Consumer<NetChannel<String,String>>() {
    public void accept(final NetChannel<String,process incoming data
      conn.in().consume(new Consumer<String>() {
        public void accept(String line) {
          // handle line feed data

          // respond to client (newline will be added by codec)
          conn.send("Hello World!");
        }
      });

    }
  })
  .get()
  .start();

Consumer<NetChannel<String,String>> consumer = new Consumer<NetChannel<String,String>>() {
  public void accept(final NetChannel<String,String> conn) {

    final Consumer<String> out = conn.out();

    // result will be accepted into Consumer and passed to client
    service.doLongLookup(out);

  }
};