[incubator-wayang] 06/07: [WAYANG-28] add EmptyInjector implementation

[be...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

bertty pushed a commit to branch debugger
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git

commit d22790782f5623c2fdb29398a8c6f6aa16d9424a
Author: Bertty Contreras-Rojas <be...@scalytics.io>
AuthorDate: Tue Apr 6 11:53:17 2021 -0400

    [WAYANG-28] add EmptyInjector implementation
---
 .../hackit/core/iterator/ElementPlusIterator.java  | 36 ++++++++++++++++
 .../hackit/core/sniffer/inject/EmptyInjector.java  | 49 ++++++++++++++++++++++
 2 files changed, 85 insertions(+)

diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/iterator/ElementPlusIterator.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/iterator/ElementPlusIterator.java
new file mode 100644
index 0000000..73353bc
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/iterator/ElementPlusIterator.java
@@ -0,0 +1,36 @@
+package org.apache.wayang.plugin.hackit.core.iterator;
+
+import java.util.Iterator;
+
+public class ElementPlusIterator<T> implements Iterator<T> {
+
+    private boolean element_consumed = false;
+    private T element;
+    private Iterator<T> iterator;
+
+
+    public ElementPlusIterator(T element, Iterator<T> iterator) {
+        if(element == null){
+            throw new RuntimeException("the element can't be null");
+        }
+        this.element = element;
+        this.iterator = iterator;
+    }
+
+    @Override
+    public boolean hasNext() {
+        if( ! this.element_consumed ){
+            return true;
+        }
+        return this.iterator.hasNext();
+    }
+
+    @Override
+    public T next() {
+        if( ! this.element_consumed ){
+            this.element_consumed = true;
+            return this.element;
+        }
+        return this.iterator.next();
+    }
+}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/inject/EmptyInjector.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/inject/EmptyInjector.java
new file mode 100644
index 0000000..689882d
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/inject/EmptyInjector.java
@@ -0,0 +1,49 @@
+package org.apache.wayang.plugin.hackit.core.sniffer.inject;
+
+import org.apache.wayang.plugin.hackit.core.iterator.ElementPlusIterator;
+
+import java.util.Iterator;
+import java.util.Spliterator;
+import java.util.Spliterators;
+import java.util.stream.StreamSupport;
+
+public class EmptyInjector<T> implements Injector<T>{
+
+    @Override
+    public Iterator<T> inject(T element, Iterator<T> iterator) {
+        return
+            StreamSupport.stream(
+                Spliterators.spliteratorUnknownSize(
+                    new ElementPlusIterator<T>(
+                        element,
+                        iterator
+                    ),
+                    Spliterator.ORDERED
+                ),
+                false
+            ).map(
+                (T _element) -> {
+                    if (this.is_halt_job(_element)) {
+                        try {
+                            Thread.sleep(1000);
+                        } catch (InterruptedException e) {
+                            e.printStackTrace();
+                        }
+                    }
+                    return _element;
+                }
+            ).filter(
+                (T _element) -> ! this.is_skip_element(_element)
+            ).iterator();
+    }
+
+    @Override
+    public boolean is_skip_element(T element) {
+        return false;
+    }
+
+    @Override
+    public boolean is_halt_job(T element) {
+        return false;
+    }
+}