[ISSUE apache#199]Support rsqldb (apache#200)

* feat(nest join) support nest join

* maintain(example) remove MqttSourceExample

Author: ni-ze

rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/sinkcache/impl/MessageCache.java

@@ -26,6 +26,7 @@
 import org.apache.rocketmq.streams.common.channel.sinkcache.DataSourceAutoFlushTask;
 import org.apache.rocketmq.streams.common.channel.sinkcache.IMessageCache;
 import org.apache.rocketmq.streams.common.channel.sinkcache.IMessageFlushCallBack;
+import org.apache.rocketmq.streams.common.context.Message;
 import org.apache.rocketmq.streams.common.schedule.ScheduleManager;
 import org.apache.rocketmq.streams.common.schedule.ScheduleTask;
 

rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/context/MessageHeader.java

@@ -99,6 +99,8 @@ public class MessageHeader {
 
     protected String msgRouteFromLable;//消息从哪里来的标签，标记上游节点的标记，主要是通过build table name来标记
 
+    private String originTable;
+
     protected String logFingerprintValue;//日志指纹的值
 
     public MessageHeader copy() {
@@ -359,4 +361,12 @@ public String getPipelineName() {
     public void setPipelineName(String pipelineName) {
         this.pipelineName = pipelineName;
     }
+
+    public String getOriginTable() {
+        return originTable;
+    }
+
+    public void setOriginTable(String originTable) {
+        this.originTable = originTable;
+    }
 }

rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/topology/ChainPipeline.java

@@ -290,6 +290,7 @@ public void doNextStages(AbstractContext context, String msgPrevSourceName, Stri
             //boolean needFlush = needFlush(msg);
             if (StringUtil.isNotEmpty(oriMsgPrewSourceName)) {
                 msg.getHeader().setMsgRouteFromLable(oriMsgPrewSourceName);
+                msg.getHeader().setOriginTable(oriMsgPrewSourceName);
             }
             boolean isContinue = executeStage(stage, msg, copyContext);
 

rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/topology/stages/JoinChainStage.java

@@ -44,8 +44,14 @@ public class JoinChainStage<T extends IMessage> extends AbstractWindowStage<T> {
         @Override
         protected IMessage doProcess(IMessage message, AbstractContext context) {
             String lable = message.getHeader().getMsgRouteFromLable();
+            String originTable = message.getHeader().getOriginTable();
+
             String joinFlag = null;
             if (lable != null) {
+                if ((lable.equals("left") || lable.equals("right")) && originTable != null) {
+                    lable = originTable;
+                }
+
                 if (lable.equals(rightDependentTableName)) {
                     joinFlag = MessageHeader.JOIN_RIGHT;
                 } else {
@@ -61,9 +67,7 @@ protected IMessage doProcess(IMessage message, AbstractContext context) {
             } else {
                 rightPipeline.doMessage(message, context);
             }
-            //if(!MessageGloableTrace.existFinshBranch(message)){
-            //    context.setBreak(true);
-            //}
+
             context.breakExecute();
             return message;
         }

rocketmq-streams-examples/src/main/java/org/apache/rocketmq/streams/examples/source/MqttSourceExample.java

@@ -102,7 +102,6 @@ public void shuffleCalculate(List<IMessage> messages, WindowInstance instance, S
 
             if (WindowJoinType.left.name().equalsIgnoreCase(routeLabel)) {
                 storage.putWindowBaseValue(queueId, windowInstanceId, WindowType.JOIN_WINDOW, WindowJoinType.left, temp);
-
             } else if (WindowJoinType.right.name().equalsIgnoreCase(routeLabel)) {
                 storage.putWindowBaseValue(queueId, windowInstanceId, WindowType.JOIN_WINDOW, WindowJoinType.right, temp);
             } else {
@@ -490,6 +489,22 @@ protected void sendMessage(JSONObject message, boolean needFlush) {
         if (needFlush) {
             nextMessage.getHeader().setNeedFlush(true);
         }
+
+        String routeLabel = nextMessage.getHeader().getMsgRouteFromLable();
+        if (routeLabel == null) {
+            //嵌套join，内部join后没有routeLabel，需要设置结果的routeLabel
+            String configureName = this.getConfigureName();
+            String[] tempList = configureName.split("_");
+            for (int i = tempList.length -1; i > 0; i--) {
+                if ("left".equalsIgnoreCase(tempList[i]) || "right".equalsIgnoreCase(tempList[i])) {
+                    routeLabel = tempList[i];
+                    System.out.println("nested join, routeLabel=" + routeLabel);
+                    break;
+                }
+            }
+            nextMessage.getHeader().setMsgRouteFromLable(routeLabel);
+        }
+
         AbstractContext context = new Context(nextMessage);
         boolean isWindowTest = ComponentCreator.getPropertyBooleanValue("window.fire.isTest");
         if (isWindowTest) {

rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/join/JoinWindow.java

@@ -229,6 +229,11 @@ public void addNewSplit(IMessage message, AbstractContext context, NewSplitMessa
         for (String splitId : splitIds) {
              this.loadResult.put(splitId, future);
         }
+
+        if (message.getHeader().isSystemMessage() && window.getFireReceiver() == null) {
+            return;
+        }
+
         window.getFireReceiver().doMessage(message, context);
     }