PipelinR

PipelinR is a lightweight command processing pipeline ❍ ⇢ ❍ ⇢ ❍ for your awesome Java app.

PipelinR has been battle-proven on production, as a service layer in some cool FinTech apps. PipelinR has helped teams switch from a giant service classes handling all use cases to small handlers following single responsibility principle. It's similar to a popular MediatR library for .NET.

💡 Join Effective Java Software Design course to learn more about building great Java enterprise applications.

Table of contents

• How to use
• Commands
• Handlers
• Pipeline
• Notifications
• Spring Example
• Async
• How to contribute
• Alternatives
• Contributors

How to use

PipelinR has no dependencies. All you need is a single 15KB library:

Maven:

<dependency>
  <groupId>an.awesome</groupId>
  <artifactId>pipelinr</artifactId>
  <version>0.5</version>
</dependency>

<repositories>
  <repository>
    <id>central</id>
    <name>bintray</name>
    <url>https://jcenter.bintray.com</url>
  </repository>
</repositories>

Gradle:

repositories {
    jcenter()
}

dependencies {
    compile 'an.awesome:pipelinr:0.5'
}

Java version required: 1.8+.

Commands

Commands is a request that can return a value. The Ping command below returns a string:

class Ping implements Command<String> {

    public final String host;
    
    public Ping(String host) {
        this.host = host;
    }
}

If a command has nothing to return, you can use a built-in Voidy return type:

class Ping implements Command<Voidy> {

    public final String host;
    
    public Ping(String host) {
        this.host = host;
    }
}

Handlers

For every command you must define a Handler, that knows how to handle the command.

Create a handler by implementing Command.Handler<C, R> interface, where C is a command type and R is a return type. Handler's return type must match command's return type:

class Pong implements Command.Handler<Ping, String> {

    @Override
    public String handle(Ping command) {
        return "Pong from " + command.host;
    }
}

Pipeline

A pipeline mediates between commands and handlers. You send commands to the pipeline. When the pipeline receives a command, it sends the command through a sequence of middlewares and finally invokes the matching command handler. Pipelinr is a default implementation of Pipeline interface.

To construct a Pipeline, create an instance of Pipelinr and provide a list of command handlers:

Pipeline pipeline = new Pipelinr()
    .with(
        () -> Stream.of(new Pong())
    );

Send a command for handling:

pipeline.send(new Ping("localhost"));

since v0.4, you can execute commands more naturally:

new Ping("localhost").execute(pipeline);

Pipelinr can receive an optional, ordered list of custom middlewares. Every command will go through the middlewares before being handled. Use middlewares when you want to add extra behavior to command handlers, such as logging, transactions or metrics:

// middleware that logs every command and the result it returns
class Loggable implements Command.Middleware {

    @Override
    public <R, C extends Command<R>> R invoke(C command, Next<R> next) {
        // log command
        R response = next.invoke();
        // log response
        return response;
    }
}

// middleware that wraps a command in a transaction
class Transactional implements Command.Middleware {

    @Override
    public <R, C extends Command<R>> R invoke(C command, Next<R> next) {
        // start tx
        R response = next.invoke();
        // end tx
        return response;
    }
}

In the following pipeline, every command and its response will be logged, plus commands will be wrapped in a transaction:

Pipeline pipeline = new Pipelinr()
    .with(() -> Stream.of(new Pong())),
    .with(() -> Stream.of(new Loggable(), new Transactional()))
);

By default, command handlers are being resolved using generics. By overriding command handler's matches method, you can dynamically select a matching handler:

class LocalhostPong implements Command.Handler<Ping, String> {

    @Override
    public boolean matches(Ping command) {
        return command.host.equals("localhost");
    }

}

class NonLocalhostPong implements Command.Handler<Ping, String> {
    
    @Override
    public boolean matches(Ping command) {
        return !command.host.equals("localhost");
    } 
}

Notifications

Since version 0.5, PipelinR supports Notifications, dispatched to multiple handlers.

For notifications, first create your notification message:

class Ping implements Notification {
}

