Feature: batch import files

src/main/java/com/mokasz/pool/image/ImportUrlsFromFileThreadedApplication.java

@@ -0,0 +1,82 @@
+package com.mokasz.pool.image;
+
+import com.mokasz.pool.image.worker.StoreUrlWorker;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.util.NumberUtils;
+
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+@SpringBootApplication
+@Slf4j
+public class ImportUrlsFromFileThreadedApplication implements CommandLineRunner {
+    BeanFactory beanFactory;
+
+    ImportUrlsFromFileThreadedApplication (BeanFactory beanFactory) {
+        this.beanFactory = beanFactory;
+    }
+
+    public static void main(String[] args) {
+        SpringApplication.run(ImportUrlsFromFileThreadedApplication.class, args);
+    }
+
+    private static void usage () {
+        System.err.println("Usage: <FileName> [Offset=0] [Thread=32]");
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T extends Number> T convertNumber (String[] args, int index, T defaultValue) {
+        try {
+            return NumberUtils.parseNumber(args[index], (Class<? extends T>) defaultValue.getClass());
+        } catch (NumberFormatException | ArrayIndexOutOfBoundsException e) {
+            return defaultValue;
+        }
+    }
+
+    @Override
+    public void run (String... args) {
+        if (args.length < 1) {
+            usage();
+            return;
+        }
+        final String file = args[0].trim();
+        final long skip = convertNumber(args, 1, 0L);
+        final int threads = convertNumber(args, 2, 32);
+
+        ExecutorService executorService = Executors.newFixedThreadPool(threads);
+        long count = 0L;
+        int buffer = 0;
+        try (BufferedReader br = new BufferedReader(new FileReader(file))) {
+            String line;
+            while ((line = br.readLine()) != null) {
+                count++;
+                if (count <= skip) continue;
+                // process the line.
+                if (line.startsWith("http")) {
+                    StoreUrlWorker worker = beanFactory.getBean(StoreUrlWorker.class);
+                    worker.setUrl(line);
+                    executorService.execute(worker);
+                }
+                if (++buffer == threads) {
+                    executorService.shutdown();
+                    while (!executorService.isTerminated()) {
+                        if (!executorService.awaitTermination(3L, TimeUnit.SECONDS)) log.info("Wait");
+                    }
+                    log.info("Done: {}", count);
+                    executorService = Executors.newFixedThreadPool(threads);
+                    buffer = 0;
+                }
+            }
+        } catch (IOException | InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+}

src/main/java/com/mokasz/pool/image/TestProgram.java

@@ -0,0 +1,84 @@
+package com.mokasz.pool.image;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.http.HttpResponse;
+import org.apache.http.NameValuePair;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.entity.UrlEncodedFormEntity;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.message.BasicNameValuePair;
+
+import javax.xml.bind.DatatypeConverter;
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+
+/**
+ * @author wudi
+ */
+@Slf4j
+public class TestProgram {
+    public static void main(String[] args) {
+        if (args.length < 1) {
+            System.err.println("Usage: <urls.txt>");
+            return;
+        }
+        long count = 0;
+        long skip = args.length >= 2 ? DatatypeConverter.parseLong(args[1]) : 0;
+        ExecutorService executorService = Executors.newFixedThreadPool(16);
+        AtomicLong finished = new AtomicLong(0);
+        try (BufferedReader br = new BufferedReader(new FileReader(args[0]))) {
+            String line;
+            while ((line = br.readLine()) != null) {
+                count++;
+                if (count <= skip) continue;
+                // process the line.
+                if (line.startsWith("http")) {
+                    executorService.execute(new HttpWorker(line, finished));
+                }
+                while (!executorService.isTerminated()) {
+                    if (executorService.awaitTermination(100L, TimeUnit.MILLISECONDS)) log.info("Done: {}", finished);
+                }
+            }
+            executorService.shutdown();
+        } catch (IOException | InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+}
+
+@Slf4j
+class HttpWorker implements Runnable {
+    private final String url;
+    private final AtomicLong counter;
+
+    HttpWorker(String url, AtomicLong counter) {
+        this.url = url;
+        this.counter = counter;
+    }
+
+    @Override
+    public void run() {
+        try {
+            HttpClient httpClient = HttpClientBuilder.create().build();
+            HttpPost httpPost = new HttpPost("http://image-pool-service-svc:8080/service/image/store/url");
+            List<NameValuePair> params = new ArrayList<>();
+            params.add(new BasicNameValuePair("url", url));
+            httpPost.setEntity(new UrlEncodedFormEntity(params));
+            HttpResponse response = httpClient.execute(httpPost);
+            log.debug(response.getStatusLine().getReasonPhrase());
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        long done = this.counter.incrementAndGet();
+        if (done % 1000 == 0) log.info("Executed: {}", done);
+    }
+}

src/main/java/com/mokasz/pool/image/worker/StoreUrlWorker.java

@@ -0,0 +1,59 @@
+package com.mokasz.pool.image.worker;
+
+import com.mokasz.pool.image.algorithm.HashTool;
+import com.mokasz.pool.image.algorithm.OpenCvService;
+import com.mokasz.pool.image.restful.service.DownloadBytes;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Service;
+
+import java.io.IOException;
+import java.security.NoSuchAlgorithmException;
+
+@Service
+@Slf4j
+public class StoreUrlWorker implements Runnable {
+
+    private final OpenCvService service;
+
+    private final JdbcTemplate template;
+
+    String url;
+
+    public StoreUrlWorker(OpenCvService service, JdbcTemplate template) {
+        this.service = service;
+        this.template = template;
+    }
+
+    public void setUrl(String url) {
+        this.url = url;
+    }
+
+    @SneakyThrows
+    @Override
+    public void run() {
+        doStoreImage(url, 0);
+    }
+
+    public void doStoreImage(String url, int retry) throws NoSuchAlgorithmException {
+        try {
+            byte[] fileData = DownloadBytes.downloadAsBytesArray(url);
+            String md5 = HashTool.calculateMd5(fileData);
+            float[] floats = new float[8];
+            service.calculatePHash(fileData, floats);
+            float[] binaries = new float[64];
+            service.transformBinaryArray(floats, binaries);
+            String hash = service.getReadableHash(floats);
+            template.update(
+                    "INSERT INTO image_pool.url_hash_alt VALUES (?, ?, ?, NOW()) ON CONFLICT (url) DO NOTHING",
+                    url,
+                    md5,
+                    hash);
+        } catch (IOException e) {
+            if (retry > 3) return;
+            log.warn("Possible bad URL: {}", url);
+            doStoreImage(url, retry + 1);
+        }
+    }
+}