package com.tpg.nnode;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.stream.ActorMaterializer;
import akka.stream.actor.AbstractActorPublisher;
import akka.stream.javadsl.Source;
import com.tpg.nnode.rabbit.RabbitQueueSourceActor;
import com.tpg.rabbit.RabbitConn;
import org.reactivestreams.Publisher;
import scala.runtime.BoxedUnit;

import java.io.IOException;


public class ProcessingApp {

    public static void main(String[] args) throws IOException {

        ActorSystem aSys = ActorSystem.create("aSys");
        ActorMaterializer aMaterializer = ActorMaterializer.create(aSys);

        ActorRef producer = aSys.actorOf(Props.create(RabbitQueueSourceActor.class), "producer");
        Publisher<RabbitQueueSourceActor.RabbitMsg> pub = AbstractActorPublisher.create(producer);

        final ActorRef listener = aSys.actorOf(Props.create(ListenerActor.class, () -> new ListenerActor(producer)),
                "listener");
        RabbitConn.setUpRabbit(aSys, listener);

        final Source<RabbitQueueSourceActor.RabbitMsg, BoxedUnit> rabbitMessagesSrc = Source.from(pub);
        rabbitMessagesSrc.runForeach(
                m -> {
                    final String rm = m.getMsg();
                    System.out.println(String.format("Sink-ed: %s", rm));
                },
                aMaterializer);

        System.in.read();
        aSys.shutdown();

    }
}

@@ -294,25 +294,216 @@
 tpg.
-rabbit.RabbitConn
+pnode.rules.PasswordRule;%0Aimport com.tpg.pnode.rules.Rule;%0Aimport com.tpg.pnode.rules.RuleSet;%0Aimport com.tpg.pnode.rules.RuleSetBuilder;%0Aimport com.tpg.rabbit.RabbitConn;%0Aimport org.easyrules.api.RulesEngine
 ;%0Aim
@@ -599,16 +599,38 @@
 eption;%0A
+import java.util.Set;%0A
 %0A%0Apublic
@@ -1237,24 +1237,175 @@
 listener);%0A%0A
+        final RuleSet reSet = (new RuleSetBuilder()).add(new PasswordRule()).build();%0A        final RulesEngine rulesEngine = reSet.getRulesEngine();%0A%0A
         fina
@@ -1603,24 +1603,24 @@
 m.getMsg();%0A
-
             
@@ -1681,16 +1681,101 @@
 , rm));%0A
+                    reSet.setInput(rm);%0A                    rulesEngine.fireRules();%0A
         

add rules engine to java implementation