You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by ok...@apache.org on 2019/03/13 20:51:43 UTC
[tinkerpop] 20/41: got FlatMapFunctions working. Added __.union().
Added all the Iterator utilities from TP3.
This is an automated email from the ASF dual-hosted git repository.
okram pushed a commit to branch tp4
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 193f012802014532c18860ac4e7483ab5e54244e
Author: Marko A. Rodriguez <ok...@gmail.com>
AuthorDate: Mon Mar 11 12:54:19 2019 -0600
got FlatMapFunctions working. Added __.union(). Added all the Iterator utilities from TP3.
---
.../org/apache/tinkerpop/language/Symbols.java | 1 +
.../org/apache/tinkerpop/language/Traversal.java | 12 +-
.../java/org/apache/tinkerpop/language/__.java | 21 +-
.../tinkerpop/machine/bytecode/BytecodeUtil.java | 23 +-
.../machine/functions/FlatMapFunction.java | 2 +-
.../machine/functions/NestedFunction.java | 7 +-
.../machine/functions/flatMap/UnionFlatMap.java | 85 ++++
.../tinkerpop/machine/functions/map/MapMap.java | 25 +-
.../machine/processor/EmptyProcessorFactory.java | 7 +-
.../machine/processor/ProcessorFactory.java | 14 +-
.../machine/traversers/CompleteTraverser.java | 12 +
.../tinkerpop/machine/traversers/Traverser.java | 8 +-
.../ArrayIterator.java} | 34 +-
.../org/apache/tinkerpop/util/DoubleIterator.java | 65 ++++
.../org/apache/tinkerpop/util/IteratorUtils.java | 433 +++++++++++++++++++++
.../org/apache/tinkerpop/util/MultiIterator.java | 104 +++++
.../SingleIterator.java} | 33 +-
.../org/apache/tinkerpop/machine/beam/Beam.java | 12 +-
.../tinkerpop/machine/beam/BeamProcessor.java | 8 +-
.../apache/tinkerpop/machine/beam/FlatMapFn.java} | 22 +-
.../org/apache/tinkerpop/machine/beam/MapFn.java | 16 +-
.../apache/tinkerpop/machine/beam/BeamTest.java | 7 +-
.../tinkerpop/machine/pipes/FlatMapStep.java} | 30 +-
.../org/apache/tinkerpop/machine/pipes/Pipes.java | 15 +-
.../tinkerpop/machine/pipes/PipesProcessor.java | 9 +-
.../apache/tinkerpop/machine/pipes/PipesTest.java | 6 +-
26 files changed, 920 insertions(+), 91 deletions(-)
diff --git a/java/core/src/main/java/org/apache/tinkerpop/language/Symbols.java b/java/core/src/main/java/org/apache/tinkerpop/language/Symbols.java
index 4f5ed5f..7964d07 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/language/Symbols.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/language/Symbols.java
@@ -40,4 +40,5 @@ public final class Symbols {
public static final String MAP = "map";
public static final String PATH = "path";
public static final String SUM = "sum";
+ public static final String UNION = "union";
}
diff --git a/java/core/src/main/java/org/apache/tinkerpop/language/Traversal.java b/java/core/src/main/java/org/apache/tinkerpop/language/Traversal.java
index e9ebb85..4eef5cd 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/language/Traversal.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/language/Traversal.java
@@ -22,6 +22,7 @@ import org.apache.tinkerpop.machine.bytecode.Bytecode;
import org.apache.tinkerpop.machine.bytecode.BytecodeUtil;
import org.apache.tinkerpop.machine.coefficients.Coefficient;
import org.apache.tinkerpop.machine.coefficients.LongCoefficient;
+import org.apache.tinkerpop.machine.processor.EmptyProcessorFactory;
import org.apache.tinkerpop.machine.processor.Processor;
import org.apache.tinkerpop.machine.processor.ProcessorFactory;
import org.apache.tinkerpop.machine.traversers.Path;
@@ -47,7 +48,7 @@ public class Traversal<C, S, E> implements Iterator<E> {
public Traversal(final Bytecode<C> bytecode) {
this.bytecode = bytecode;
this.currentCoefficient = BytecodeUtil.getCoefficient(this.bytecode).orElse((Coefficient<C>) LongCoefficient.create());
- this.processorFactory = BytecodeUtil.getProcessorFactory(this.bytecode).orElse(null);
+ this.processorFactory = BytecodeUtil.getProcessorFactory(this.bytecode).orElse(EmptyProcessorFactory.instance());
}
public Traversal<C, S, E> as(final String label) {
@@ -105,6 +106,15 @@ public class Traversal<C, S, E> implements Iterator<E> {
return (Traversal) this;
}
+ public <R> Traversal<C, S, R> union(final Traversal<C, E, R>... traversals) {
+ final Bytecode<C>[] bytecodes = new Bytecode[traversals.length];
+ for (int i = 0; i < traversals.length; i++) {
+ bytecodes[i] = traversals[i].bytecode;
+ }
+ this.bytecode.addInstruction(this.currentCoefficient, Symbols.UNION, bytecodes);
+ return (Traversal) this;
+ }
+
///////
diff --git a/java/core/src/main/java/org/apache/tinkerpop/language/__.java b/java/core/src/main/java/org/apache/tinkerpop/language/__.java
index 80bb253..4c2b4ba 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/language/__.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/language/__.java
@@ -19,16 +19,29 @@
package org.apache.tinkerpop.language;
import org.apache.tinkerpop.machine.bytecode.Bytecode;
-import org.apache.tinkerpop.machine.coefficients.LongCoefficient;
-import org.apache.tinkerpop.machine.processor.EmptyProcessorFactory;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
public class __ {
+ private static <C, S> Traversal<C, S, S> start() {
+ return new Traversal<>(new Bytecode<>());
+ }
+
+ public static <C, S> Traversal<C, S, S> c(final C coefficient) {
+ return __.<C, S>start().c(coefficient);
+ }
+
public static <C> Traversal<C, Long, Long> incr() {
- // this is bad -- it needs to know how to get the withCoefficient of the parent traversal
- return (Traversal) new Traversal<>(new Bytecode<>()).incr();
+ return __.<C, Long>start().incr();
+ }
+
+ public static <C, S extends Number> Traversal<C, S, S> sum() {
+ return __.<C, S>start().sum();
+ }
+
+ public static <C, S> Traversal<C, S, Long> count() {
+ return __.<C, S>start().count();
}
}
diff --git a/java/core/src/main/java/org/apache/tinkerpop/machine/bytecode/BytecodeUtil.java b/java/core/src/main/java/org/apache/tinkerpop/machine/bytecode/BytecodeUtil.java
index 61882b7..10a9b2f 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/machine/bytecode/BytecodeUtil.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/machine/bytecode/BytecodeUtil.java
@@ -23,6 +23,7 @@ import org.apache.tinkerpop.machine.coefficients.Coefficient;
import org.apache.tinkerpop.machine.functions.CFunction;
import org.apache.tinkerpop.machine.functions.filter.IdentityFilter;
import org.apache.tinkerpop.machine.functions.filter.IsFilter;
+import org.apache.tinkerpop.machine.functions.flatMap.UnionFlatMap;
import org.apache.tinkerpop.machine.functions.initial.InjectInitial;
import org.apache.tinkerpop.machine.functions.map.IncrMap;
import org.apache.tinkerpop.machine.functions.map.MapMap;
@@ -58,14 +59,6 @@ public final class BytecodeUtil {
return bytecode;
}
- public static <C> List<CFunction<C>> compile(final Bytecode<C> bytecode) {
- final List<CFunction<C>> functions = new ArrayList<>();
- for (final Instruction<C> instruction : bytecode.getInstructions()) {
- functions.add(BytecodeUtil.generateFunction(instruction));
- }
- return functions;
- }
-
public static <C> List<Strategy> getStrategies(final Bytecode<C> bytecode) {
try {
final List<Strategy> strategies = new ArrayList<>();
@@ -113,6 +106,14 @@ public final class BytecodeUtil {
return Optional.of(new CompleteTraverserFactory<C>());
}
+ public static <C> List<CFunction<C>> compile(final Bytecode<C> bytecode) {
+ final List<CFunction<C>> functions = new ArrayList<>();
+ for (final Instruction<C> instruction : bytecode.getInstructions()) {
+ functions.add(BytecodeUtil.generateFunction(instruction));
+ }
+ return functions;
+ }
+
private static <C> CFunction<C> generateFunction(final Instruction<C> instruction) {
final String op = instruction.op();
final Coefficient<C> coefficient = instruction.coefficient();
@@ -134,6 +135,12 @@ public final class BytecodeUtil {
return new PathMap<>(coefficient, labels);
case Symbols.SUM:
return new SumReduce<>(coefficient, labels);
+ case Symbols.UNION:
+ final List<List<CFunction<C>>> branchFunctions = new ArrayList<>();
+ for (final Bytecode<C> arg : (Bytecode<C>[]) instruction.args()) {
+ branchFunctions.add(compile(arg));
+ }
+ return new UnionFlatMap<>(coefficient, labels, branchFunctions);
default:
throw new RuntimeException("This is an unknown instruction:" + instruction.op());
}
diff --git a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/FlatMapFunction.java b/java/core/src/main/java/org/apache/tinkerpop/machine/functions/FlatMapFunction.java
index 3bc5d68..dea4af3 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/FlatMapFunction.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/machine/functions/FlatMapFunction.java
@@ -26,5 +26,5 @@ import java.util.function.Function;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public interface FlatMapFunction<C, S, E> extends Function<Traverser<C, S>, Iterator<Traverser<C, E>>>, CFunction<C> {
+public interface FlatMapFunction<C, S, E> extends Function<Traverser<C, S>, Iterator<E>>, CFunction<C> {
}
diff --git a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/NestedFunction.java b/java/core/src/main/java/org/apache/tinkerpop/machine/functions/NestedFunction.java
index 3ecfed6..a63f226 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/NestedFunction.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/machine/functions/NestedFunction.java
@@ -18,7 +18,8 @@
*/
package org.apache.tinkerpop.machine.functions;
-import org.apache.tinkerpop.machine.processor.Processor;
+import org.apache.tinkerpop.machine.processor.ProcessorFactory;
+import org.apache.tinkerpop.machine.traversers.TraverserFactory;
import java.util.List;
@@ -27,7 +28,7 @@ import java.util.List;
*/
public interface NestedFunction<C, S, E> extends CFunction<C> {
- public void setProcessor(final Processor<C, S, E> processor);
+ public void setProcessor(final TraverserFactory<C> traverserFactory, final ProcessorFactory processorFactory);
- public List<CFunction<C>> getFunctions();
+ public List<List<CFunction<C>>> getFunctions();
}
diff --git a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/flatMap/UnionFlatMap.java b/java/core/src/main/java/org/apache/tinkerpop/machine/functions/flatMap/UnionFlatMap.java
new file mode 100644
index 0000000..7544580
--- /dev/null
+++ b/java/core/src/main/java/org/apache/tinkerpop/machine/functions/flatMap/UnionFlatMap.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.tinkerpop.machine.functions.flatMap;
+
+import org.apache.tinkerpop.machine.coefficients.Coefficient;
+import org.apache.tinkerpop.machine.functions.AbstractFunction;
+import org.apache.tinkerpop.machine.functions.CFunction;
+import org.apache.tinkerpop.machine.functions.FlatMapFunction;
+import org.apache.tinkerpop.machine.functions.NestedFunction;
+import org.apache.tinkerpop.machine.processor.Processor;
+import org.apache.tinkerpop.machine.processor.ProcessorFactory;
+import org.apache.tinkerpop.machine.traversers.Traverser;
+import org.apache.tinkerpop.machine.traversers.TraverserFactory;
+import org.apache.tinkerpop.util.IteratorUtils;
+import org.apache.tinkerpop.util.MultiIterator;
+import org.apache.tinkerpop.util.StringFactory;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class UnionFlatMap<C, S, E> extends AbstractFunction<C, S, Iterator<E>> implements FlatMapFunction<C, S, E>, NestedFunction<C, S, E> {
+
+ private final List<List<CFunction<C>>> branchFunctions;
+ private transient List<Processor<C, S, E>> processors;
+ private TraverserFactory<C> traverserFactory;
+ private ProcessorFactory processorFactory;
+
+ public UnionFlatMap(final Coefficient<C> coefficient, final Set<String> labels, final List<List<CFunction<C>>> branchFunctions) {
+ super(coefficient, labels);
+ this.branchFunctions = branchFunctions;
+ }
+
+ @Override
+ public Iterator<E> apply(final Traverser<C, S> traverser) {
+ if (null == this.processors) {
+ this.processors = new ArrayList<>(this.branchFunctions.size());
+ for (final List<CFunction<C>> functions : this.branchFunctions) {
+ this.processors.add(processorFactory.mint(traverserFactory, functions));
+ }
+ }
+ final MultiIterator<E> iterator = new MultiIterator<>();
+ for (final Processor<C, S, E> processor : this.processors) {
+ processor.reset();
+ processor.addStart(traverser.clone());
+ iterator.addIterator(IteratorUtils.map(processor, Traverser::object));
+ }
+
+ return iterator;
+ }
+
+ public void setProcessor(final TraverserFactory<C> traverserFactory, final ProcessorFactory processorFactory) {
+ this.traverserFactory = traverserFactory;
+ this.processorFactory = processorFactory;
+ }
+
+ public List<List<CFunction<C>>> getFunctions() {
+ return this.branchFunctions;
+ }
+
+ @Override
+ public String toString() {
+ return StringFactory.makeFunctionString(this, this.branchFunctions.toArray());
+ }
+}
diff --git a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/map/MapMap.java b/java/core/src/main/java/org/apache/tinkerpop/machine/functions/map/MapMap.java
index 8237fbf..08d4961 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/map/MapMap.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/machine/functions/map/MapMap.java
@@ -24,10 +24,12 @@ import org.apache.tinkerpop.machine.functions.CFunction;
import org.apache.tinkerpop.machine.functions.MapFunction;
import org.apache.tinkerpop.machine.functions.NestedFunction;
import org.apache.tinkerpop.machine.processor.Processor;
-import org.apache.tinkerpop.machine.traversers.CompleteTraverser;
+import org.apache.tinkerpop.machine.processor.ProcessorFactory;
import org.apache.tinkerpop.machine.traversers.Traverser;
+import org.apache.tinkerpop.machine.traversers.TraverserFactory;
import org.apache.tinkerpop.util.StringFactory;
+import java.util.Collections;
import java.util.List;
import java.util.Set;
@@ -37,7 +39,10 @@ import java.util.Set;
public class MapMap<C, S, E> extends AbstractFunction<C, S, E> implements MapFunction<C, S, E>, NestedFunction<C, S, E> {
private final List<CFunction<C>> mapFunctions;
- private Processor<C, S, E> processor;
+ private TraverserFactory<C> traverserFactory;
+ private ProcessorFactory processorFactory;
+
+ private transient Processor<C, S, E> processor;
public MapMap(final Coefficient<C> coefficient, final Set<String> labels, final List<CFunction<C>> mapFunctions) {
super(coefficient, labels);
@@ -46,17 +51,23 @@ public class MapMap<C, S, E> extends AbstractFunction<C, S, E> implements MapFun
@Override
public E apply(final Traverser<C, S> traverser) {
- this.processor.reset();
+ if (null == this.processor)
+ this.processor = processorFactory.mint(traverserFactory, this.mapFunctions);
+ else
+ this.processor.reset();
this.processor.addStart(traverser);
return this.processor.next().object();
}
- public void setProcessor(final Processor<C, S, E> processor) {
- this.processor = processor;
+ @Override
+ public void setProcessor(final TraverserFactory<C> traverserFactory, final ProcessorFactory processorFactory) {
+ this.traverserFactory = traverserFactory;
+ this.processorFactory = processorFactory;
}
- public List<CFunction<C>> getFunctions() {
- return this.mapFunctions;
+ @Override
+ public List<List<CFunction<C>>> getFunctions() {
+ return Collections.singletonList(this.mapFunctions);
}
@Override
diff --git a/java/core/src/main/java/org/apache/tinkerpop/machine/processor/EmptyProcessorFactory.java b/java/core/src/main/java/org/apache/tinkerpop/machine/processor/EmptyProcessorFactory.java
index 89233df..5623322 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/machine/processor/EmptyProcessorFactory.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/machine/processor/EmptyProcessorFactory.java
@@ -18,7 +18,10 @@
*/
package org.apache.tinkerpop.machine.processor;
-import org.apache.tinkerpop.machine.bytecode.Bytecode;
+import org.apache.tinkerpop.machine.functions.CFunction;
+import org.apache.tinkerpop.machine.traversers.TraverserFactory;
+
+import java.util.List;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -32,7 +35,7 @@ public class EmptyProcessorFactory implements ProcessorFactory {
}
@Override
- public <C, S, E> Processor<C, S, E> mint(final Bytecode<C> bytecode) {
+ public <C, S, E> Processor<C, S, E> mint(final TraverserFactory<C> traverserFactory, final List<CFunction<C>> functions) {
return EmptyProcessor.instance();
}
diff --git a/java/core/src/main/java/org/apache/tinkerpop/machine/processor/ProcessorFactory.java b/java/core/src/main/java/org/apache/tinkerpop/machine/processor/ProcessorFactory.java
index 9eeed95..c6e2d11 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/machine/processor/ProcessorFactory.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/machine/processor/ProcessorFactory.java
@@ -19,11 +19,21 @@
package org.apache.tinkerpop.machine.processor;
import org.apache.tinkerpop.machine.bytecode.Bytecode;
+import org.apache.tinkerpop.machine.bytecode.BytecodeUtil;
+import org.apache.tinkerpop.machine.functions.CFunction;
+import org.apache.tinkerpop.machine.traversers.TraverserFactory;
+
+import java.io.Serializable;
+import java.util.List;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public interface ProcessorFactory {
+public interface ProcessorFactory extends Serializable {
+
+ public default <C, S, E> Processor<C, S, E> mint(final Bytecode<C> bytecode) {
+ return this.mint(BytecodeUtil.getTraverserFactory(bytecode).get(), BytecodeUtil.compile(BytecodeUtil.strategize(bytecode)));
+ }
- public <C, S, E> Processor<C, S, E> mint(final Bytecode<C> bytecode);
+ public <C, S, E> Processor<C, S, E> mint(final TraverserFactory<C> traverserFactory, final List<CFunction<C>> functions);
}
diff --git a/java/core/src/main/java/org/apache/tinkerpop/machine/traversers/CompleteTraverser.java b/java/core/src/main/java/org/apache/tinkerpop/machine/traversers/CompleteTraverser.java
index d33821b..b6c471e 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/machine/traversers/CompleteTraverser.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/machine/traversers/CompleteTraverser.java
@@ -79,5 +79,17 @@ public class CompleteTraverser<C, S> implements Traverser<C, S> {
return this.object.toString();
}
+ @Override
+ public CompleteTraverser<C, S> clone() {
+ try {
+ final CompleteTraverser<C, S> clone = (CompleteTraverser<C, S>) super.clone();
+ clone.object = this.object;
+ clone.coefficient = this.coefficient.clone();
+ clone.path = new Path(this.path);
+ return clone;
+ } catch (final CloneNotSupportedException e) {
+ throw new RuntimeException(e.getMessage(), e);
+ }
+ }
}
diff --git a/java/core/src/main/java/org/apache/tinkerpop/machine/traversers/Traverser.java b/java/core/src/main/java/org/apache/tinkerpop/machine/traversers/Traverser.java
index 8a3a320..4164fe6 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/machine/traversers/Traverser.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/machine/traversers/Traverser.java
@@ -24,16 +24,16 @@ import org.apache.tinkerpop.machine.functions.FilterFunction;
import org.apache.tinkerpop.machine.functions.FlatMapFunction;
import org.apache.tinkerpop.machine.functions.MapFunction;
import org.apache.tinkerpop.machine.functions.ReduceFunction;
+import org.apache.tinkerpop.util.IteratorUtils;
import java.io.Serializable;
-import java.util.Collections;
import java.util.Iterator;
import java.util.Set;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public interface Traverser<C, S> extends Serializable {
+public interface Traverser<C, S> extends Serializable, Cloneable {
public Coefficient<C> coefficient();
@@ -63,7 +63,7 @@ public interface Traverser<C, S> extends Serializable {
}
public default <E> Iterator<Traverser<C, E>> flatMap(final FlatMapFunction<C, S, E> function) {
- return Collections.emptyIterator();
+ return IteratorUtils.map(function.apply(this), e -> this.split(function, e));
}
//public default void sideEffect(final SideEffectFunction<C,S> function);
@@ -73,4 +73,6 @@ public interface Traverser<C, S> extends Serializable {
}
public <E> Traverser<C, E> split(final CFunction<C> function, final E object);
+
+ public Traverser<C, S> clone();
}
diff --git a/java/core/src/main/java/org/apache/tinkerpop/machine/processor/EmptyProcessorFactory.java b/java/core/src/main/java/org/apache/tinkerpop/util/ArrayIterator.java
similarity index 55%
copy from java/core/src/main/java/org/apache/tinkerpop/machine/processor/EmptyProcessorFactory.java
copy to java/core/src/main/java/org/apache/tinkerpop/util/ArrayIterator.java
index 89233df..ac3ad2b 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/machine/processor/EmptyProcessorFactory.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/util/ArrayIterator.java
@@ -16,27 +16,41 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.tinkerpop.machine.processor;
+package org.apache.tinkerpop.util;
-import org.apache.tinkerpop.machine.bytecode.Bytecode;
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Iterator;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public class EmptyProcessorFactory implements ProcessorFactory {
+public final class ArrayIterator<T> implements Iterator<T>, Serializable {
- private static final EmptyProcessorFactory INSTANCE = new EmptyProcessorFactory();
+ private final T[] array;
+ private int current = 0;
- private EmptyProcessorFactory() {
+ public ArrayIterator(final T[] array) {
+ this.array = array;
+ }
+ @Override
+ public boolean hasNext() {
+ return this.current < this.array.length;
}
@Override
- public <C, S, E> Processor<C, S, E> mint(final Bytecode<C> bytecode) {
- return EmptyProcessor.instance();
+ public T next() {
+ if (this.hasNext()) {
+ this.current++;
+ return this.array[this.current - 1];
+ } else {
+ throw FastNoSuchElementException.instance();
+ }
}
- public static EmptyProcessorFactory instance() {
- return INSTANCE;
+ @Override
+ public String toString() {
+ return Arrays.asList(array).toString();
}
-}
+}
\ No newline at end of file
diff --git a/java/core/src/main/java/org/apache/tinkerpop/util/DoubleIterator.java b/java/core/src/main/java/org/apache/tinkerpop/util/DoubleIterator.java
new file mode 100644
index 0000000..56d1d6c
--- /dev/null
+++ b/java/core/src/main/java/org/apache/tinkerpop/util/DoubleIterator.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.tinkerpop.util;
+
+import java.io.Serializable;
+import java.util.Iterator;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+final class DoubleIterator<T> implements Iterator<T>, Serializable {
+
+ private T a;
+ private T b;
+ private char current = 'a';
+
+ protected DoubleIterator(final T a, final T b) {
+ this.a = a;
+ this.b = b;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return this.current != 'x';
+ }
+
+ @Override
+ public void remove() {
+ if (this.current == 'b')
+ this.a = null;
+ else if (this.current == 'x')
+ this.b = null;
+ }
+
+ @Override
+ public T next() {
+ if (this.current == 'x')
+ throw FastNoSuchElementException.instance();
+ else {
+ if (this.current == 'a') {
+ this.current = 'b';
+ return this.a;
+ } else {
+ this.current = 'x';
+ return this.b;
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/java/core/src/main/java/org/apache/tinkerpop/util/IteratorUtils.java b/java/core/src/main/java/org/apache/tinkerpop/util/IteratorUtils.java
new file mode 100644
index 0000000..09ddb9d
--- /dev/null
+++ b/java/core/src/main/java/org/apache/tinkerpop/util/IteratorUtils.java
@@ -0,0 +1,433 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.tinkerpop.util;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.Spliterator;
+import java.util.Spliterators;
+import java.util.function.BiFunction;
+import java.util.function.BinaryOperator;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class IteratorUtils {
+
+ private IteratorUtils() {
+ }
+
+ public static final <S> Iterator<S> of(final S a) {
+ return new SingleIterator<>(a);
+ }
+
+ public static final <S> Iterator<S> of(final S a, S b) {
+ return new DoubleIterator<>(a, b);
+ }
+
+ ///////////////
+
+ public static final <S extends Collection<T>, T> S fill(final Iterator<T> iterator, final S collection) {
+ while (iterator.hasNext()) {
+ collection.add(iterator.next());
+ }
+ return collection;
+ }
+
+ public static void iterate(final Iterator iterator) {
+ while (iterator.hasNext()) {
+ iterator.next();
+ }
+ }
+
+ public static final long count(final Iterator iterator) {
+ long ix = 0;
+ for (; iterator.hasNext(); ++ix) iterator.next();
+ return ix;
+ }
+
+ public static final long count(final Iterable iterable) {
+ return IteratorUtils.count(iterable.iterator());
+ }
+
+ public static <S> List<S> list(final Iterator<S> iterator) {
+ return fill(iterator, new ArrayList<>());
+ }
+
+ public static <S> List<S> list(final Iterator<S> iterator, final Comparator comparator) {
+ final List<S> l = list(iterator);
+ Collections.sort(l, comparator);
+ return l;
+ }
+
+ public static <S> Set<S> set(final Iterator<S> iterator) {
+ return fill(iterator, new HashSet<>());
+ }
+
+ public static <S> Iterator<S> limit(final Iterator<S> iterator, final int limit) {
+ return new Iterator<S>() {
+ private int count = 0;
+
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext() && this.count < limit;
+ }
+
+ @Override
+ public void remove() {
+ iterator.remove();
+ }
+
+ @Override
+ public S next() {
+ if (this.count++ >= limit)
+ throw FastNoSuchElementException.instance();
+ return iterator.next();
+ }
+ };
+ }
+
+ ///////////////////
+
+ public static <T> boolean allMatch(final Iterator<T> iterator, final Predicate<T> predicate) {
+ while (iterator.hasNext()) {
+ if (!predicate.test(iterator.next())) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ public static <T> boolean anyMatch(final Iterator<T> iterator, final Predicate<T> predicate) {
+ while (iterator.hasNext()) {
+ if (predicate.test(iterator.next())) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ public static <T> boolean noneMatch(final Iterator<T> iterator, final Predicate<T> predicate) {
+ while (iterator.hasNext()) {
+ if (predicate.test(iterator.next())) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ ///////////////////
+
+ public static <K, S> Map<K, S> collectMap(final Iterator<S> iterator, final Function<S, K> key) {
+ return collectMap(iterator, key, Function.identity());
+ }
+
+ public static <K, S, V> Map<K, V> collectMap(final Iterator<S> iterator, final Function<S, K> key, final Function<S, V> value) {
+ final Map<K, V> map = new HashMap<>();
+ while (iterator.hasNext()) {
+ final S obj = iterator.next();
+ map.put(key.apply(obj), value.apply(obj));
+ }
+ return map;
+ }
+
+ public static <K, S> Map<K, List<S>> groupBy(final Iterator<S> iterator, final Function<S, K> groupBy) {
+ final Map<K, List<S>> map = new HashMap<>();
+ while (iterator.hasNext()) {
+ final S obj = iterator.next();
+ map.computeIfAbsent(groupBy.apply(obj), k -> new ArrayList<>()).add(obj);
+ }
+ return map;
+ }
+
+ public static <S> S reduce(final Iterator<S> iterator, final S identity, final BinaryOperator<S> accumulator) {
+ S result = identity;
+ while (iterator.hasNext()) {
+ result = accumulator.apply(result, iterator.next());
+ }
+
+ return result;
+ }
+
+ public static <S> S reduce(final Iterable<S> iterable, final S identity, final BinaryOperator<S> accumulator) {
+ return reduce(iterable.iterator(), identity, accumulator);
+ }
+
+ public static <S, E> E reduce(final Iterator<S> iterator, final E identity, final BiFunction<E, S, E> accumulator) {
+ E result = identity;
+ while (iterator.hasNext()) {
+ result = accumulator.apply(result, iterator.next());
+ }
+
+ return result;
+ }
+
+ public static <S, E> E reduce(final Iterable<S> iterable, final E identity, final BiFunction<E, S, E> accumulator) {
+ return reduce(iterable.iterator(), identity, accumulator);
+ }
+
+ ///////////////
+
+ public static final <S> Iterator<S> consume(final Iterator<S> iterator, final Consumer<S> consumer) {
+ return new Iterator<S>() {
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public void remove() {
+ iterator.remove();
+ }
+
+ @Override
+ public S next() {
+ final S s = iterator.next();
+ consumer.accept(s);
+ return s;
+ }
+ };
+ }
+
+ public static final <S> Iterable<S> consume(final Iterable<S> iterable, final Consumer<S> consumer) {
+ return () -> IteratorUtils.consume(iterable.iterator(), consumer);
+ }
+
+
+ ///////////////
+
+ public static final <S, E> Iterator<E> map(final Iterator<S> iterator, final Function<S, E> function) {
+ return new Iterator<E>() {
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public void remove() {
+ iterator.remove();
+ }
+
+ @Override
+ public E next() {
+ return function.apply(iterator.next());
+ }
+ };
+ }
+
+ public static final <S, E> Iterable<E> map(final Iterable<S> iterable, final Function<S, E> function) {
+ return () -> IteratorUtils.map(iterable.iterator(), function);
+ }
+
+ ///////////////
+
+ public static final <S> Iterator<S> filter(final Iterator<S> iterator, final Predicate<S> predicate) {
+
+
+ return new Iterator<S>() {
+ S nextResult = null;
+
+ @Override
+ public boolean hasNext() {
+ if (null != this.nextResult) {
+ return true;
+ } else {
+ advance();
+ return null != this.nextResult;
+ }
+ }
+
+ @Override
+ public void remove() {
+ iterator.remove();
+ }
+
+ @Override
+ public S next() {
+ try {
+ if (null != this.nextResult) {
+ return this.nextResult;
+ } else {
+ advance();
+ if (null != this.nextResult)
+ return this.nextResult;
+ else
+ throw FastNoSuchElementException.instance();
+ }
+ } finally {
+ this.nextResult = null;
+ }
+ }
+
+ private final void advance() {
+ this.nextResult = null;
+ while (iterator.hasNext()) {
+ final S s = iterator.next();
+ if (predicate.test(s)) {
+ this.nextResult = s;
+ return;
+ }
+ }
+ }
+ };
+ }
+
+ public static final <S> Iterable<S> filter(final Iterable<S> iterable, final Predicate<S> predicate) {
+ return () -> IteratorUtils.filter(iterable.iterator(), predicate);
+ }
+
+ ///////////////////
+
+ public static final <S, E> Iterator<E> flatMap(final Iterator<S> iterator, final Function<S, Iterator<E>> function) {
+ return new Iterator<E>() {
+
+ private Iterator<E> currentIterator = Collections.emptyIterator();
+
+ @Override
+ public boolean hasNext() {
+ if (this.currentIterator.hasNext())
+ return true;
+ else {
+ while (iterator.hasNext()) {
+ this.currentIterator = function.apply(iterator.next());
+ if (this.currentIterator.hasNext())
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public void remove() {
+ iterator.remove();
+ }
+
+ @Override
+ public E next() {
+ if (this.hasNext())
+ return this.currentIterator.next();
+ else
+ throw FastNoSuchElementException.instance();
+ }
+ };
+ }
+
+
+ ///////////////////
+
+ public static final <S> Iterator<S> concat(final Iterator<S>... iterators) {
+ final MultiIterator<S> iterator = new MultiIterator<>();
+ for (final Iterator<S> itty : iterators) {
+ iterator.addIterator(itty);
+ }
+ return iterator;
+ }
+
+ ///////////////////
+
+ public static Iterator asIterator(final Object o) {
+ final Iterator itty;
+ if (o instanceof Iterable)
+ itty = ((Iterable) o).iterator();
+ else if (o instanceof Iterator)
+ itty = (Iterator) o;
+ else if (o instanceof Object[])
+ itty = new ArrayIterator<>((Object[]) o);
+ else if (o instanceof Stream)
+ itty = ((Stream) o).iterator();
+ else if (o instanceof Map)
+ itty = ((Map) o).entrySet().iterator();
+ else if (o instanceof Throwable)
+ itty = of(((Throwable) o).getMessage());
+ else
+ itty = of(o);
+ return itty;
+ }
+
+ public static List asList(final Object o) {
+ return list(asIterator(o));
+ }
+
+ /**
+ * Construct a {@link Stream} from an {@link Iterator}.
+ */
+ public static <T> Stream<T> stream(final Iterator<T> iterator) {
+ return StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, Spliterator.IMMUTABLE | Spliterator.SIZED), false);
+ }
+
+ public static <T> Stream<T> stream(final Iterable<T> iterable) {
+ return IteratorUtils.stream(iterable.iterator());
+ }
+
+ public static <T> Iterator<T> noRemove(final Iterator<T> iterator) {
+ return new Iterator<T>() {
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public void remove() {
+ // do nothing
+ }
+
+ @Override
+ public T next() {
+ return iterator.next();
+ }
+ };
+ }
+
+ public static <T> Iterator<T> removeOnNext(final Iterator<T> iterator) {
+ return new Iterator<T>() {
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public void remove() {
+ iterator.remove();
+ }
+
+ @Override
+ public T next() {
+ final T object = iterator.next();
+ iterator.remove();
+ return object;
+ }
+ };
+ }
+}
diff --git a/java/core/src/main/java/org/apache/tinkerpop/util/MultiIterator.java b/java/core/src/main/java/org/apache/tinkerpop/util/MultiIterator.java
new file mode 100644
index 0000000..a27b794
--- /dev/null
+++ b/java/core/src/main/java/org/apache/tinkerpop/util/MultiIterator.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.tinkerpop.util;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class MultiIterator<T> implements Iterator<T>, Serializable {
+
+ private final List<Iterator<T>> iterators = new ArrayList<>();
+ private int current = 0;
+
+ public void addIterator(final Iterator<T> iterator) {
+ this.iterators.add(iterator);
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (this.current >= this.iterators.size())
+ return false;
+
+ Iterator<T> currentIterator = this.iterators.get(this.current);
+
+ while (true) {
+ if (currentIterator.hasNext()) {
+ return true;
+ } else {
+ this.current++;
+ if (this.current >= iterators.size())
+ break;
+ currentIterator = iterators.get(this.current);
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public void remove() {
+ this.iterators.get(this.current).remove();
+ }
+
+ @Override
+ public T next() {
+ if (this.iterators.isEmpty()) throw FastNoSuchElementException.instance();
+
+ Iterator<T> currentIterator = iterators.get(this.current);
+ while (true) {
+ if (currentIterator.hasNext()) {
+ return currentIterator.next();
+ } else {
+ this.current++;
+ if (this.current >= iterators.size())
+ break;
+ currentIterator = iterators.get(current);
+ }
+ }
+ throw FastNoSuchElementException.instance();
+ }
+
+ public void clear() {
+ this.iterators.clear();
+ this.current = 0;
+ }
+
+}
\ No newline at end of file
diff --git a/java/core/src/main/java/org/apache/tinkerpop/machine/processor/EmptyProcessorFactory.java b/java/core/src/main/java/org/apache/tinkerpop/util/SingleIterator.java
similarity index 60%
copy from java/core/src/main/java/org/apache/tinkerpop/machine/processor/EmptyProcessorFactory.java
copy to java/core/src/main/java/org/apache/tinkerpop/util/SingleIterator.java
index 89233df..dedb923 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/machine/processor/EmptyProcessorFactory.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/util/SingleIterator.java
@@ -16,27 +16,40 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.tinkerpop.machine.processor;
+package org.apache.tinkerpop.util;
-import org.apache.tinkerpop.machine.bytecode.Bytecode;
+import java.io.Serializable;
+import java.util.Iterator;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public class EmptyProcessorFactory implements ProcessorFactory {
+final class SingleIterator<T> implements Iterator<T>, Serializable {
- private static final EmptyProcessorFactory INSTANCE = new EmptyProcessorFactory();
+ private T t;
+ private boolean alive = true;
- private EmptyProcessorFactory() {
+ protected SingleIterator(final T t) {
+ this.t = t;
+ }
+ @Override
+ public boolean hasNext() {
+ return this.alive;
}
@Override
- public <C, S, E> Processor<C, S, E> mint(final Bytecode<C> bytecode) {
- return EmptyProcessor.instance();
+ public void remove() {
+ this.t = null;
}
- public static EmptyProcessorFactory instance() {
- return INSTANCE;
+ @Override
+ public T next() {
+ if (!this.alive)
+ throw FastNoSuchElementException.instance();
+ else {
+ this.alive = false;
+ return t;
+ }
}
-}
+}
\ No newline at end of file
diff --git a/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/Beam.java b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/Beam.java
index 3e36565..fa06730 100644
--- a/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/Beam.java
+++ b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/Beam.java
@@ -31,9 +31,12 @@ import org.apache.tinkerpop.machine.coefficients.Coefficient;
import org.apache.tinkerpop.machine.coefficients.LongCoefficient;
import org.apache.tinkerpop.machine.functions.CFunction;
import org.apache.tinkerpop.machine.functions.FilterFunction;
+import org.apache.tinkerpop.machine.functions.FlatMapFunction;
import org.apache.tinkerpop.machine.functions.InitialFunction;
import org.apache.tinkerpop.machine.functions.MapFunction;
+import org.apache.tinkerpop.machine.functions.NestedFunction;
import org.apache.tinkerpop.machine.functions.ReduceFunction;
+import org.apache.tinkerpop.machine.pipes.PipesProcessor;
import org.apache.tinkerpop.machine.processor.Processor;
import org.apache.tinkerpop.machine.traversers.Traverser;
import org.apache.tinkerpop.machine.traversers.TraverserFactory;
@@ -54,7 +57,7 @@ public class Beam<C, S, E> implements Processor<C, S, E> {
Iterator<Traverser> iterator = null;
- public Beam(final List<CFunction<C>> functions, final TraverserFactory<C> traverserFactory) {
+ public Beam(final TraverserFactory<C> traverserFactory, final List<CFunction<C>> functions) {
this.pipeline = Pipeline.create();
this.pipeline.getCoderRegistry().registerCoderForClass(Traverser.class, new TraverserCoder<>());
PCollection collection = this.pipeline.apply(Create.of(traverserFactory.create((Coefficient) LongCoefficient.create(), 1L)));
@@ -62,10 +65,15 @@ public class Beam<C, S, E> implements Processor<C, S, E> {
DoFn fn = null;
for (final CFunction<?> function : functions) {
+ if (function instanceof NestedFunction)
+ ((NestedFunction<C, ?, ?>) function).setProcessor(traverserFactory, new PipesProcessor());
+
if (function instanceof InitialFunction) {
fn = new InitialFn<>((InitialFunction) function, traverserFactory);
} else if (function instanceof FilterFunction) {
fn = new FilterFn<>((FilterFunction) function);
+ } else if (function instanceof FlatMapFunction) {
+ fn = new FlatMapFn<>((FlatMapFunction) function);
} else if (function instanceof MapFunction) {
fn = new MapFn<>((MapFunction) function);
} else if (function instanceof ReduceFunction) {
@@ -88,7 +96,7 @@ public class Beam<C, S, E> implements Processor<C, S, E> {
}
public Beam(final Bytecode<C> bytecode) {
- this(BytecodeUtil.compile(BytecodeUtil.strategize(bytecode)), BytecodeUtil.getTraverserFactory(bytecode).get());
+ this(BytecodeUtil.getTraverserFactory(bytecode).get(), BytecodeUtil.compile(BytecodeUtil.strategize(bytecode)));
}
@Override
diff --git a/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/BeamProcessor.java b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/BeamProcessor.java
index 7da3f47..f063362 100644
--- a/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/BeamProcessor.java
+++ b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/BeamProcessor.java
@@ -19,8 +19,12 @@
package org.apache.tinkerpop.machine.beam;
import org.apache.tinkerpop.machine.bytecode.Bytecode;
+import org.apache.tinkerpop.machine.functions.CFunction;
import org.apache.tinkerpop.machine.processor.Processor;
import org.apache.tinkerpop.machine.processor.ProcessorFactory;
+import org.apache.tinkerpop.machine.traversers.TraverserFactory;
+
+import java.util.List;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -28,7 +32,7 @@ import org.apache.tinkerpop.machine.processor.ProcessorFactory;
public class BeamProcessor implements ProcessorFactory {
@Override
- public <C, S, E> Processor<C, S, E> mint(final Bytecode<C> bytecode) {
- return new Beam<>(bytecode);
+ public <C, S, E> Processor<C, S, E> mint(TraverserFactory<C> traverserFactory, List<CFunction<C>> cFunctions) {
+ return new Beam<>(traverserFactory,cFunctions);
}
}
diff --git a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/FlatMapFunction.java b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/FlatMapFn.java
similarity index 58%
copy from java/core/src/main/java/org/apache/tinkerpop/machine/functions/FlatMapFunction.java
copy to java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/FlatMapFn.java
index 3bc5d68..e3d2cd9 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/FlatMapFunction.java
+++ b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/FlatMapFn.java
@@ -16,15 +16,29 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.tinkerpop.machine.functions;
+package org.apache.tinkerpop.machine.beam;
+import org.apache.tinkerpop.machine.functions.FlatMapFunction;
import org.apache.tinkerpop.machine.traversers.Traverser;
import java.util.Iterator;
-import java.util.function.Function;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public interface FlatMapFunction<C, S, E> extends Function<Traverser<C, S>, Iterator<Traverser<C, E>>>, CFunction<C> {
-}
+public class FlatMapFn<C, S, E> extends AbstractFn<C, S, E> {
+
+ private FlatMapFunction<C, S, E> flatMapFunction;
+
+ public FlatMapFn(final FlatMapFunction<C, S, E> flatMapFunction) {
+ super(flatMapFunction);
+ this.flatMapFunction = flatMapFunction;
+ }
+
+ @ProcessElement
+ public void processElement(final @Element Traverser<C, S> traverser, final OutputReceiver<Traverser<C, E>> output) {
+ Iterator<Traverser<C, E>> iterator = traverser.flatMap(this.flatMapFunction);
+ while (iterator.hasNext())
+ output.output(iterator.next());
+ }
+}
\ No newline at end of file
diff --git a/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/MapFn.java b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/MapFn.java
index 296f491..4a67d92 100644
--- a/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/MapFn.java
+++ b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/MapFn.java
@@ -41,20 +41,8 @@ public class MapFn<C, S, E> extends AbstractFn<C, S, E> {
@ProcessElement
public void processElement(final @Element Traverser<C, S> traverser, final OutputReceiver<Traverser<C, E>> output) {
- if (this.first) {
- if (this.mapFunction instanceof NestedFunction) {
- Pipes beam = new Pipes(((NestedFunction) this.mapFunction).getFunctions(), new CompleteTraverserFactory());
- ((NestedFunction) this.mapFunction).setProcessor(beam);
- while (!this.traverserSet.isEmpty()) {
- beam.addStart(this.traverserSet.remove());
- }
- }
- this.first = false;
- }
- try {
+
output.output(traverser.map(this.mapFunction));
- } catch(final NoSuchElementException e) {
- // do nothing
- }
+
}
}
diff --git a/java/machine/beam/src/test/java/org/apache/tinkerpop/machine/beam/BeamTest.java b/java/machine/beam/src/test/java/org/apache/tinkerpop/machine/beam/BeamTest.java
index 910b792..5a5e020 100644
--- a/java/machine/beam/src/test/java/org/apache/tinkerpop/machine/beam/BeamTest.java
+++ b/java/machine/beam/src/test/java/org/apache/tinkerpop/machine/beam/BeamTest.java
@@ -38,7 +38,7 @@ public class BeamTest {
.withProcessor(BeamProcessor.class)
.withStrategy(IdentityStrategy.class);
- Traversal<Long, Long, Long> traversal = g.inject(7L, 10L, 12L).as("a").c(3L).map(__.incr()).identity().incr();
+ Traversal<Long, Long, Long> traversal = g.inject(7L, 10L, 12L).as("a").c(3L).map(__.incr()).identity().incr().sum();
System.out.println(TraversalUtil.getBytecode(traversal));
System.out.println(traversal);
System.out.println(traversal.toList());
@@ -47,5 +47,10 @@ public class BeamTest {
System.out.println(TraversalUtil.getBytecode(traversal));
System.out.println(traversal);
System.out.println(traversal.toList());
+ System.out.println("\n----------\n");
+ traversal = g.inject(7L).union(__.<Long>incr().incr(),__.incr());
+ System.out.println(TraversalUtil.getBytecode(traversal));
+ System.out.println(traversal);
+ System.out.println(traversal.toList());
}
}
diff --git a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/FlatMapFunction.java b/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/FlatMapStep.java
similarity index 51%
copy from java/core/src/main/java/org/apache/tinkerpop/machine/functions/FlatMapFunction.java
copy to java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/FlatMapStep.java
index 3bc5d68..468672a 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/FlatMapFunction.java
+++ b/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/FlatMapStep.java
@@ -16,15 +16,37 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.tinkerpop.machine.functions;
+package org.apache.tinkerpop.machine.pipes;
+import org.apache.tinkerpop.machine.functions.FlatMapFunction;
import org.apache.tinkerpop.machine.traversers.Traverser;
+import java.util.Collections;
import java.util.Iterator;
-import java.util.function.Function;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public interface FlatMapFunction<C, S, E> extends Function<Traverser<C, S>, Iterator<Traverser<C, E>>>, CFunction<C> {
-}
+public class FlatMapStep<C, S, E> extends AbstractStep<C, S, E> {
+
+ private final FlatMapFunction<C, S, E> flatMapFunction;
+ private Iterator<Traverser<C, E>> iterator = Collections.emptyIterator();
+
+ public FlatMapStep(final AbstractStep<C, ?, S> previousStep, final FlatMapFunction<C, S, E> flatMapFunction) {
+ super(previousStep, flatMapFunction);
+ this.flatMapFunction = flatMapFunction;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return this.iterator.hasNext() || super.hasNext();
+ }
+
+ @Override
+ public Traverser<C, E> next() {
+ if (!this.iterator.hasNext()) {
+ this.iterator = super.getPreviousTraverser().flatMap(this.flatMapFunction);
+ }
+ return this.iterator.next();
+ }
+}
\ No newline at end of file
diff --git a/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/Pipes.java b/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/Pipes.java
index a1730fc..7cd8872 100644
--- a/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/Pipes.java
+++ b/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/Pipes.java
@@ -18,10 +18,9 @@
*/
package org.apache.tinkerpop.machine.pipes;
-import org.apache.tinkerpop.machine.bytecode.Bytecode;
-import org.apache.tinkerpop.machine.bytecode.BytecodeUtil;
import org.apache.tinkerpop.machine.functions.CFunction;
import org.apache.tinkerpop.machine.functions.FilterFunction;
+import org.apache.tinkerpop.machine.functions.FlatMapFunction;
import org.apache.tinkerpop.machine.functions.InitialFunction;
import org.apache.tinkerpop.machine.functions.MapFunction;
import org.apache.tinkerpop.machine.functions.NestedFunction;
@@ -43,18 +42,20 @@ public class Pipes<C, S, E> implements Processor<C, S, E> {
private Step<C, ?, E> endStep;
private Step<C, S, ?> startStep = EmptyStep.instance();
- public Pipes(final List<CFunction<C>> functions, final TraverserFactory<C> traverserFactory) {
+ public Pipes(final TraverserFactory<C> traverserFactory, final List<CFunction<C>> functions) {
AbstractStep<C, ?, ?> previousStep = EmptyStep.instance();
for (final CFunction<?> function : functions) {
if (function instanceof NestedFunction)
- ((NestedFunction<C, ?, ?>) function).setProcessor(new Pipes(((NestedFunction<C, S, E>) function).getFunctions(), traverserFactory));
+ ((NestedFunction<C, ?, ?>) function).setProcessor(traverserFactory, new PipesProcessor());
+ /////////
final AbstractStep nextStep;
if (function instanceof FilterFunction)
nextStep = new FilterStep(previousStep, (FilterFunction<C, ?>) function);
+ else if (function instanceof FlatMapFunction)
+ nextStep = new FlatMapStep(previousStep, (FlatMapFunction<C, ?, ?>) function);
else if (function instanceof MapFunction)
nextStep = new MapStep(previousStep, (MapFunction<C, ?, ?>) function);
else if (function instanceof InitialFunction)
- // TODO: traverser factory
nextStep = new InitialStep((InitialFunction<C, S>) function, traverserFactory);
else if (function instanceof ReduceFunction)
nextStep = new ReduceStep(previousStep, (ReduceFunction<C, ?, ?>) function, new BasicReducer<>(((ReduceFunction<C, ?, ?>) function).getInitialValue()), traverserFactory);
@@ -70,10 +71,6 @@ public class Pipes<C, S, E> implements Processor<C, S, E> {
this.endStep = (Step<C, ?, E>) previousStep;
}
- public Pipes(final Bytecode<C> bytecode) {
- this(BytecodeUtil.compile(BytecodeUtil.strategize(bytecode)), BytecodeUtil.getTraverserFactory(bytecode).get());
- }
-
@Override
public void addStart(final Traverser<C, S> traverser) {
this.startStep.addStart(traverser);
diff --git a/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/PipesProcessor.java b/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/PipesProcessor.java
index 14f2e7a..5144d29 100644
--- a/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/PipesProcessor.java
+++ b/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/PipesProcessor.java
@@ -18,9 +18,12 @@
*/
package org.apache.tinkerpop.machine.pipes;
+import org.apache.tinkerpop.machine.functions.CFunction;
import org.apache.tinkerpop.machine.processor.Processor;
import org.apache.tinkerpop.machine.processor.ProcessorFactory;
-import org.apache.tinkerpop.machine.bytecode.Bytecode;
+import org.apache.tinkerpop.machine.traversers.TraverserFactory;
+
+import java.util.List;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -28,7 +31,7 @@ import org.apache.tinkerpop.machine.bytecode.Bytecode;
public class PipesProcessor implements ProcessorFactory {
@Override
- public <C, S, E> Processor<C, S, E> mint(final Bytecode<C> bytecode) {
- return new Pipes<>(bytecode);
+ public <C, S, E> Processor<C, S, E> mint(final TraverserFactory<C> traverserFactory, final List<CFunction<C>> functions) {
+ return new Pipes<>(traverserFactory,functions);
}
}
diff --git a/java/machine/pipes/src/test/java/org/apache/tinkerpop/machine/pipes/PipesTest.java b/java/machine/pipes/src/test/java/org/apache/tinkerpop/machine/pipes/PipesTest.java
index 94fdb20..1431049 100644
--- a/java/machine/pipes/src/test/java/org/apache/tinkerpop/machine/pipes/PipesTest.java
+++ b/java/machine/pipes/src/test/java/org/apache/tinkerpop/machine/pipes/PipesTest.java
@@ -25,7 +25,6 @@ import org.apache.tinkerpop.language.TraversalUtil;
import org.apache.tinkerpop.language.__;
import org.apache.tinkerpop.machine.coefficients.LongCoefficient;
import org.apache.tinkerpop.machine.strategies.IdentityStrategy;
-import org.apache.tinkerpop.machine.traversers.Path;
import org.junit.jupiter.api.Test;
/**
@@ -49,5 +48,10 @@ public class PipesTest {
System.out.println(TraversalUtil.getBytecode(traversal));
System.out.println(traversal);
System.out.println(traversal.toList());
+ System.out.println("\n----------\n");
+ traversal = g.inject(7L).union(__.<Long>incr().incr().count(), __.<Long, Long>c(10L).incr().sum()).count();
+ System.out.println(TraversalUtil.getBytecode(traversal));
+ System.out.println(traversal);
+ System.out.println(traversal.toList());
}
}