More operators, improve coverage

Author: akarnokd

README.md

@@ -1,6 +1,10 @@
 ixjava
 =================
 
+<a href='https://travis-ci.org/akarnokd/ixjava/builds'><img src='https://travis-ci.org/akarnokd/ixjava.svg?branch=1.x'></a>
+[![codecov.io](http://codecov.io/github/akarnokd/ixjava/coverage.svg?branch=1.x)](http://codecov.io/github/akarnokd/ixjava?branch=1.x)
+[![Maven Central](https://maven-badges.herokuapp.com/maven-central/com.github.akarnokd/ixjava/badge.svg)](https://maven-badges.herokuapp.com/maven-central/com.github.akarnokd/ixjava)
+
 Interactive Extensions for Java, the dual of RxJava. Originally implemented in the Reactive4Java framework, now converted to work with RxJava.
 
 The aim is to provide pull-based datastream support with the same naming as in RxJava mainly for the pre-Java-8 world. The Stream API in Java 8 is not exactly the same thing because Streams can be only consumed once while Iterables can be consumed many times. Google Guava features a lot of Iterable operators but without method chaining support.
@@ -10,11 +14,6 @@ and interactive dataflows.**
 
 # Releases
 
-<a href='https://travis-ci.org/akarnokd/ixjava/builds'><img src='https://travis-ci.org/akarnokd/ixjava.svg?branch=1.x'></a>
-[![codecov.io](http://codecov.io/github/akarnokd/ixjava/coverage.svg?branch=1.x)](http://codecov.io/github/akarnokd/ixjava?branch=1.x)
-[![Maven Central](https://maven-badges.herokuapp.com/maven-central/com.github.akarnokd/ixjava/badge.svg)](https://maven-badges.herokuapp.com/maven-central/com.github.akarnokd/ixjava)
-
-
 **gradle**
 
 ```

src/main/java/ix/Ix.java

@@ -170,9 +170,13 @@ public static <T> Ix<T> repeatValue(T value, long count) {
     }
 
     public static <T> Ix<T> repeatValue(T value, Pred0 stopPredicate) {
-        return new IxRepeatPredicate<T>(value, stopPredicate);
+        return repeatValue(value, Long.MAX_VALUE, stopPredicate);
     }
-    
+
+    public static <T> Ix<T> repeatValue(T value, long count, Pred0 stopPredicate) {
+        return new IxRepeatPredicate<T>(value, count, stopPredicate);
+    }
+
     public static <T, R> Ix<R> forloop(T seed, Pred<? super T> condition, 
             Func1<? super T, ? extends T> next,
             Func1<? super T, ? extends R> selector) {
@@ -395,18 +399,21 @@ public final Ix<T> takeUntil(Pred<? super T> stopPredicate) {
     }
 
     public final Ix<List<T>> buffer(int size) {
-        // TODO implement
-        throw new UnsupportedOperationException();
+        return new IxBuffer<T>(this, size);
     }
 
     public final Ix<List<T>> buffer(int size, int skip) {
-        // TODO implement
-        throw new UnsupportedOperationException();
+        if (size == skip) {
+            return buffer(size);
+        }
+        if (size < skip) {
+            return new IxBufferSkip<T>(this, size, skip);
+        }
+        return new IxBufferOverlap<T>(this, size, skip);
     }
 
     public final <K> Ix<GroupedIx<K, T>> groupBy(Func1<? super T, ? extends K> keySelector) {
-        // TODO implement
-        throw new UnsupportedOperationException();
+        return groupBy(keySelector, IdentityHelper.<T>instance());
     }
 
     public final <K, V> Ix<GroupedIx<K, V>> groupBy(Func1<? super T, ? extends K> keySelector,
@@ -416,23 +423,19 @@ public final <K, V> Ix<GroupedIx<K, V>> groupBy(Func1<? super T, ? extends K> ke
     }
 
     public final Ix<T> repeat() {
-        // TODO implement
-        throw new UnsupportedOperationException();
+        return concat(repeatValue(this));
     }
 
     public final Ix<T> repeat(long times) {
-        // TODO implement
-        throw new UnsupportedOperationException();
+        return concat(repeatValue(this, times));
     }
 
     public final Ix<T> repeat(Pred0 predicate) {
-        // TODO implement
-        throw new UnsupportedOperationException();
+        return concat(repeatValue(this, predicate));
     }
 
     public final Ix<T> repeat(long times, Pred0 predicate) {
-        // TODO implement
-        throw new UnsupportedOperationException();
+        return concat(repeatValue(this, times, predicate));
     }
 
     public final Ix<T> publish() {
@@ -659,6 +662,9 @@ public final void print(CharSequence separator, int charsPerLine) {
                     System.out.println();;
                     System.out.print(s);
                     len = s.length();
+                } else {
+                    System.out.print(s);
+                    len += s.length();
                 }
             }
 

src/main/java/ix/IxBuffer.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2011-2016 David Karnok
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ix;
+
+import java.util.*;
+
+final class IxBuffer<T> extends IxSource<T, List<T>> {
+
+    final int size;
+    
+    public IxBuffer(Iterable<T> source, int size) {
+        super(source);
+        this.size = size;
+    }
+    
+    @Override
+    public Iterator<List<T>> iterator() {
+        return new BufferIterator<T>(source.iterator(), size);
+    }
+
+    static final class BufferIterator<T> extends IxSourceIterator<T, List<T>> {
+        final int size;
+
+        public BufferIterator(Iterator<T> it, int size) {
+            super(it);
+            this.size = size;
+        }
+
+        @Override
+        protected boolean moveNext() {
+            int s = size;
+            
+            Iterator<T> it = this.it;
+            
+            List<T> list = new ArrayList<T>();
+            
+            while (s != 0 && it.hasNext()) {
+                list.add(it.next());
+                s--;
+            }
+            
+            if (list.isEmpty()) {
+                done = true;
+                return false;
+            }
+            value = list;
+            hasValue = true;
+            if (s != 0) {
+                done = true;
+            }
+            return true;
+        }
+        
+        
+    }
+}

src/main/java/ix/IxBufferOverlap.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2011-2016 David Karnok
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ix;
+
+import java.util.*;
+
+import rx.functions.Action2;
+
+final class IxBufferOverlap<T> extends IxSource<T, List<T>> {
+
+    final int size;
+    
+    final int skip;
+    
+    public IxBufferOverlap(Iterable<T> source, int size, int skip) {
+        super(source);
+        this.size = size;
+        this.skip = skip;
+    }
+    
+    @Override
+    public Iterator<List<T>> iterator() {
+        return new BufferIterator<T>(source.iterator(), size, skip);
+    }
+
+    static final class BufferIterator<T> extends IxSourceQueuedIterator<T, List<T>, List<T>>
+    implements Action2<List<T>, T> {
+        final int size;
+        
+        final int skip;
+        
+        int index;
+        
+        public BufferIterator(Iterator<T> it, int size, int skip) {
+            super(it);
+            this.size = size;
+            this.skip = skip;
+        }
+
+        @SuppressWarnings("unchecked")
+        @Override
+        protected boolean moveNext() {
+            Iterator<T> it = this.it;
+            
+            int s = size;
+            int k = skip;
+            
+            int i = index;
+            
+            while (it.hasNext()) {
+                if (i == 0) {
+                    offer(new ArrayList<T>());
+                }
+                T v = it.next();
+                foreach(this, v);
+                if (++i == k) {
+                    i = 0;
+                }
+                
+                if (((List<T>)peek()).size() == s) {
+                    break;
+                }
+            }
+            index = i;
+            
+            List<T> list = fromObject(poll());
+            if (list == null) {
+                done = true;
+                return false;
+            }
+            
+            value = list;
+            hasValue = true;
+            return true;
+        }
+        
+        @Override
+        public void call(List<T> t1, T t2) {
+            t1.add(t2);
+        }
+    }
+}

src/main/java/ix/IxBufferSkip.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright 2011-2016 David Karnok
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ix;
+
+import java.util.*;
+
+final class IxBufferSkip<T> extends IxSource<T, List<T>> {
+
+    final int size;
+    
+    final int skip;
+    
+    public IxBufferSkip(Iterable<T> source, int size, int skip) {
+        super(source);
+        this.size = size;
+        this.skip = skip;
+    }
+    
+    @Override
+    public Iterator<List<T>> iterator() {
+        return new BufferSkipIterator<T>(source.iterator(), size, skip);
+    }
+
+    static final class BufferSkipIterator<T> extends IxSourceIterator<T, List<T>> {
+        final int size;
+        
+        final int skip;
+        
+        boolean once;
+
+        public BufferSkipIterator(Iterator<T> it, int size, int skip) {
+            super(it);
+            this.size = size;
+            this.skip = skip;
+        }
+
+        @Override
+        protected boolean moveNext() {
+            Iterator<T> it = this.it;
+            
+            int s = size;
+            
+            if (once) {
+                int k = skip - s;
+                while (k != 0 && it.hasNext()) {
+                    it.next();
+                    k--;
+                }
+                
+            } else {
+                once = true;
+            }
+
+            List<T> list = new ArrayList<T>();
+            
+            while (s != 0 && it.hasNext()) {
+                list.add(it.next());
+                s--;
+            }
+            
+            
+            if (list.isEmpty()) {
+                done = true;
+                return false;
+            }
+            value = list;
+            hasValue = true;
+            if (s != 0) {
+                done = true;
+            }
+            return true;
+        }
+        
+        
+    }
+}

src/main/java/ix/IxDoOn.java

@@ -1,3 +1,19 @@
+/*
+ * Copyright 2011-2016 David Karnok
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package ix;
 
 import java.util.Iterator;