Suggestion: a repo of more complex architectural reactive pattern examples

[flowt-au]

I am new to Java and reactive programming. My backend experience has been with CFML (ColdFusion and Lucee) and with Vue.js and Quasar on the front end.

I am loving Quarkus. One thing that is done really well is the sample code examples - they are well explained and they actually work (which for some frameworks can be a bit hit-and-miss).

As a newcomer, one thing I found missing was a translation of general software patterns into the reactive way of thinking. Basically, how to construct pipelines beyond the sample html request -> transform to uppercase -> html response type examples. That is not a criticism - those examples have been really helpful and I am now at a stage of more "real-world" implementation.

One such pattern is the request -> validate -> before -> action -> after -> response where each stage can cancel the remaining ones and jump to a graceful response. The reactive implementation of that pattern was not obvious to me and it occurred to me that a repo of examples of the more complex patterns would help newcomers orient their thinking to the reactive way. It would have sped up the process of learning reactive programming implementation for me, anyway.

I did some Vertx courses, watched the YouTube videos and read the Quarkus docs and blog posts, as well as reading the excellent Reactive Systems in Java book. Still, examples of common architectural patterns done in the reactive way eluded me.

So, as a newbie I offer something in that regard below. It is a minimalist sample to focus on the pipeline aspect of the implementation of the request -> validate -> before -> action -> after -> response pattern. I am sure it can be improved! I have implemented this pattern in the app I am developing and it all works well, so far at least!

To keep it simple the implementation uses a series of url paths to determine what outcome you want to have happen and the terminal logs show the "result" in terms of which stages were executed. e.g. http://localhost:8080/chain/{forcedResult} where forcedResult can be one of ok, crashbefore, cancelbefore, crashaction, cancelaction, crashafter, cancelafter and the use of anything other than those causes a failure in the validation stage.

Again, to keep it as simple as possible while maintaining the focus on the pipeline chaining, the server returns plain strings, not json, and always returns status code 200 (the log messages describe what actually happens).

In my initial confusion I started looking at how to use the subscription cancellation as the "short circuiting" mechanism before I realised I was misunderstanding what "cancel" meant in this context (well... I think I misunderstood!).

The basic logic is: if each stage is a success, pass to the next stage with (optionally) updated data. The "short circuit" is process is: any stage can be "cancelled" ie due to some Business Logic restriction, and any stage can "fail" i.e. some exception (a "crash"). Cancellation and Crash skips to the subscriber "onFailure" (ie via a Uni fail) where the two are differentiated and appropriate action is taken. See: onFailure() in the processRequest method.

The pipeline passes a Packet bean which gets updated at each stage and is expressed in the subscriber. The Packet bean:

package org.acme;

public class Packet {
  Boolean success = false;
  String message = "";
  String data = "";
  String action = "";

  // getters and setters here ...

  @Override
  public String toString() {
    return "action=" + this.getAction() + ", success=" + this.getSuccess() + ", data=" + this.getData() + ", message=" + this.getMessage();
  }
}

The demo RestEasy path handler:
{forcedResult} must be one of:
ok, crashbefore, cancelbefore, crashaction, cancelaction, crashafter, cancelafter

@GET
@Produces(MediaType.TEXT_PLAIN)
@Path("/chain/{forcedResult}")
public Uni<Response> handleRequest(String forcedResult) {
    Log.info("=========== Request with: " + forcedResult);

    // True = return the Response to RestEasy where its subscriber will handle it.
    // False = use our own local subscriber so we can use the success or failure locally,
    //         while returning a canned 200 Response.
    Boolean returnResult = false;

    if (returnResult) {
        // Return the Response Uni so the RestEasy subscriber gets it.
        return processRequest (forcedResult)
            .onFailure().recoverWithUni(fail -> {
                // For this demo, gracefully handle exceptions.
                return Uni.createFrom().item("Exception happenned: " + fail.getMessage());

                // Or, return an empty string to return a StatusCode 500
                // return Uni.createFrom().item("");
            })

            // Build the http response for the RestEasy subscription.
            .onItem()
            .transform( packet -> !packet.equals("") ? Response.ok(packet) : Response.status(Response.Status.INTERNAL_SERVER_ERROR))
            .onItem()
            .transform(Response.ResponseBuilder::build);
    } else {
        // Or... handle the subscription locally to peek at it via the Terminal log.
        processRequest (forcedResult)
        .subscribe().with(
            success -> Log.info("Subscriber success=" + success),
            failure -> Log.error("Subscriber failure=" + failure.getMessage())
        );

        // Return a canned 200 Response Uni for the RestEasy subscriber.
        String msg = "Returning HTTP request with: '" + forcedResult + "' completed";
        Log.info(msg);

        return Uni.createFrom().item(Response.ok(msg).build());
    }
}

