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/26 20:33:26 UTC
[tinkerpop] branch tp4 updated: Lots of clean up on TraverserServer,
RemoteMachine,
and MachineServer. Figured out cool Runnable Java functional-style.
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
The following commit(s) were added to refs/heads/tp4 by this push:
new d1b677f Lots of clean up on TraverserServer, RemoteMachine, and MachineServer. Figured out cool Runnable Java functional-style.
d1b677f is described below
commit d1b677f0e7bd47f1d75c464a57c82f8ca15cc400
Author: Marko A. Rodriguez <ok...@gmail.com>
AuthorDate: Tue Mar 26 14:33:16 2019 -0600
Lots of clean up on TraverserServer, RemoteMachine, and MachineServer. Figured out cool Runnable Java functional-style.
---
.../language/gremlin/TraversalSource.java | 2 +-
.../java/org/apache/tinkerpop/machine/Machine.java | 4 +-
.../tinkerpop/machine/species/BasicMachine.java | 6 +--
.../tinkerpop/machine/species/LocalMachine.java | 7 ++--
.../machine/species/remote/MachineServer.java | 43 +++++++++++-----------
.../machine/species/remote/RemoteMachine.java | 41 +++++++++++----------
.../tinkerpop/machine/species/remote/Request.java | 4 +-
.../machine/species/remote/TraverserServer.java | 43 +++++++++++-----------
.../machine/species/LocalMachineTest.java | 2 +-
.../tinkerpop/machine/processor/beam/Beam.java | 12 +++---
.../tinkerpop/machine/processor/beam/OutputFn.java | 7 ++--
.../tinkerpop/machine/processor/beam/ReduceFn.java | 4 +-
.../{serialization => io}/CoefficientCoder.java | 2 +-
.../beam/{serialization => io}/ReducerCoder.java | 2 +-
.../beam/{serialization => io}/TraverserCoder.java | 2 +-
.../processor/beam/sideeffect/InMemoryReducer.java | 2 +-
.../machine/processor/beam/util/TopologyUtil.java | 2 +-
.../tinkerpop/machine/processor/beam/BeamTest.java | 8 +++-
.../machine/processor/pipes/PipesTest.java | 2 +-
19 files changed, 101 insertions(+), 94 deletions(-)
diff --git a/java/language/gremlin/src/main/java/org/apache/tinkerpop/language/gremlin/TraversalSource.java b/java/language/gremlin/src/main/java/org/apache/tinkerpop/language/gremlin/TraversalSource.java
index e27f680..e8c0bd1 100644
--- a/java/language/gremlin/src/main/java/org/apache/tinkerpop/language/gremlin/TraversalSource.java
+++ b/java/language/gremlin/src/main/java/org/apache/tinkerpop/language/gremlin/TraversalSource.java
@@ -112,7 +112,7 @@ public class TraversalSource<C> implements Cloneable {
}
public void close() {
- this.machine.close(this.bytecode);
+ this.machine.unregister(this.bytecode);
}
@Override
diff --git a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/Machine.java b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/Machine.java
index fb9074f..8e4a781 100644
--- a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/Machine.java
+++ b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/Machine.java
@@ -33,5 +33,7 @@ public interface Machine extends Closeable {
public <C, E> Iterator<Traverser<C, E>> submit(final Bytecode<C> bytecode); // TODO: should return a "ShellTraverser"
- public <C> void close(final Bytecode<C> sourceCode);
+ public <C> void unregister(final Bytecode<C> sourceCode);
+
+ public void close();
}
diff --git a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/BasicMachine.java b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/BasicMachine.java
index 2bf8b2c..5dbe4aa 100644
--- a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/BasicMachine.java
+++ b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/BasicMachine.java
@@ -23,7 +23,6 @@ import org.apache.tinkerpop.machine.bytecode.Bytecode;
import org.apache.tinkerpop.machine.bytecode.compiler.Compilation;
import org.apache.tinkerpop.machine.traverser.Traverser;
-import java.io.IOException;
import java.util.Iterator;
/**
@@ -41,7 +40,7 @@ public final class BasicMachine implements Machine {
}
@Override
- public <C> void close(final Bytecode<C> sourceCode) {
+ public <C> void unregister(final Bytecode<C> sourceCode) {
}
@@ -55,7 +54,6 @@ public final class BasicMachine implements Machine {
}
@Override
- public void close() throws IOException {
-
+ public void close() {
}
}
diff --git a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/LocalMachine.java b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/LocalMachine.java
index 9e192ee..b29298a 100644
--- a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/LocalMachine.java
+++ b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/LocalMachine.java
@@ -26,7 +26,6 @@ import org.apache.tinkerpop.machine.bytecode.compiler.Compilation;
import org.apache.tinkerpop.machine.bytecode.compiler.SourceCompilation;
import org.apache.tinkerpop.machine.traverser.Traverser;
-import java.io.IOException;
import java.util.Iterator;
import java.util.Map;
import java.util.Optional;
@@ -62,12 +61,12 @@ public final class LocalMachine implements Machine {
this.sources.put(uuid, new SourceCompilation<>(sourceCode));
final Bytecode<C> registeredBytecode = new Bytecode<>();
registeredBytecode.addSourceInstruction(WITH_SOURCE_CODE, uuid.toString());
- registeredBytecode.getInstructions().addAll(sourceCode.getInstructions()); // all bytecode is returned
+ registeredBytecode.getInstructions().addAll(sourceCode.getInstructions()); // all bytecode is returned (though, typically, this will be empty)
return registeredBytecode;
}
@Override
- public <C> void close(final Bytecode<C> sourceCode) {
+ public <C> void unregister(final Bytecode<C> sourceCode) {
LocalMachine.getSourceId(sourceCode).ifPresent(this.sources::remove);
}
@@ -94,7 +93,7 @@ public final class LocalMachine implements Machine {
}
@Override
- public void close() throws IOException {
+ public void close() {
this.sources.clear();
}
}
diff --git a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/MachineServer.java b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/MachineServer.java
index 8c9377a..8c01808 100644
--- a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/MachineServer.java
+++ b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/MachineServer.java
@@ -23,7 +23,6 @@ import org.apache.tinkerpop.machine.species.LocalMachine;
import org.apache.tinkerpop.machine.traverser.Traverser;
import org.apache.tinkerpop.machine.traverser.species.EmptyTraverser;
-import java.io.Closeable;
import java.io.EOFException;
import java.io.IOException;
import java.io.ObjectInputStream;
@@ -36,23 +35,25 @@ import java.util.concurrent.atomic.AtomicBoolean;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public class MachineServer implements Runnable, Closeable {
+public final class MachineServer implements AutoCloseable {
- private final int serverPort;
- private ServerSocket serverSocket;
+ private static final int FLUSH_AMOUNT = 10;
+
+ private final int machineServerPort;
+ private ServerSocket machineServerSocket;
private AtomicBoolean serverAlive = new AtomicBoolean(Boolean.TRUE);
private final Machine machine = LocalMachine.open();
- public MachineServer(final int serverPort) {
- this.serverPort = serverPort;
- new Thread(this).start();
+ public MachineServer(final int machineServerPort) {
+ this.machineServerPort = machineServerPort;
+ new Thread(this::run).start();
}
- public void run() {
+ private void run() {
try {
- this.serverSocket = new ServerSocket(this.serverPort);
+ this.machineServerSocket = new ServerSocket(this.machineServerPort);
while (this.isAlive()) {
- final Socket clientSocket = this.serverSocket.accept();
+ final Socket clientSocket = this.machineServerSocket.accept();
new Thread(new Worker(clientSocket)).start();
}
} catch (final Exception e) {
@@ -68,14 +69,14 @@ public class MachineServer implements Runnable, Closeable {
public synchronized void close() {
try {
this.serverAlive.set(Boolean.FALSE);
- this.serverSocket.close();
+ this.machineServerSocket.close();
this.machine.close();
} catch (final IOException e) {
throw new RuntimeException(e.getMessage(), e);
}
}
- public class Worker implements Runnable {
+ private class Worker implements Runnable {
private final Socket clientSocket;
@@ -88,25 +89,25 @@ public class MachineServer implements Runnable, Closeable {
final ObjectInputStream input = new ObjectInputStream(this.clientSocket.getInputStream());
final ObjectOutputStream output = new ObjectOutputStream(this.clientSocket.getOutputStream());
while (true) {
- final Request<Object> request = (Request) input.readObject();
+ final Request<Object> request = (Request<Object>) input.readObject();
if (Request.Type.register == request.type) {
- output.writeObject(machine.register(request.bytecode));
+ output.writeObject(MachineServer.this.machine.register(request.bytecode));
output.flush();
} else if (Request.Type.submit == request.type) {
final Socket traverserServerSocket = new Socket(request.traverserServerLocation, request.traverserServerPort);
final ObjectOutputStream traverserOutput = new ObjectOutputStream(traverserServerSocket.getOutputStream());
- final Iterator<Traverser<Object, Object>> iterator = machine.submit(request.bytecode);
+ final Iterator<Traverser<Object, Object>> iterator = MachineServer.this.machine.submit(request.bytecode);
+ int flushCounter = 0;
while (iterator.hasNext()) {
+ flushCounter++;
traverserOutput.writeObject(iterator.next());
- traverserOutput.flush();
+ if (0 == flushCounter % FLUSH_AMOUNT) traverserOutput.flush();
}
- traverserOutput.writeObject(EmptyTraverser.instance());
+ traverserOutput.writeObject(EmptyTraverser.instance()); // this tells a TraverserServer that there are no more traversers
traverserOutput.flush();
traverserOutput.close();
- } else if (Request.Type.close == request.type) {
- machine.close(request.bytecode);
- } else {
- throw new IllegalStateException("This shouldn't happen: " + request);
+ } else { // Request.Type.close == request.type
+ MachineServer.this.machine.unregister(request.bytecode);
}
}
} catch (final EOFException e) {
diff --git a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/RemoteMachine.java b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/RemoteMachine.java
index 8874b2b..156193d 100644
--- a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/RemoteMachine.java
+++ b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/RemoteMachine.java
@@ -32,22 +32,22 @@ import java.util.Iterator;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public class RemoteMachine implements Machine {
+public final class RemoteMachine implements Machine, AutoCloseable {
- private final String localTraverserServerLocation;
- private final int localTraverserServerPort;
+ private final String traverserServerLocation;
+ private final int traverserServerPort;
private final Socket machineServer;
private final ObjectInputStream inputStream;
private final ObjectOutputStream outputStream;
- private RemoteMachine(final int localTraverserServerPort, final String machineLocation, final int machinePort) {
+ private RemoteMachine(final int traverserServerPort, final String machineServerLocation, final int machineServerPort) {
try {
- this.localTraverserServerLocation = InetAddress.getLocalHost().getHostAddress();
- this.localTraverserServerPort = localTraverserServerPort;
- this.machineServer = new Socket(machineLocation, machinePort);
- this.outputStream = new ObjectOutputStream(machineServer.getOutputStream());
- this.inputStream = new ObjectInputStream(machineServer.getInputStream());
+ this.traverserServerLocation = InetAddress.getLocalHost().getHostAddress();
+ this.traverserServerPort = traverserServerPort;
+ this.machineServer = new Socket(machineServerLocation, machineServerPort);
+ this.outputStream = new ObjectOutputStream(this.machineServer.getOutputStream());
+ this.inputStream = new ObjectInputStream(this.machineServer.getInputStream());
} catch (final Exception e) {
throw new RuntimeException(e.getMessage(), e);
}
@@ -67,9 +67,8 @@ public class RemoteMachine implements Machine {
@Override
public <C, E> Iterator<Traverser<C, E>> submit(final Bytecode<C> bytecode) {
try {
- final TraverserServer<C, E> traverserServer = new TraverserServer<>(this.localTraverserServerPort);
- new Thread(traverserServer).start();
- this.outputStream.writeObject(Request.submit(bytecode, this.localTraverserServerLocation, this.localTraverserServerPort));
+ final TraverserServer<C, E> traverserServer = new TraverserServer<>(this.traverserServerPort);
+ this.outputStream.writeObject(Request.submit(bytecode, this.traverserServerLocation, this.traverserServerPort));
this.outputStream.flush();
return traverserServer;
} catch (final IOException e) {
@@ -78,7 +77,7 @@ public class RemoteMachine implements Machine {
}
@Override
- public <C> void close(final Bytecode<C> sourceCode) {
+ public <C> void unregister(final Bytecode<C> sourceCode) {
try {
this.outputStream.writeObject(Request.close(sourceCode));
this.outputStream.flush();
@@ -87,14 +86,18 @@ public class RemoteMachine implements Machine {
}
}
- public static Machine open(final int localTraverserServerPort, final String machineLocation, final int machinePort) {
- return new RemoteMachine(localTraverserServerPort, machineLocation, machinePort);
+ public static Machine open(final int traverserServerPort, final String machineServerLocation, final int machineServerPort) {
+ return new RemoteMachine(traverserServerPort, machineServerLocation, machineServerPort);
}
@Override
- public void close() throws IOException {
- this.inputStream.close();
- this.outputStream.close();
- this.machineServer.close();
+ public void close() {
+ try {
+ this.inputStream.close();
+ this.outputStream.close();
+ this.machineServer.close();
+ } catch (final IOException e) {
+ throw new RuntimeException(e.getMessage(), e);
+ }
}
}
diff --git a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/Request.java b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/Request.java
index 27c1edc..48e471c 100644
--- a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/Request.java
+++ b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/Request.java
@@ -33,8 +33,8 @@ final class Request<C> implements Serializable {
public final Type type;
public final Bytecode<C> bytecode;
- public final String traverserServerLocation;
- public final int traverserServerPort;
+ final String traverserServerLocation;
+ final int traverserServerPort;
private Request(final Type type, final Bytecode<C> bytecode, final String traverserServerLocation, final int traverserServerPort) {
this.type = type;
diff --git a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/TraverserServer.java b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/TraverserServer.java
index 4bb1683..9c45994 100644
--- a/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/TraverserServer.java
+++ b/java/machine/machine-core/src/main/java/org/apache/tinkerpop/machine/species/remote/TraverserServer.java
@@ -33,7 +33,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public final class TraverserServer<C, S> implements Runnable, Iterator<Traverser<C, S>> {
+public final class TraverserServer<C, S> implements AutoCloseable, Iterator<Traverser<C, S>> {
private final TraverserSet<C, S> traverserSet = new TraverserSet<>();
private final int serverPort;
@@ -42,12 +42,13 @@ public final class TraverserServer<C, S> implements Runnable, Iterator<Traverser
public TraverserServer(final int serverPort) {
this.serverPort = serverPort;
+ new Thread(this::run).start();
}
- public void run() {
+ private void run() {
try {
this.serverSocket = new ServerSocket(this.serverPort);
- while (this.isAlive()) {
+ while (this.serverAlive.get()) {
final Socket clientSocket = this.serverSocket.accept();
new Thread(new Worker(clientSocket)).start();
}
@@ -57,25 +58,12 @@ public final class TraverserServer<C, S> implements Runnable, Iterator<Traverser
}
}
- private boolean isAlive() {
- return this.serverAlive.get();
- }
-
- public synchronized void stop() {
- try {
- this.serverAlive.set(Boolean.FALSE);
- this.serverSocket.close();
- } catch (final IOException e) {
- throw new RuntimeException(e.getMessage(), e);
- }
- }
-
@Override
public boolean hasNext() {
if (!this.traverserSet.isEmpty())
return true;
else {
- while (this.isAlive()) {
+ while (this.serverAlive.get()) {
if (!this.traverserSet.isEmpty())
return true;
}
@@ -88,6 +76,18 @@ public final class TraverserServer<C, S> implements Runnable, Iterator<Traverser
return this.traverserSet.remove();
}
+ @Override
+ public synchronized void close() {
+ if (this.serverAlive.get()) {
+ try {
+ this.serverAlive.set(Boolean.FALSE);
+ this.serverSocket.close();
+ } catch (final IOException e) {
+ throw new RuntimeException(e.getMessage(), e);
+ }
+ }
+ }
+
public class Worker implements Runnable {
private final Socket clientSocket;
@@ -101,12 +101,13 @@ public final class TraverserServer<C, S> implements Runnable, Iterator<Traverser
final ObjectInputStream input = new ObjectInputStream(this.clientSocket.getInputStream());
while (true) {
final Traverser<C, S> traverser = (Traverser<C, S>) input.readObject();
- if (traverser instanceof EmptyTraverser) {
- stop();
+ if (traverser instanceof EmptyTraverser) { // EmptyTraverser kills server
+ TraverserServer.this.close();
break;
- }
- traverserSet.add(traverser);
+ } else
+ TraverserServer.this.traverserSet.add(traverser);
}
+ input.close();
} catch (final EOFException e) {
// okay -- this is how the worker closes
} catch (final IOException | ClassNotFoundException e) {
diff --git a/java/machine/machine-core/src/test/java/org/apache/tinkerpop/machine/species/LocalMachineTest.java b/java/machine/machine-core/src/test/java/org/apache/tinkerpop/machine/species/LocalMachineTest.java
index ae08b75..0d45458 100644
--- a/java/machine/machine-core/src/test/java/org/apache/tinkerpop/machine/species/LocalMachineTest.java
+++ b/java/machine/machine-core/src/test/java/org/apache/tinkerpop/machine/species/LocalMachineTest.java
@@ -75,7 +75,7 @@ class LocalMachineTest {
assertEquals(1, machine.sources.size());
assertNotEquals(bytecode, machine.register(bytecode));
assertEquals(2, machine.sources.size());
- machine.close(bytecode);
+ machine.unregister(bytecode);
assertEquals(1, machine.sources.size());
}
}
diff --git a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/Beam.java b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/Beam.java
index 6db2e6f..fd0bf65 100644
--- a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/Beam.java
+++ b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/Beam.java
@@ -25,7 +25,7 @@ import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.tinkerpop.machine.bytecode.compiler.Compilation;
import org.apache.tinkerpop.machine.processor.Processor;
-import org.apache.tinkerpop.machine.processor.beam.serialization.TraverserCoder;
+import org.apache.tinkerpop.machine.processor.beam.io.TraverserCoder;
import org.apache.tinkerpop.machine.processor.beam.util.ExecutionPlanner;
import org.apache.tinkerpop.machine.processor.beam.util.TopologyUtil;
import org.apache.tinkerpop.machine.species.remote.TraverserServer;
@@ -89,14 +89,12 @@ public class Beam<C, S, E> implements Processor<C, S, E> {
private void setupPipeline() {
if (null == this.iterator) {
- if (this.createTraverserServer) {
- this.iterator = new TraverserServer<>(this.traverserServerPort);
- new Thread((TraverserServer) this.iterator).start();
- } else
- this.iterator = Collections.emptyIterator();
+ this.iterator = this.createTraverserServer ?
+ new TraverserServer<>(this.traverserServerPort) :
+ Collections.emptyIterator();
this.pipeline.run().waitUntilFinish();
if (this.iterator instanceof TraverserServer)
- ((TraverserServer<C, E>) this.iterator).stop();
+ ((TraverserServer<C, E>) this.iterator).close();
}
}
}
diff --git a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/OutputFn.java b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/OutputFn.java
index ee31a1a..0b27905 100644
--- a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/OutputFn.java
+++ b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/OutputFn.java
@@ -51,7 +51,7 @@ public class OutputFn<C, S> extends DoFn<Traverser<C, S>, Void> {
@StartBundle
public void startBundle() {
- // only create a connection if results are generated
+ // only create a connection if results are generated at this branch
if (null == this.traverserServerSocket) {
try {
this.traverserServerSocket = new Socket(this.traverserServerLocation, this.traverserServerPort);
@@ -76,6 +76,7 @@ public class OutputFn<C, S> extends DoFn<Traverser<C, S>, Void> {
if (null != this.traverserServerSocket) {
try {
this.outputStream.flush();
+ this.outputStream.close();
this.traverserServerSocket.close();
} catch (final Exception e) {
throw new RuntimeException(e.getMessage(), e);
@@ -83,8 +84,8 @@ public class OutputFn<C, S> extends DoFn<Traverser<C, S>, Void> {
}
}
- @Override
+ /*@Override
public String toString() {
return "";
- }
+ }*/
}
\ No newline at end of file
diff --git a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/ReduceFn.java b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/ReduceFn.java
index 91a280c..fa16ce8 100644
--- a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/ReduceFn.java
+++ b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/ReduceFn.java
@@ -22,8 +22,8 @@ import org.apache.beam.sdk.coders.CannotProvideCoderException;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderRegistry;
import org.apache.beam.sdk.transforms.Combine;
-import org.apache.tinkerpop.machine.processor.beam.serialization.ReducerCoder;
-import org.apache.tinkerpop.machine.processor.beam.serialization.TraverserCoder;
+import org.apache.tinkerpop.machine.processor.beam.io.ReducerCoder;
+import org.apache.tinkerpop.machine.processor.beam.io.TraverserCoder;
import org.apache.tinkerpop.machine.processor.beam.sideeffect.InMemoryReducer;
import org.apache.tinkerpop.machine.function.ReduceFunction;
import org.apache.tinkerpop.machine.traverser.Traverser;
diff --git a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/serialization/CoefficientCoder.java b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/io/CoefficientCoder.java
similarity index 96%
rename from java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/serialization/CoefficientCoder.java
rename to java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/io/CoefficientCoder.java
index 5cd6440..6aaf360 100644
--- a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/serialization/CoefficientCoder.java
+++ b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/io/CoefficientCoder.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.tinkerpop.machine.processor.beam.serialization;
+package org.apache.tinkerpop.machine.processor.beam.io;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderException;
diff --git a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/serialization/ReducerCoder.java b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/io/ReducerCoder.java
similarity index 97%
rename from java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/serialization/ReducerCoder.java
rename to java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/io/ReducerCoder.java
index 8bf18f3..614cec0 100644
--- a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/serialization/ReducerCoder.java
+++ b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/io/ReducerCoder.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.tinkerpop.machine.processor.beam.serialization;
+package org.apache.tinkerpop.machine.processor.beam.io;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderException;
diff --git a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/serialization/TraverserCoder.java b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/io/TraverserCoder.java
similarity index 96%
rename from java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/serialization/TraverserCoder.java
rename to java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/io/TraverserCoder.java
index d68f11b..1285ace 100644
--- a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/serialization/TraverserCoder.java
+++ b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/io/TraverserCoder.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.tinkerpop.machine.processor.beam.serialization;
+package org.apache.tinkerpop.machine.processor.beam.io;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderException;
diff --git a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/sideeffect/InMemoryReducer.java b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/sideeffect/InMemoryReducer.java
index 18c3cb1..bb01baf 100644
--- a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/sideeffect/InMemoryReducer.java
+++ b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/sideeffect/InMemoryReducer.java
@@ -21,7 +21,7 @@ package org.apache.tinkerpop.machine.processor.beam.sideeffect;
import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.tinkerpop.machine.function.ReduceFunction;
-import org.apache.tinkerpop.machine.processor.beam.serialization.ReducerCoder;
+import org.apache.tinkerpop.machine.processor.beam.io.ReducerCoder;
import org.apache.tinkerpop.machine.traverser.Traverser;
import org.apache.tinkerpop.machine.traverser.TraverserFactory;
diff --git a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/util/TopologyUtil.java b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/util/TopologyUtil.java
index b1eba6b..f3ccf45 100644
--- a/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/util/TopologyUtil.java
+++ b/java/machine/processor/beam/src/main/java/org/apache/tinkerpop/machine/processor/beam/util/TopologyUtil.java
@@ -45,7 +45,7 @@ import org.apache.tinkerpop.machine.processor.beam.ReduceFn;
import org.apache.tinkerpop.machine.processor.beam.RepeatDeadEndFn;
import org.apache.tinkerpop.machine.processor.beam.RepeatEndFn;
import org.apache.tinkerpop.machine.processor.beam.RepeatStartFn;
-import org.apache.tinkerpop.machine.processor.beam.serialization.TraverserCoder;
+import org.apache.tinkerpop.machine.processor.beam.io.TraverserCoder;
import org.apache.tinkerpop.machine.traverser.Traverser;
import org.apache.tinkerpop.machine.traverser.TraverserFactory;
diff --git a/java/machine/processor/beam/src/test/java/org/apache/tinkerpop/machine/processor/beam/BeamTest.java b/java/machine/processor/beam/src/test/java/org/apache/tinkerpop/machine/processor/beam/BeamTest.java
index 5e00a82..f3471c5 100644
--- a/java/machine/processor/beam/src/test/java/org/apache/tinkerpop/machine/processor/beam/BeamTest.java
+++ b/java/machine/processor/beam/src/test/java/org/apache/tinkerpop/machine/processor/beam/BeamTest.java
@@ -35,6 +35,7 @@ import org.junit.jupiter.api.Test;
import java.util.List;
import java.util.Map;
+import java.util.Random;
import static org.apache.tinkerpop.language.gremlin.core.__.constant;
import static org.apache.tinkerpop.language.gremlin.core.__.incr;
@@ -69,12 +70,15 @@ public class BeamTest {
}
@Test
- public void shouldWork() throws Exception {
+ public void shouldWork() {
final MachineServer server = new MachineServer(7777);
final Machine machine = RemoteMachine.open(6666, "localhost", 7777);
final TraversalSource<Long> g = Gremlin.<Long>traversal(machine)
//.withCoefficient(LongCoefficient.class)
- .withProcessor(BeamProcessor.class, Map.of("beam.traverserServer.location", "localhost", "beam.traverserServer.port", 6666))
+ //.withProcessor(BeamProcessor.class)
+ .withProcessor(BeamProcessor.class, new Random().nextBoolean() ?
+ Map.of("beam.traverserServer.location", "localhost", "beam.traverserServer.port", 6666) :
+ Map.of())
.withStrategy(IdentityStrategy.class);
Traversal<Long, ?, ?> traversal = g.inject(List.of(1L, 1L)).<Long>unfold().map(incr()).c(4L).repeat(incr()).until(__.is(__.constant(8L).incr().incr())).sum();
diff --git a/java/machine/processor/pipes/src/test/java/org/apache/tinkerpop/machine/processor/pipes/PipesTest.java b/java/machine/processor/pipes/src/test/java/org/apache/tinkerpop/machine/processor/pipes/PipesTest.java
index 5445757..5f3c20a 100644
--- a/java/machine/processor/pipes/src/test/java/org/apache/tinkerpop/machine/processor/pipes/PipesTest.java
+++ b/java/machine/processor/pipes/src/test/java/org/apache/tinkerpop/machine/processor/pipes/PipesTest.java
@@ -130,8 +130,8 @@ public class PipesTest {
System.out.println(traversal.hasNext());
///
g.close();
+ machine.close();
server.close();
-
}
@Test