Add ByteSourcePayloadIterator to BasePayloadSlicer

ByteSourcePayloadIterator avoids buffering the entire slice like
InputStreamPayloadIterator does.  Also rename PayloadIterator to
better reflect its intent and reduce its visibility.

core/src/main/java/org/jclouds/io/internal/BasePayloadSlicer.java

@@ -19,10 +19,7 @@ package org.jclouds.io.internal;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 
-import java.io.ByteArrayInputStream;
 import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Arrays;
@@ -36,6 +33,7 @@ import org.jclouds.io.Payload;
 import org.jclouds.io.Payloads;
 import org.jclouds.io.PayloadSlicer;
 import org.jclouds.io.payloads.BaseMutableContentMetadata;
+import org.jclouds.io.payloads.ByteSourcePayload;
 
 import com.google.common.base.Charsets;
 import com.google.common.base.Throwables;
@@ -47,16 +45,16 @@ import com.google.common.io.Files;
 @Singleton
 public class BasePayloadSlicer implements PayloadSlicer {
 
-   public static class PayloadIterator implements Iterable<Payload>, Iterator<Payload> {
+   private static class InputStreamPayloadIterator implements Iterable<Payload>, Iterator<Payload> {
 
       private final InputStream input;
       private final ContentMetadata metaData;
       private Payload nextPayload;
       private final int readLen;
 
-      public PayloadIterator(InputStream input, ContentMetadata meta) {
+      InputStreamPayloadIterator(InputStream input, ContentMetadata metaData) {
          this.input = checkNotNull(input, "input");
-         this.metaData = checkNotNull(meta, "meta");
+         this.metaData = checkNotNull(metaData, "metaData");
          this.readLen = checkNotNull(this.metaData.getContentLength(), "content-length").intValue();
 
          this.nextPayload = getNextPayload();
@@ -82,7 +80,7 @@ public class BasePayloadSlicer implements PayloadSlicer {
 
       @Override
       public void remove() {
-         throw new UnsupportedOperationException();
+         throw new UnsupportedOperationException("Payload iterator does not support removal");
       }
 
       @Override
@@ -127,6 +125,71 @@ public class BasePayloadSlicer implements PayloadSlicer {
 
    }
 
+   private static class ByteSourcePayloadIterator implements Iterable<Payload>, Iterator<Payload> {
+      private final ByteSource input;
+      private final ContentMetadata metaData;
+      private Payload nextPayload;
+      private long offset = 0;
+      private final long readLen;
+
+      ByteSourcePayloadIterator(ByteSource input, ContentMetadata metaData) {
+         this.input = checkNotNull(input, "input");
+         this.metaData = checkNotNull(metaData, "metaData");
+         this.readLen = checkNotNull(this.metaData.getContentLength(), "content-length").longValue();
+         this.nextPayload = getNextPayload();
+      }
+
+      @Override
+      public boolean hasNext() {
+         return nextPayload != null;
+      }
+
+      @Override
+      public Payload next() {
+         if (!hasNext()) {
+            throw new NoSuchElementException();
+         }
+
+         Payload payload = nextPayload;
+         nextPayload = getNextPayload();
+
+         return payload;
+      }
+
+      @Override
+      public void remove() {
+         throw new UnsupportedOperationException("Payload iterator does not support removal");
+      }
+
+      @Override
+      public Iterator<Payload> iterator() {
+         return this;
+      }
+
+      private Payload getNextPayload() {
+         ByteSource byteSource;
+         long byteSourceSize;
+         try {
+            if (offset >= input.size()) {
+               return null;
+            }
+            byteSource = input.slice(offset, readLen);
+            byteSourceSize = byteSource.size();
+         } catch (IOException e) {
+            throw Throwables.propagate(e);
+         }
+
+         Payload nextPayload = new ByteSourcePayload(byteSource);
+         ContentMetadata cm = metaData.toBuilder()
+               .contentLength(byteSourceSize)
+               .contentMD5((HashCode) null)
+               .build();
+         nextPayload.setContentMetadata(BaseMutableContentMetadata.fromContentMetadata(cm));
+         offset += byteSourceSize;
+         return nextPayload;
+      }
+   }
+
    /**
     * {@inheritDoc}
     */
@@ -209,6 +272,8 @@ public class BasePayloadSlicer implements PayloadSlicer {
          return doSlice((byte[]) rawContent, meta);
       } else if (rawContent instanceof InputStream) {
          return doSlice((InputStream) rawContent, meta);
+      } else if (rawContent instanceof ByteSource) {
+         return doSlice((ByteSource) rawContent, meta);
       } else {
          return doSlice(input, meta);
       }
@@ -220,23 +285,22 @@ public class BasePayloadSlicer implements PayloadSlicer {
    }
 
    protected Iterable<Payload> doSlice(String rawContent, ContentMetadata meta) {
-      return doSlice(rawContent.getBytes(Charsets.UTF_8), meta);
+      return doSlice(ByteSource.wrap(rawContent.getBytes(Charsets.UTF_8)), meta);
    }
 
    protected Iterable<Payload> doSlice(byte[] rawContent, ContentMetadata meta) {
-      return doSlice(new ByteArrayInputStream(rawContent), meta);
+      return doSlice(ByteSource.wrap(rawContent), meta);
    }
 
    protected Iterable<Payload> doSlice(File rawContent, ContentMetadata meta) {
-      try {
+      return doSlice(Files.asByteSource(rawContent), meta);
-         return doSlice(new FileInputStream(rawContent), meta);
-      } catch (FileNotFoundException e) {
-         throw Throwables.propagate(e);
-      }
    }
 
    protected Iterable<Payload> doSlice(InputStream rawContent, ContentMetadata meta) {
-      return new PayloadIterator(rawContent, meta);
+      return new InputStreamPayloadIterator(rawContent, meta);
    }
 
+   protected Iterable<Payload> doSlice(ByteSource rawContent, ContentMetadata meta) {
+      return new ByteSourcePayloadIterator(rawContent, meta);
+   }
 }

core/src/test/java/org/jclouds/io/internal/BasePayloadSlicerTest.java

@@ -24,7 +24,6 @@ import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.util.Iterator;
 
-import org.jclouds.io.ByteSources;
 import org.jclouds.io.Payload;
 import org.jclouds.io.PayloadSlicer;
 import org.jclouds.io.payloads.ByteSourcePayload;
@@ -33,7 +32,6 @@ import org.jclouds.util.Strings2;
 import org.testng.annotations.Test;
 
 import com.google.common.base.Charsets;
-import com.google.common.collect.Iterables;
 import com.google.common.io.ByteSource;
 
 @Test
@@ -71,14 +69,46 @@ public class BasePayloadSlicerTest {
    }
 
    @Test
-   public void testIterableSliceWithRepeatingByteSource() throws IOException {
+   public void testIterableSliceWithRepeatingByteSourceSmallerPartSize() throws IOException {
-      String content = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz\n"; /* 53 chars */
-      byte[] contentBytes = content.getBytes(Charsets.UTF_8);
-      ByteSource byteSource = ByteSources.repeatingArrayByteSource(contentBytes).slice(0, 1024);
       PayloadSlicer slicer = new BasePayloadSlicer();
+      ByteSource byteSource = ByteSource.wrap("aaaaaaaaaabbbbbbbbbbccccc".getBytes(Charsets.UTF_8)); /* 25 chars */
       Payload payload = new ByteSourcePayload(byteSource);
 
-      assertEquals(Iterables.size(slicer.slice(payload, 100)), 11);
+      Iterator<Payload> iter = slicer.slice(payload, 10).iterator();
-      assertEquals(Iterables.size(slicer.slice(payload, 53)), 20);
+      Payload part;
+
+      assertTrue(iter.hasNext(), "Not enough results");
+      part = iter.next();
+      assertEquals(Strings2.toStringAndClose(part.getInput()), "aaaaaaaaaa");
+      assertEquals(part.getContentMetadata().getContentLength(), Long.valueOf(10));
+
+      assertTrue(iter.hasNext(), "Not enough results");
+      part = iter.next();
+      assertEquals(Strings2.toStringAndClose(part.getInput()), "bbbbbbbbbb");
+      assertEquals(part.getContentMetadata().getContentLength(), Long.valueOf(10));
+
+      assertTrue(iter.hasNext(), "Not enough results");
+      part = iter.next();
+      assertEquals(Strings2.toStringAndClose(part.getInput()), "ccccc");
+      assertEquals(part.getContentMetadata().getContentLength(), Long.valueOf(5));
+
+      assertFalse(iter.hasNext());
+   }
+
+   @Test
+   public void testIterableSliceWithRepeatingByteSourceLargerPartSize() throws IOException {
+      PayloadSlicer slicer = new BasePayloadSlicer();
+      ByteSource byteSource = ByteSource.wrap("aaaaaaaaaabbbbbbbbbbccccc".getBytes(Charsets.UTF_8)); /* 25 chars */
+      Payload payload = new ByteSourcePayload(byteSource);
+
+      Iterator<Payload> iter = slicer.slice(payload, 50).iterator();
+      Payload part;
+
+      assertTrue(iter.hasNext(), "Not enough results");
+      part = iter.next();
+      assertEquals(Strings2.toStringAndClose(part.getInput()), "aaaaaaaaaabbbbbbbbbbccccc");
+      assertEquals(part.getContentMetadata().getContentLength(), Long.valueOf(25));
+
+      assertFalse(iter.hasNext());
    }
 }