The request action handler. ie the pipeline chain bit.

// Handle the request Validate -> Before -> Action -> After pipeline
Uni<String> processRequest (String forcedResult) {
    Packet incomingPacket = new Packet();
    incomingPacket.setAction(forcedResult);

    // Implement the Uni pipeline chain
    return
        validateRequest(incomingPacket)
        .chain(packet -> beforeAction(packet))
        .chain(packet -> action(packet))
        .chain(packet -> afterAction(packet))

        // End of the chain.
        // Transform the packet to a String we can return to satisfy RestEasy.
        .onItem().transformToUni(packet -> Uni.createFrom().item(packet.toString()))

        // Gracefully handle the failure which could either be 
        // an exception or a Business Logic fail eg cancelBefore.
        .onFailure().recoverWithUni(fail -> {
            String failMessage = fail.getMessage();

            // Q&D check for Business Logic "fail". 
            // Normally you would have a JSON packet to work with here
            if (failMessage.contains("action=")) {
                return Uni.createFrom().item("my messsage=" + failMessage);
            } else {
                // Is an exception so treat it as a failure
                // just so we can deal with it as such in handleRequest()
                return Uni.createFrom().failure(fail);
            }
        });
}

The methods for each stage:

// Validate the incoming data (action)
Uni<Packet> validateRequest(Packet incomingPacket) {
    String action = incomingPacket.getAction();

    Log.info("Validating the request for " + action);
    Packet outgoingPacket = new Packet();

    try {
        String validActions = "ok,crashbefore,cancelbefore,crashaction,cancelaction,crashafter,cancelafter";

        // "Validate" the request...
        Boolean valid = validActions.contains(action);

        if (valid) {
            outgoingPacket.setAction(action);
            outgoingPacket.setSuccess(true);
        } else {
            // Cause a failure so the pipeline skips to the final onFailure()
            outgoingPacket.setSuccess(false);
            outgoingPacket.setMessage("Expected action '" + action + "' is unknown.");
            throw new Exception(outgoingPacket.toString());
        }

        return Uni.createFrom().item(outgoingPacket);

    } catch (Exception e) {
        return Uni.createFrom().failure(e);
    }
}

// Do the Before action and optionally "cancel" the subsequent stages
Uni<Packet> beforeAction(Packet incomingPacket) {
    Log.info("In beforeAction()");
    String newData = incomingPacket.getData() + "Some data from BEFORE";
    return helper(incomingPacket.getAction(), "crashbefore", "cancelbefore", newData, "before action");
}

// Do the Action and optionally "cancel" the subsequent stages
Uni<Packet> action(Packet incomingPacket) {
    Log.info("In action()");
    String newData = incomingPacket.getData() + " and some data from ACTION";
    return helper(incomingPacket.getAction(), "crashaction", "cancelaction", newData, "the action");
}

// Do the After action and optionally "cancel" the subsequent stages
Uni<Packet> afterAction(Packet incomingPacket) {
    Log.info("In afterAction()");
    String newData = incomingPacket.getData() + " and some data from AFTER the action";
    return helper(incomingPacket.getAction(), "crashafter", "cancelafter", newData, "after action");
}

// DRY up the code.
Uni<Packet> helper(String action, String crash, String cancel, String data, String msg) {
    Packet outgoingPacket = new Packet();

    try {
        // If asked, pretend we have an "unexpected" coding exception while processing the data
        if (action.equals(crash)) {
            Integer x = (1 / 0);
        }

        outgoingPacket.setAction(action);

        if (action.equals(cancel)) {
            outgoingPacket.setSuccess(false);
            outgoingPacket.setMessage("Sorry, I cancelled " + msg + " because your request is not allowed for business logic reasons.");

            // Cause a failure so the pipeline skips to the final onFailure()
            throw new Exception(outgoingPacket.toString());
        }

        // All is well...
        outgoingPacket.setSuccess(true);
        outgoingPacket.setData(data);

        return Uni.createFrom().item(outgoingPacket);

    } catch (Exception e) {
        return Uni.createFrom().failure(e);
    }
}

Feedback please: if there is a simpler way to implement the pattern I (and probably other newcomers) would love to know about it.

And, whatever the final "best practice" example is, a repo of examples of these kinds of patterns would be a great addition to the "Quarkiverse" :-)

Thanks,
Murray