Merge pull request #273 from NigelWu95/dev

Dev

Author: 吴炳亨

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.qiniu</groupId>
     <artifactId>qsuits</artifactId>
-    <version>8.1.0</version>
+    <version>8.2.1</version>
     <name>qsuits</name>
     <description>qiniu-suits is a efficient tools for qiniu api implemented by java8.</description>
     <url>https://github.com/NigelWu95/qiniu-suits-java</url>

src/main/java/com/qiniu/common/SuitsException.java

@@ -13,19 +13,19 @@ public SuitsException(int statusCode, String error) {
     }
 
     public SuitsException(Exception e, int statusCode) {
-        super(String.join(", ", String.valueOf(statusCode), e.getMessage()));
+        super(String.join(", ", String.valueOf(statusCode), e.getCause().getMessage()));
         this.exception = e;
         this.statusCode = statusCode;
     }
 
     public SuitsException(Exception e, int statusCode, String error) {
-        super(String.join(", ", String.valueOf(statusCode), error, e.getMessage()));
+        super(String.join(", ", String.valueOf(statusCode), error, e.getCause().getMessage()));
         this.exception = e;
         this.statusCode = statusCode;
     }
 
     public SuitsException(SuitsException e, String message) {
-        super(String.join(", ", e.getMessage(), message));
+        super(String.join(", ", e.getCause().getMessage(), message));
         this.exception = e;
         this.statusCode = e.getStatusCode();
     }

src/main/java/com/qiniu/datasource/CloudStorageContainer.java

@@ -222,6 +222,7 @@ public void export(IStorageLister<E> lister, IResultOutput saver, ILineProcess<T
                     errorLogger.error("process objects: {}", lister.getPrefix(), e);
                     if (e.response != null) e.response.close();
                 }