Next, create zero or more handlers for your notification:

public class Pong1 implements Notification.Handler<Ping> {

    @Override
    public void handle(Ping notification) {
      System.out.printn("Pong 1");
    }
}

public class Pong2 implements Notification.Handler<Ping> {

    @Override
    public void handle(Ping notification) {
      System.out.printn("Pong 2");
    }
}

Finally, send notification to the pipeline:

new Ping().send(pipeline);

💡 Remember to provide notification handlers to PipelinR:

new Pipelinr()
  .with(
    () -> Stream.of(new Pong1(), new Pong2())
  )

Notification middlewares

Notifications, like commands, support middlewares. Notification middlewares will run before every notification handler:

class Transactional implements Notification.Middleware {

    @Override
    public <N extends Notification> void invoke(N notification, Next next) {
        // start tx
        next.invoke();
        // stop tx
    }
}

new Pipelinr().with(() -> Stream.of(new Transactional()))

Notification handling strategies

The default implementation loops through the notification handlers and awaits each one. This ensures each handler is run after one another.

Depending on your use-case for sending notifications, you might need a different strategy for handling the notifications, such running handlers in parallel.

PipelinR supports the following strategies:

• an.awesome.pipelinr.StopOnException (default)
• an.awesome.pipelinr.ContinueOnException
• an.awesome.pipelinr.Async
• an.awesome.pipelinr.ParallelNoWait
• an.awesome.pipelinr.ParallelWhenAny
• an.awesome.pipelinr.ParallelWhenAll

See each class' JavaDocs for the details.

You can override default strategy via:

new Pipelinr().with(new ContinueOnException());

Spring Example

PipelinR works well with Spring and Spring Boot.

Start by configuring a Pipeline. Create an instance of Pipelinr and inject all command handlers and ordered middlewares via the constructor:

@Configuration
class PipelinrConfiguration {

    @Bean
    Pipeline pipeline(ObjectProvider<Command.Handler> commandHandlers, ObjectProvider<Notification.Handler> notificationHandlers, ObjectProvider<Command.Middleware> middlewares) {
        return new Pipelinr()
          .with(commandHandlers::stream)
          .with(notificationHandlers::stream)
          .with(middlewares::orderedStream);
    }
}

Define a command:

class Wave implements Command<String> {
}

Define a handler and annotate it with @Component annotation:

@Component
class WaveBack implements Command.Handler<Wave, String> {
    // ...
}

Optionally, define Order-ed middlewares:

@Component
@Order(1)
class Loggable implements Command.Middleware {
    // ...
}

@Component
@Order(2)
class Transactional implements Command.Middleware {
    // ...
}

To use notifications, define a notification:

class Ping implements Notification {
}

Define notification handlers and annotate them with @Component annotation:

@Component
class Pong1 implements Notification.Handler<Ping> {
    // ...
}

@Component
class Pong2 implements Notification.Handler<Ping> {
    // ...
}

Remember that notifications, like commands, also support Middlewares.

We're ready to go! Inject Pipeline into your application, and start sending commands or notifications:

class Application {

    @Autowired
    Pipeline pipeline;

    public void run() {
        String response = new Wave().execute(pipeline);
        System.out.println(response); 
        
        // ... or
        
        new Ping().send(pipeline); // should trigger Pong1 and Pong2 notification handlers
        
    }
}

Async

PipelinR works well in async or reactive applications. For example, a command can return CompletableFuture:

class AsyncPing implements Command<CompletableFuture<String>> {
    
    @Component
    static class Handler implements Command.Handler<AsyncPing, CompletableFuture<String>> {

        @Override
        public CompletableFuture<String> handle(AsyncPing command) {
            return CompletableFuture.completedFuture("OK");
        }
    }
}

Sending AsyncPing to the pipeline returns CompletableFuture:

CompletableFuture<String> okInFuture = new Ping().execute(pipeline);

How to contribute

Just fork the repo and send us a pull request.

Alternatives

• MediatR – Simple, unambitious mediator implementation in .NET

Contributors

• @sizovs as Eduards Sizovs (Blog)