noop(cleanup) dedup w/factored-out (dtinit#1392) counter

portability-transfer/build.gradle

@@ -36,6 +36,7 @@ configurations {
 
 dependencies {
     compile project(':portability-api-launcher')
+    compile project(':portability-spi-api')
     compile project(':portability-spi-service')
     compile project(':portability-spi-cloud')
     compile project(':portability-spi-transfer')

portability-transfer/src/main/java/org/datatransferproject/transfer/CallableSizeCalculator.java

@@ -1,7 +1,7 @@
 package org.datatransferproject.transfer;
 
-import java.io.IOException;
-import java.io.InputStream;
+import static org.datatransferproject.spi.api.transport.DiscardingStreamCounter.discardForLength;
+
 import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.Map;
@@ -19,7 +19,9 @@ public class CallableSizeCalculator implements Callable<Map<String, Long>> {
   private final Collection<? extends DownloadableItem> items;
 
   public CallableSizeCalculator(
-      UUID jobId, ConnectionProvider connectionProvider, Collection<? extends DownloadableItem> items) {
+      UUID jobId,
+      ConnectionProvider connectionProvider,
+      Collection<? extends DownloadableItem> items) {
     this.jobId = Objects.requireNonNull(jobId);
     this.connectionProvider = Objects.requireNonNull(connectionProvider);
     this.items = Objects.requireNonNull(items);
@@ -32,26 +34,12 @@ public Map<String, Long> call() throws Exception {
       InputStreamWrapper stream = connectionProvider.getInputStreamForItem(jobId, item);
       long size = stream.getBytes();
       if (size <= 0) {
-        size = computeSize(stream);
+        size = discardForLength(stream.getStream());
       }
 
       result.put(item.getIdempotentId(), size);
     }
 
     return result;
   }
-
-  // Reads the input stream in full
-  private Long computeSize(InputStreamWrapper stream) throws IOException {
-    long size = 0;
-    try (InputStream inStream = stream.getStream()) {
-      byte[] buffer = new byte[1024 * 1024]; // 1MB
-      int chunkBytesRead;
-      while ((chunkBytesRead = inStream.read(buffer)) != -1) {
-        size += chunkBytesRead;
-      }
-    }
-
-    return size;
-  }
 }