+//                statistics.addAndGet(convertedList.size());
             }
             if (hasNext) {
                 json.addProperty("marker", lister.getMarker());
@@ -411,7 +412,7 @@ private List<IStorageLister<E>> filteredListerByPrefixes(Stream<String> prefixes
         return prefixesLister;
     }
 
-    private void processNodeLister(IStorageLister<E> lister) {
+    void processNodeLister(IStorageLister<E> lister) {
         if (lister.currents().size() > 0 || lister.hasNext()) {
             executorPool.execute(() -> listing(lister));
         } else {

src/main/java/com/qiniu/datasource/DatasourceActor.java

@@ -14,6 +14,7 @@
 import java.time.LocalDateTime;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicLong;
 
 import static com.qiniu.entry.CommonParams.lineFormats;
 
@@ -41,6 +42,7 @@ public abstract class DatasourceActor {
     protected ConcurrentMap<String, IResultOutput> saverMap;
     protected ConcurrentMap<String, ILineProcess> processorMap;
     protected boolean stopped;
+    protected AtomicLong statistics;
     protected ConcurrentMap<String, String> progressMap;
 
     public DatasourceActor(int unitLen, int threads) throws IOException {
@@ -49,6 +51,7 @@ public DatasourceActor(int unitLen, int threads) throws IOException {
         this.threads = threads;
         saverMap = new ConcurrentHashMap<>(threads);
         processorMap = new ConcurrentHashMap<>(threads);
+        statistics = new AtomicLong(0);
         progressMap = new ConcurrentHashMap<>(threads);
     }
 
@@ -71,7 +74,7 @@ public void setRetryTimes(int retryTimes) {
 
     void recordLister(String key, String record) {
         try { FileUtils.createIfNotExists(procedureLogFile); } catch (IOException ignored) {}
-        procedureLogger.info("{}:{}", key, record);
+        procedureLogger.info("{}-|-{}", key, record);
         progressMap.put(key, record);
     }
 

src/main/java/com/qiniu/datasource/FileContainer.java

@@ -118,6 +118,7 @@ private void setDirectoriesAndMap(Map<String, Map<String, String>> directoriesMa
             String end = value == null ? null : value.get("end");
             File tempFile = new File(temp);
             if (!tempFile.exists()) tempFile = new File(realPath, temp);
+            directories = new ArrayList<>();
             if (tempFile.isDirectory()) directories.add(tempFile);
             else throw new IOException(temp + " is not valid directory.");
             String forCheckPath = tempFile.getCanonicalPath() + FileUtils.pathSeparator;
@@ -247,6 +248,7 @@ public void export(IFileLister<E, File> lister, IResultOutput saver, ILineProces
                     errorLogger.error("process objects: {}", lister.getName(), e);
                     if (e.response != null) e.response.close();
                 }
+                statistics.addAndGet(convertedList.size());
             }
             if (hasNext) {
                 json.addProperty("start", lister.currentEndFilepath());
@@ -372,7 +374,7 @@ private List<File> loopForFutures(List<Future<IFileLister<E, File>>> futures) th
     }
 
     private Lock lock = new ReentrantLock();
-    private AtomicInteger integer = new AtomicInteger(threads);
+    private AtomicInteger integer = new AtomicInteger(0);
 
     private List<File> listForNextIteratively(List<File> directories) throws Exception {
         List<Future<IFileLister<E, File>>> futures = new ArrayList<>();
@@ -411,7 +413,7 @@ private List<File> listForNextIteratively(List<File> directories) throws Excepti
                         }
                     } catch (Exception e) {
                         try { FileUtils.createIfNotExists(errorLogFile); } catch (IOException ignored) {}
-                        errorLogger.error("excute lister failed", e);
+                        errorLogger.error("execute lister failed", e);
                     } finally {
                         lock.unlock();
                     }
@@ -496,6 +498,7 @@ private List<IFileLister<E, File>> checkListerInPool(int cValue, int tiny) {
             }
             sleep(1000);
             count++;
+            rootLogger.info("finished count: {}.", statistics.get());
         }
         if (notCheck) return new ArrayList<>();
         else return list;
@@ -506,7 +509,7 @@ private void directoriesListing() throws Exception {
 //            directories = directories.parallelStream().map(this::directoriesFromLister).filter(Objects::nonNull)
 //                    .reduce((list1, list2) -> { list1.addAll(list2); return list1; }).orElse(null);
 //        }
-        while (directories != null && directories.size() > 0) directories = listForNextIteratively(directories);
+        while (directories.size() > 0) directories = listForNextIteratively(directories);
         executorPool.shutdown();
         if (threads > 1) {
             int cValue = threads >= 10 ? threads / 2 : 3;
@@ -557,7 +560,10 @@ private void directoriesListing() throws Exception {
                 list = checkListerInPool(cValue, tiny);
             }
         }
-        while (!executorPool.isTerminated()) sleep(1000);
+        while (!executorPool.isTerminated()) {
+            sleep(1000);
+            rootLogger.info("finished count: {}.", statistics.get());
+        }
     }
 
     @Override

src/main/java/com/qiniu/datasource/TextContainer.java

@@ -66,6 +66,7 @@ private void setUrisAndMap(Map<String, Map<String, String>> urisMap) throws IOEx
             this.urisMap = new HashMap<>(threads);
             this.urisMap.putAll(urisMap);
             int size = this.urisMap.size();
+            uris = new ArrayList<>();
             Iterator<String> iterator = this.urisMap.keySet().stream().sorted().collect(Collectors.toList()).iterator();
             while (iterator.hasNext() && size > 0) {
                 size--;
@@ -162,12 +163,13 @@ public void export(ITextReader<S> reader, IResultOutput saver, ILineProcess<T> p
                 if (HttpRespUtils.checkException(e, 2) < -1) throw e;
                 if (e.response != null) e.response.close();
             }
-            json.addProperty("start", reader.currentEndLine());
+            statistics.addAndGet(convertedList.size());
+            lastLine = reader.currentEndLine();
+            json.addProperty("start", lastLine);
             record = json.toString();
             progressMap.put(reader.getName(), record);
             try { FileUtils.createIfNotExists(procedureLogFile); } catch (IOException ignored) {}
             procedureLogger.info("{}-|-{}", reader.getName(), record);
-            lastLine = reader.currentEndLine();
         }
     }
 
@@ -233,11 +235,11 @@ public void export() throws Exception {
             readers = getReaders(FileUtils.convertToRealPath(path));
         } else {
             if (hasAntiPrefixes) {
-                uris.parallelStream().filter(this::checkPrefix).forEach(this::recordListerByUri);
+                uris = uris.parallelStream().filter(this::checkPrefix).peek(this::recordListerByUri).collect(Collectors.toList());
             } else {
                 uris.parallelStream().forEach(this::recordListerByUri);
             }
-            readers = uris.parallelStream().filter(this::checkPrefix).map(uri -> {
+            readers = uris.parallelStream().map(uri -> {
                 try {
                     return generateReader(uri);
                 } catch (IOException e) {
@@ -252,6 +254,11 @@ public void export() throws Exception {
             } else {
                 executorPool = Executors.newFixedThreadPool(threads);
                 readers.parallelStream().forEach(this::reading);
+                executorPool.shutdown();
+                while (!executorPool.isTerminated()) {
+                    sleep(1000);
+                    rootLogger.info("finished count: {}.", statistics.get());
+                }
                 rootLogger.info("{} finished, results in {}.", info, savePath);
             }
             endAction();

src/main/java/com/qiniu/datasource/TextFileContainer.java

@@ -94,11 +94,12 @@ protected List<ITextReader<File>> getReaders(String path) throws IOException {
         } else {
             throw new IOException("");
         }
-        return files.parallelStream().map(file1 -> {
+        return files.parallelStream().map(pFile -> {
             try {
-                return new TextFileReader(file, null, unitLen);
+                return new TextFileReader(pFile, null, unitLen);
             } catch (IOException e) {
                 e.printStackTrace();
+                errorLogger.error("generate lister failed by {}\t{}", pFile.getPath(), urisMap.get(pFile.getPath()), e);
                 return null;
             }
         }).filter(Objects::nonNull).peek(reader -> recordListerByUri(reader.getName())).collect(Collectors.toList());