extract method

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/side/AsyncReqRow.java

@@ -111,6 +111,12 @@ protected void dealMissKey(Row input, ResultFuture<Row> resultFuture){
         }
     }
 
+    protected void dealCacheData(String key, CacheObj missKeyObj) {
+        if (openCache()) {
+            putCache(key, missKeyObj);
+        }
+    }
+
     @Override
     public void timeout(Row input, ResultFuture<Row> resultFuture) throws Exception {
         StreamRecordQueueEntry<Row> future = (StreamRecordQueueEntry<Row>)resultFuture;

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAsyncReqRow.java

@@ -132,7 +132,6 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                 dealMissKey(inputRow, resultFuture);
                 return;
             }
-
             refData.put(sideInfo.getEqualFieldList().get(i), equalObj);
         }
 

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java

@@ -96,11 +96,7 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                     return;
                 } else if (ECacheContentType.MultiLine == val.getType()) {
                     try {
-                        List<Row> rowList = Lists.newArrayList();
-                        for (Object jsonArray : (List) val.getContent()) {
-                            Row row = fillData(inputRow, jsonArray);
-                            rowList.add(row);
-                        }
+                        List<Row> rowList = getRows(inputRow, null, (List) val.getContent());
                         resultFuture.complete(rowList);
                     } catch (Exception e) {
                         dealFillDataError(resultFuture, e, inputRow);
@@ -127,33 +123,19 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                     resultFuture.completeExceptionally(rs.cause());
                     return;
                 }
-
                 List<JsonArray> cacheContent = Lists.newArrayList();
-
-                int resultSize = rs.result().getResults().size();
-                if (resultSize > 0) {
+                List<JsonArray> results = rs.result().getResults();
+                if (results.size() > 0) {
                     try {
-                        List<Row> rowList = Lists.newArrayList();
-                        for (JsonArray line : rs.result().getResults()) {
-                            Row row = fillData(inputRow, line);
-                            if (openCache()) {
-                                cacheContent.add(line);
-                            }
-                            rowList.add(row);
-                        }
-                        if (openCache()) {
-                            putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, cacheContent));
-                        }
-
+                        List<Row> rowList = getRows(inputRow, cacheContent, results);
+                        dealCacheData(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, cacheContent));
                         resultFuture.complete(rowList);
                     } catch (Exception e){
                         dealFillDataError(resultFuture, e, inputRow);
                     }
                 } else {
                     dealMissKey(inputRow, resultFuture);
-                    if (openCache()) {
-                        putCache(key, CacheMissVal.getMissKeyObj());
-                    }
+                    dealCacheData(key, CacheMissVal.getMissKeyObj());
                 }
 
                 // and close the connection
@@ -166,6 +148,18 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
         });
     }
 
+    protected List<Row> getRows(Row inputRow, List<JsonArray> cacheContent, List<JsonArray> results) {
+        List<Row> rowList = Lists.newArrayList();
+        for (JsonArray line : results) {
+            Row row = fillData(inputRow, line);
+            if (null != cacheContent && openCache()) {
+                cacheContent.add(line);
+            }
+            rowList.add(row);
+        }
+        return rowList;
+    }
+
     @Override
     public Row fillData(Row input, Object line) {
         JsonArray jsonArray = (JsonArray) line;

redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java

@@ -172,18 +172,14 @@ public void accept(List<KeyValue<String, String>> keyValues) {
                         }
                         try {
                             Row row = fillData(inputRow, keyValue);
-                            if (openCache()) {
-                                putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, keyValue));
-                            }
+                            dealCacheData(key,CacheObj.buildCacheObj(ECacheContentType.MultiLine, keyValue));
                             resultFuture.complete(Collections.singleton(row));
                         } catch (Exception e) {
                             dealFillDataError(resultFuture, e, inputRow);
                         }
                     } else {
                         dealMissKey(inputRow, resultFuture);
-                        if (openCache()) {
-                            putCache(key, CacheMissVal.getMissKeyObj());
-                        }
+                        dealCacheData(key,CacheMissVal.getMissKeyObj());
                     }
                 }
             });