You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@edgent.apache.org by dl...@apache.org on 2016/07/21 13:17:14 UTC
[13/54] [abbrv] [partial] incubator-quarks git commit: add
"org.apache." prefix to edgent package names
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/main/java/edgent/providers/direct/DirectTester.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/main/java/edgent/providers/direct/DirectTester.java b/providers/direct/src/main/java/edgent/providers/direct/DirectTester.java
deleted file mode 100644
index 46c072b..0000000
--- a/providers/direct/src/main/java/edgent/providers/direct/DirectTester.java
+++ /dev/null
@@ -1,130 +0,0 @@
-/*
-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 edgent.providers.direct;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicLong;
-
-import edgent.topology.TStream;
-import edgent.topology.spi.tester.AbstractTester;
-import edgent.topology.tester.Condition;
-
-class DirectTester extends AbstractTester {
-
- private final DirectTopology topology;
-
- DirectTester(DirectTopology topology) {
- this.topology = topology;
- }
-
- @Override
- public DirectTopology topology() {
- return topology;
- }
-
- @Override
- public Condition<Long> tupleCount(TStream<?> stream, final long expectedCount) {
- AtomicLong count = new AtomicLong();
- stream.sink(t -> {
- count.incrementAndGet();
- });
- return new Condition<Long>() {
-
- @Override
- public boolean valid() {
- return count.get() == expectedCount;
- }
-
- @Override
- public Long getResult() {
- return count.get();
- }
- };
- }
-
- @Override
- public <T> Condition<List<T>> streamContents(TStream<T> stream, @SuppressWarnings("unchecked") T... values) {
- List<T> contents = Collections.synchronizedList(new ArrayList<>());
- stream.sink(t -> contents.add(t));
- return new Condition<List<T>>() {
-
- @Override
- public boolean valid() {
- synchronized (contents) {
- return Arrays.asList(values).equals(contents);
- }
- }
-
- @Override
- public List<T> getResult() {
- return contents;
- }
- };
- }
-
- @Override
- public Condition<Long> atLeastTupleCount(TStream<?> stream, long expectedCount) {
- AtomicLong count = new AtomicLong();
- stream.sink(t -> {
- count.incrementAndGet();
- });
- return new Condition<Long>() {
-
- @Override
- public boolean valid() {
- return count.get() >= expectedCount;
- }
-
- @Override
- public Long getResult() {
- return count.get();
- }
- };
- }
-
- @Override
- public <T> Condition<List<T>> contentsUnordered(TStream<T> stream, @SuppressWarnings("unchecked") T... values) {
- List<T> contents = Collections.synchronizedList(new ArrayList<>());
- stream.sink(t -> contents.add(t));
- return new Condition<List<T>>() {
-
- @Override
- public boolean valid() {
- synchronized (contents) {
- if (contents.size() != values.length)
- return false;
- List<T> copy = new ArrayList<T>(contents);
- for (T expected : values) {
- if (!copy.remove(expected))
- return false;
- }
- return copy.isEmpty();
- }
- }
-
- @Override
- public List<T> getResult() {
- return contents;
- }
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/main/java/edgent/providers/direct/DirectTopology.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/main/java/edgent/providers/direct/DirectTopology.java b/providers/direct/src/main/java/edgent/providers/direct/DirectTopology.java
deleted file mode 100644
index 036c3e6..0000000
--- a/providers/direct/src/main/java/edgent/providers/direct/DirectTopology.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
-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 edgent.providers.direct;
-
-import java.util.concurrent.Callable;
-import java.util.concurrent.Future;
-
-import com.google.gson.JsonElement;
-import com.google.gson.JsonNull;
-import com.google.gson.JsonObject;
-
-import edgent.execution.Configs;
-import edgent.execution.Job;
-import edgent.execution.services.RuntimeServices;
-import edgent.execution.services.ServiceContainer;
-import edgent.function.Supplier;
-import edgent.graph.Graph;
-import edgent.runtime.etiao.Executable;
-import edgent.runtime.etiao.graph.DirectGraph;
-import edgent.topology.spi.graph.GraphTopology;
-
-/**
- * {@code DirectTopology} is a {@link GraphTopology} that
- * is executed in threads in the current virtual machine.
- * <P>
- * The topology is backed by a {@code DirectGraph} and its
- * execution is controlled and monitored by a {@code EtiaoJob}.
- * </P>
- */
-public class DirectTopology extends GraphTopology<DirectTester> {
-
- private final DirectGraph eg;
- private final Executable executable;
- private Job job; // created at submit time
-
- /**
- * Creates a {@code DirectTopology} instance.
- *
- * @param name topology name
- * @param container container which provides runtime services
- */
- DirectTopology(String name, ServiceContainer container) {
- super(name);
-
- this.eg = new DirectGraph(name, container);
- this.executable = eg.executable();
- }
-
- @Override
- public Graph graph() {
- return eg;
- }
-
- Executable getExecutable() {
- return executable;
- }
-
- @Override
- public Supplier<RuntimeServices> getRuntimeServiceSupplier() {
- return () -> getExecutable();
- }
-
- @Override
- protected DirectTester newTester() {
- return new DirectTester(this);
- }
-
- private Callable<Job> getCallable() {
- return new Callable<Job>() {
-
- @Override
- public Job call() throws Exception {
- execute();
- return job;
- }
- };
- }
-
- Future<Job> executeCallable(JsonObject config) {
- JsonElement value = null;
- if (config != null)
- value = config.get(Configs.JOB_NAME);
-
- String jobName = null;
- if (value != null && !(value instanceof JsonNull))
- jobName = value.getAsString();
-
- this.job = getExecutable().createJob(graph(), getName(), jobName);
- return getExecutable().getScheduler().submit(getCallable());
- }
-
- private void execute() {
- job.stateChange(Job.Action.INITIALIZE);
- job.stateChange(Job.Action.START);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/main/java/edgent/providers/direct/package-info.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/main/java/edgent/providers/direct/package-info.java b/providers/direct/src/main/java/edgent/providers/direct/package-info.java
deleted file mode 100644
index a8cdfee..0000000
--- a/providers/direct/src/main/java/edgent/providers/direct/package-info.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
-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.
-*/
-
-/**
- * Direct execution of a streaming topology.
- */
-package edgent.providers.direct;
-
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/main/java/org/apache/edgent/providers/direct/DirectProvider.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/main/java/org/apache/edgent/providers/direct/DirectProvider.java b/providers/direct/src/main/java/org/apache/edgent/providers/direct/DirectProvider.java
new file mode 100644
index 0000000..f8babbf
--- /dev/null
+++ b/providers/direct/src/main/java/org/apache/edgent/providers/direct/DirectProvider.java
@@ -0,0 +1,89 @@
+/*
+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.edgent.providers.direct;
+
+import java.util.concurrent.Future;
+
+import org.apache.edgent.execution.DirectSubmitter;
+import org.apache.edgent.execution.Job;
+import org.apache.edgent.execution.services.ControlService;
+import org.apache.edgent.execution.services.ServiceContainer;
+import org.apache.edgent.runtime.jsoncontrol.JsonControlService;
+import org.apache.edgent.topology.Topology;
+import org.apache.edgent.topology.TopologyProvider;
+import org.apache.edgent.topology.spi.AbstractTopologyProvider;
+
+import com.google.gson.JsonObject;
+
+/**
+ * {@code DirectProvider} is a {@link TopologyProvider} that
+ * runs a submitted topology as a {@link Job} in threads
+ * in the current virtual machine.
+ * <P>
+ * A job (execution of a topology) continues to execute
+ * while any of its elements have remaining work,
+ * such as any of the topology's source streams are capable
+ * of generating tuples.
+ * <BR>
+ * "Endless" source streams never terminate - e.g., a stream
+ * created by {@link Topology#generate(org.apache.edgent.function.Supplier) generate()},
+ * {@link Topology#poll(org.apache.edgent.function.Supplier, long, java.util.concurrent.TimeUnit) poll()},
+ * or {@link Topology#events(org.apache.edgent.function.Consumer) events()}.
+ * Hence a job with such sources runs until either it or some other
+ * entity terminates it.
+ * </P>
+ */
+public class DirectProvider extends AbstractTopologyProvider<DirectTopology>
+ implements DirectSubmitter<Topology, Job> {
+
+ private final ServiceContainer services;
+
+ public DirectProvider() {
+ this.services = new ServiceContainer();
+
+ getServices().addService(ControlService.class, new JsonControlService());
+ }
+
+ /**
+ * {@inheritDoc}
+ * <P>
+ * The returned services instance is shared
+ * across all jobs submitted to this provider.
+ * </P>
+ */
+ @Override
+ public ServiceContainer getServices() {
+ return services;
+ }
+
+ @Override
+ public DirectTopology newTopology(String name) {
+ return new DirectTopology(name, services);
+ }
+
+ @Override
+ public Future<Job> submit(Topology topology) {
+ return submit(topology, new JsonObject());
+ }
+
+ @Override
+ public Future<Job> submit(Topology topology, JsonObject config) {
+ return ((DirectTopology) topology).executeCallable(config);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/main/java/org/apache/edgent/providers/direct/DirectTester.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/main/java/org/apache/edgent/providers/direct/DirectTester.java b/providers/direct/src/main/java/org/apache/edgent/providers/direct/DirectTester.java
new file mode 100644
index 0000000..73a3fb8
--- /dev/null
+++ b/providers/direct/src/main/java/org/apache/edgent/providers/direct/DirectTester.java
@@ -0,0 +1,130 @@
+/*
+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.edgent.providers.direct;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.edgent.topology.TStream;
+import org.apache.edgent.topology.spi.tester.AbstractTester;
+import org.apache.edgent.topology.tester.Condition;
+
+class DirectTester extends AbstractTester {
+
+ private final DirectTopology topology;
+
+ DirectTester(DirectTopology topology) {
+ this.topology = topology;
+ }
+
+ @Override
+ public DirectTopology topology() {
+ return topology;
+ }
+
+ @Override
+ public Condition<Long> tupleCount(TStream<?> stream, final long expectedCount) {
+ AtomicLong count = new AtomicLong();
+ stream.sink(t -> {
+ count.incrementAndGet();
+ });
+ return new Condition<Long>() {
+
+ @Override
+ public boolean valid() {
+ return count.get() == expectedCount;
+ }
+
+ @Override
+ public Long getResult() {
+ return count.get();
+ }
+ };
+ }
+
+ @Override
+ public <T> Condition<List<T>> streamContents(TStream<T> stream, @SuppressWarnings("unchecked") T... values) {
+ List<T> contents = Collections.synchronizedList(new ArrayList<>());
+ stream.sink(t -> contents.add(t));
+ return new Condition<List<T>>() {
+
+ @Override
+ public boolean valid() {
+ synchronized (contents) {
+ return Arrays.asList(values).equals(contents);
+ }
+ }
+
+ @Override
+ public List<T> getResult() {
+ return contents;
+ }
+ };
+ }
+
+ @Override
+ public Condition<Long> atLeastTupleCount(TStream<?> stream, long expectedCount) {
+ AtomicLong count = new AtomicLong();
+ stream.sink(t -> {
+ count.incrementAndGet();
+ });
+ return new Condition<Long>() {
+
+ @Override
+ public boolean valid() {
+ return count.get() >= expectedCount;
+ }
+
+ @Override
+ public Long getResult() {
+ return count.get();
+ }
+ };
+ }
+
+ @Override
+ public <T> Condition<List<T>> contentsUnordered(TStream<T> stream, @SuppressWarnings("unchecked") T... values) {
+ List<T> contents = Collections.synchronizedList(new ArrayList<>());
+ stream.sink(t -> contents.add(t));
+ return new Condition<List<T>>() {
+
+ @Override
+ public boolean valid() {
+ synchronized (contents) {
+ if (contents.size() != values.length)
+ return false;
+ List<T> copy = new ArrayList<T>(contents);
+ for (T expected : values) {
+ if (!copy.remove(expected))
+ return false;
+ }
+ return copy.isEmpty();
+ }
+ }
+
+ @Override
+ public List<T> getResult() {
+ return contents;
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/main/java/org/apache/edgent/providers/direct/DirectTopology.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/main/java/org/apache/edgent/providers/direct/DirectTopology.java b/providers/direct/src/main/java/org/apache/edgent/providers/direct/DirectTopology.java
new file mode 100644
index 0000000..c7bb758
--- /dev/null
+++ b/providers/direct/src/main/java/org/apache/edgent/providers/direct/DirectTopology.java
@@ -0,0 +1,112 @@
+/*
+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.edgent.providers.direct;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+
+import org.apache.edgent.execution.Configs;
+import org.apache.edgent.execution.Job;
+import org.apache.edgent.execution.services.RuntimeServices;
+import org.apache.edgent.execution.services.ServiceContainer;
+import org.apache.edgent.function.Supplier;
+import org.apache.edgent.graph.Graph;
+import org.apache.edgent.runtime.etiao.Executable;
+import org.apache.edgent.runtime.etiao.graph.DirectGraph;
+import org.apache.edgent.topology.spi.graph.GraphTopology;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonNull;
+import com.google.gson.JsonObject;
+
+/**
+ * {@code DirectTopology} is a {@link GraphTopology} that
+ * is executed in threads in the current virtual machine.
+ * <P>
+ * The topology is backed by a {@code DirectGraph} and its
+ * execution is controlled and monitored by a {@code EtiaoJob}.
+ * </P>
+ */
+public class DirectTopology extends GraphTopology<DirectTester> {
+
+ private final DirectGraph eg;
+ private final Executable executable;
+ private Job job; // created at submit time
+
+ /**
+ * Creates a {@code DirectTopology} instance.
+ *
+ * @param name topology name
+ * @param container container which provides runtime services
+ */
+ DirectTopology(String name, ServiceContainer container) {
+ super(name);
+
+ this.eg = new DirectGraph(name, container);
+ this.executable = eg.executable();
+ }
+
+ @Override
+ public Graph graph() {
+ return eg;
+ }
+
+ Executable getExecutable() {
+ return executable;
+ }
+
+ @Override
+ public Supplier<RuntimeServices> getRuntimeServiceSupplier() {
+ return () -> getExecutable();
+ }
+
+ @Override
+ protected DirectTester newTester() {
+ return new DirectTester(this);
+ }
+
+ private Callable<Job> getCallable() {
+ return new Callable<Job>() {
+
+ @Override
+ public Job call() throws Exception {
+ execute();
+ return job;
+ }
+ };
+ }
+
+ Future<Job> executeCallable(JsonObject config) {
+ JsonElement value = null;
+ if (config != null)
+ value = config.get(Configs.JOB_NAME);
+
+ String jobName = null;
+ if (value != null && !(value instanceof JsonNull))
+ jobName = value.getAsString();
+
+ this.job = getExecutable().createJob(graph(), getName(), jobName);
+ return getExecutable().getScheduler().submit(getCallable());
+ }
+
+ private void execute() {
+ job.stateChange(Job.Action.INITIALIZE);
+ job.stateChange(Job.Action.START);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/main/java/org/apache/edgent/providers/direct/package-info.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/main/java/org/apache/edgent/providers/direct/package-info.java b/providers/direct/src/main/java/org/apache/edgent/providers/direct/package-info.java
new file mode 100644
index 0000000..d46fdde
--- /dev/null
+++ b/providers/direct/src/main/java/org/apache/edgent/providers/direct/package-info.java
@@ -0,0 +1,24 @@
+/*
+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.
+*/
+
+/**
+ * Direct execution of a streaming topology.
+ */
+package org.apache.edgent.providers.direct;
+
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/edgent/test/providers/direct/DirectJobTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/edgent/test/providers/direct/DirectJobTest.java b/providers/direct/src/test/java/edgent/test/providers/direct/DirectJobTest.java
deleted file mode 100644
index b00aa01..0000000
--- a/providers/direct/src/test/java/edgent/test/providers/direct/DirectJobTest.java
+++ /dev/null
@@ -1,343 +0,0 @@
-/*
-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 edgent.test.providers.direct;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertTrue;
-
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.junit.Test;
-
-import com.google.gson.JsonObject;
-
-import edgent.execution.Configs;
-import edgent.execution.Job;
-import edgent.graph.Vertex;
-import edgent.oplet.Oplet;
-import edgent.oplet.core.PeriodicSource;
-import edgent.oplet.core.Pipe;
-import edgent.providers.direct.DirectProvider;
-import edgent.test.topology.TopologyAbstractTest;
-import edgent.topology.TStream;
-import edgent.topology.Topology;
-
-public class DirectJobTest extends TopologyAbstractTest implements DirectTestSetup {
- @Test
- public void jobName0() throws Exception {
- String[] data = new String[] {};
- String topologyName = "topoName";
- Topology t = newTopology(topologyName);
- t.strings(data);
- Job job = awaitCompleteExecution(t);
- assertTrue(job.getName().startsWith(topologyName));
- }
-
- @Test
- public void jobName1() throws Exception {
- String[] data = new String[] {};
- String topologyName = "topoName";
- Topology t = newTopology(topologyName);
- t.strings(data);
- JsonObject config = new JsonObject();
- config.addProperty(Configs.JOB_NAME, (String)null);
- Job job = awaitCompleteExecution(t, config);
- assertTrue(job.getName().startsWith(topologyName));
- }
-
- @Test
- public void jobName2() throws Exception {
- String[] data = new String[] {};
- String jobName = "testJob";
- Topology t = newTopology();
- t.strings(data);
- JsonObject config = new JsonObject();
- config.addProperty(Configs.JOB_NAME, jobName);
- Job job = awaitCompleteExecution(t, config);
- assertEquals(jobName, job.getName());
- }
-
- @Test
- public void jobDone0() throws Exception {
- String[] data = new String[] {};
- Topology t = newTopology();
- @SuppressWarnings("unused")
- TStream<String> s = t.strings(data);
-
- Job job = awaitCompleteExecution(t);
- assertEquals("job.getCurrentState() must be RUNNING", Job.State.RUNNING, job.getCurrentState());
- assertEquals("job.getCurrentState() must be HEALTHY", Job.Health.HEALTHY, job.getHealth());
- assertEquals("", job.getLastError());
-
- job.stateChange(Job.Action.CLOSE);
- assertEquals("job.getCurrentState() must be CLOSED", Job.State.CLOSED, job.getCurrentState());
- assertEquals("job.getCurrentState() must be HEALTHY", Job.Health.HEALTHY, job.getHealth());
- assertEquals("", job.getLastError());
- }
-
- @Test
- public void jobDone1() throws Exception {
- String[] data = new String[] {"a", "b", "c"};
- Topology t = newTopology();
- @SuppressWarnings("unused")
- TStream<String> s = t.strings(data);
-
- Job job = awaitCompleteExecution(t);
- assertEquals("job.getCurrentState() must be RUNNING", Job.State.RUNNING, job.getCurrentState());
- assertEquals("job.getCurrentState() must be HEALTHY", Job.Health.HEALTHY, job.getHealth());
- assertEquals("", job.getLastError());
- job.stateChange(Job.Action.CLOSE);
- assertEquals("job.getCurrentState() must be CLOSED", Job.State.CLOSED, job.getCurrentState());
- assertEquals("job.getCurrentState() must be HEALTHY", Job.Health.HEALTHY, job.getHealth());
- assertEquals("", job.getLastError());
- }
-
- @Test
- public void jobDone2() throws Exception {
- final int NUM_TUPLES = 1000000;
- Integer[] data = new Integer[NUM_TUPLES];
- AtomicInteger numTuples = new AtomicInteger();
-
- for (int i = 0; i < data.length; i++) {
- data[i] = new Integer(i);
- }
- Topology t = newTopology();
- TStream<Integer> ints = t.collection(Arrays.asList(data));
- ints.sink(tuple -> numTuples.incrementAndGet());
-
- Job job = awaitCompleteExecution(t);
- Thread.sleep(1500); // wait for numTuples visibility
- assertEquals(NUM_TUPLES, numTuples.get());
- assertEquals("job.getCurrentState() must be RUNNING", Job.State.RUNNING, job.getCurrentState());
- assertEquals("job.getCurrentState() must be HEALTHY", Job.Health.HEALTHY, job.getHealth());
- assertEquals("", job.getLastError());
- job.stateChange(Job.Action.CLOSE);
- assertEquals("job.getCurrentState() must be CLOSED", Job.State.CLOSED, job.getCurrentState());
- assertEquals("job.getCurrentState() must be HEALTHY", Job.Health.HEALTHY, job.getHealth());
- assertEquals("", job.getLastError());
- }
-
- @Test
- public void jobPeriodicSource() throws Exception {
- Topology t = newTopology();
- AtomicInteger n = new AtomicInteger(0);
- @SuppressWarnings("unused")
- TStream<Integer> ints = t.poll(() -> n.incrementAndGet(), 100, TimeUnit.MILLISECONDS);
-
- Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t);
- Job job = fj.get();
- assertEquals(Job.State.RUNNING, job.getCurrentState());
- Thread.sleep(TimeUnit.SECONDS.toMillis(2));
- assertTrue(n.get() > 0); // At least one tuple was processed
- job.stateChange(Job.Action.CLOSE);
- assertEquals(Job.State.CLOSED, job.getCurrentState());
- }
-
- @Test
- public void jobPeriodicSourceCancellation() throws Exception {
- Topology t = newTopology();
- AtomicInteger n = new AtomicInteger(0);
- @SuppressWarnings("unused")
- TStream<Integer> ints = t.poll(() -> n.incrementAndGet(), 500, TimeUnit.MILLISECONDS);
-
- // Get the source oplet
- Collection<Vertex<? extends Oplet<?, ?>, ?, ?>> vertices = t.graph().getVertices();
- PeriodicSource<?> src = null;
- for (Vertex<? extends Oplet<?, ?>, ?, ?> v : vertices) {
- Oplet<?,?> op = v.getInstance();
- assertTrue(op instanceof PeriodicSource);
- src = (PeriodicSource<?>) op;
- assertEquals(500, src.getPeriod());
- assertSame(TimeUnit.MILLISECONDS, src.getUnit());
- }
-
- // Submit job
- Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t);
- Job job = fj.get();
- assertEquals(Job.State.RUNNING, job.getCurrentState());
- Thread.sleep(TimeUnit.SECONDS.toMillis(1));
- int tupleCount = n.get();
- assertTrue("Expected more tuples than "+ tupleCount, tupleCount > 0); // At least one tuple was processed
- assertEquals(Job.Health.HEALTHY, job.getHealth());
- assertEquals("", job.getLastError());
-
- // Changing the period cancels the source's task and schedules new one
- src.setPeriod(100);
-
- Thread.sleep(TimeUnit.SECONDS.toMillis(1));
-
- // More tuples processed after resetting the period
- assertTrue("Expected more tuples than "+ n.get(), n.get() > 3*tupleCount);
-
- job.stateChange(Job.Action.CLOSE);
- assertEquals(Job.State.CLOSED, job.getCurrentState());
- assertEquals(Job.Health.HEALTHY, job.getHealth());
- assertEquals("", job.getLastError());
- }
-
- @Test
- public void jobProcessSource() throws Exception {
- Topology t = newTopology();
- AtomicInteger n = new AtomicInteger(0);
- @SuppressWarnings("unused")
- TStream<Integer> ints = t.generate(() -> n.incrementAndGet());
-
- Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t);
- Job job = fj.get();
- assertEquals(Job.State.RUNNING, job.getCurrentState());
- assertEquals(Job.Health.HEALTHY, job.getHealth());
- assertEquals("", job.getLastError());
- Thread.sleep(TimeUnit.SECONDS.toMillis(2));
- assertTrue(n.get() > 0); // At least one tuple was processed
- job.stateChange(Job.Action.CLOSE);
- assertEquals(Job.State.CLOSED, job.getCurrentState());
- assertEquals(Job.Health.HEALTHY, job.getHealth());
- assertEquals("", job.getLastError());
- }
-
- @Test(expected = TimeoutException.class)
- public void jobTimesOut() throws Exception {
- Topology t = newTopology();
- AtomicInteger n = new AtomicInteger(0);
- @SuppressWarnings("unused")
- TStream<Integer> ints = t.poll(() -> n.incrementAndGet(), 100, TimeUnit.MILLISECONDS);
-
- Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t);
- Job job = fj.get();
- assertEquals(Job.State.RUNNING, job.getCurrentState());
- try {
- job.complete(700, TimeUnit.MILLISECONDS);
- } finally {
- assertTrue(n.get() > 0); // At least one tuple was processed
- assertEquals(Job.State.RUNNING, job.getCurrentState());
- assertEquals(Job.Health.HEALTHY, job.getHealth());
- assertEquals("", job.getLastError());
- }
- }
-
- @Test(expected = ExecutionException.class)
- public void jobPeriodicSourceError() throws Exception {
- Topology t = newTopology();
- AtomicInteger n = new AtomicInteger(0);
- TStream<Integer> ints = t.poll(() -> n.incrementAndGet(), 100, TimeUnit.MILLISECONDS);
- ints.pipe(new FailedOplet<Integer>(5, 0));
-
- Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t);
- Job job = fj.get();
- assertEquals(Job.State.RUNNING, job.getCurrentState());
- try {
- job.complete(10, TimeUnit.SECONDS);
- } finally {
- // RUNNING even though execution error
- assertEquals(Job.State.RUNNING, job.getCurrentState());
- assertEquals(Job.Health.UNHEALTHY, job.getHealth());
- assertEquals("java.lang.RuntimeException: Expected Test Exception", job.getLastError());
- }
- }
-
- @Test(expected = ExecutionException.class)
- public void jobProcessSourceError() throws Exception {
- Topology t = newTopology();
- AtomicInteger n = new AtomicInteger(0);
- TStream<Integer> ints = t.generate(() -> n.incrementAndGet());
- ints.pipe(new FailedOplet<Integer>(12, 100));
-
- Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t);
- Job job = fj.get();
- assertEquals(Job.State.RUNNING, job.getCurrentState());
- try {
- job.complete(10, TimeUnit.SECONDS);
- } finally {
- // RUNNING even though execution error
- assertEquals(Job.State.RUNNING, job.getCurrentState());
- assertEquals(Job.Health.UNHEALTHY, job.getHealth());
- assertEquals("java.lang.RuntimeException: Expected Test Exception", job.getLastError());
- }
- }
-
- private Job awaitCompleteExecution(Topology t) throws InterruptedException, ExecutionException {
- return awaitCompleteExecution(t, null);
- }
-
- private Job awaitCompleteExecution(Topology t, JsonObject config) throws InterruptedException, ExecutionException {
- Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t, config);
- Job job = fj.get();
- job.complete();
- return job;
- }
-
- /**
- * Test oplet which fails after receiving a given number of tuples.
- * @param <T>
- */
- @SuppressWarnings("serial")
- private static class FailedOplet<T> extends Pipe<T,T> {
- private final int threshold;
- private final int sleepMillis;
-
- /**
- * Create test oplet.
- *
- * @param afterTuples number of tuples to receive before failing
- * @param sleepMillis milliseconds of sleep before processing each tuple
- */
- public FailedOplet(int afterTuples, int sleepMillis) {
- if (afterTuples < 0)
- throw new IllegalArgumentException("afterTuples="+afterTuples);
- if (sleepMillis < 0)
- throw new IllegalArgumentException("sleepMillis="+sleepMillis);
- this.threshold = afterTuples;
- this.sleepMillis = sleepMillis;
- }
-
- @Override
- public void close() throws Exception {
- }
- @Override
- public void accept(T tuple) {
- if (sleepMillis > 0) {
- sleep(sleepMillis);
- }
- injectError(threshold);
- submit(tuple);
- }
-
- private AtomicInteger count = new AtomicInteger(0);
- protected void injectError(int errorAt) {
- if (count.getAndIncrement() == errorAt)
- throw new RuntimeException("Expected Test Exception");
- }
-
- protected static void sleep(long millis) {
- try {
- Thread.sleep(millis);
- }
- catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/edgent/test/providers/direct/DirectTestSetup.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/edgent/test/providers/direct/DirectTestSetup.java b/providers/direct/src/test/java/edgent/test/providers/direct/DirectTestSetup.java
deleted file mode 100644
index 738de5f..0000000
--- a/providers/direct/src/test/java/edgent/test/providers/direct/DirectTestSetup.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
-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 edgent.test.providers.direct;
-
-import edgent.execution.Job;
-import edgent.execution.Submitter;
-import edgent.providers.direct.DirectProvider;
-import edgent.test.topology.TopologyTestSetup;
-import edgent.topology.Topology;
-
-public interface DirectTestSetup extends TopologyTestSetup {
- @Override
- default DirectProvider createTopologyProvider() {
- return new DirectProvider();
- }
-
- @Override
- default Submitter<Topology, Job> createSubmitter() {
- return (DirectProvider) getTopologyProvider();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/edgent/test/providers/direct/appservice/AppServiceTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/edgent/test/providers/direct/appservice/AppServiceTest.java b/providers/direct/src/test/java/edgent/test/providers/direct/appservice/AppServiceTest.java
deleted file mode 100644
index 2e8ea45..0000000
--- a/providers/direct/src/test/java/edgent/test/providers/direct/appservice/AppServiceTest.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
-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 edgent.test.providers.direct.appservice;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
-import java.io.File;
-import java.net.URL;
-
-import org.junit.Test;
-
-import edgent.providers.direct.DirectProvider;
-import edgent.runtime.appservice.AppService;
-import edgent.topology.services.ApplicationService;
-
-public class AppServiceTest {
-
-
- @Test
- public void testRegisterTopology() {
- DirectProvider direct = new DirectProvider();
- ApplicationService appService = AppService.createAndRegister(direct, direct);
-
- assertTrue(appService.getApplicationNames().isEmpty());
-
- appService.registerTopology("FirstApp", (t,c) -> t.strings("a"));
- assertEquals(1, appService.getApplicationNames().size());
- assertTrue(appService.getApplicationNames().contains("FirstApp"));
-
- appService.registerTopology("SecondApp", (t,c) -> t.strings("b"));
- assertEquals(2, appService.getApplicationNames().size());
- assertTrue(appService.getApplicationNames().contains("FirstApp"));
- assertTrue(appService.getApplicationNames().contains("SecondApp"));
- }
-
- @Test
- public void testRegisterJar() throws Exception {
- DirectProvider direct = new DirectProvider();
- ApplicationService appService = AppService.createAndRegister(direct, direct);
-
- String qd = System.getProperty("edgent.test.root.dir");
- assertNotNull(qd);
- File testAppsJar = new File(qd, "api/topology/test.classes/edgent.api.topology.APPS.TEST.jar");
- assertTrue(testAppsJar.exists());
-
- URL testAppsJarURL = testAppsJar.toURI().toURL();
-
- appService.registerJar(testAppsJarURL.toExternalForm(), null);
-
- assertEquals(3, appService.getApplicationNames().size());
- assertTrue(appService.getApplicationNames().contains("FirstJarApp"));
- assertTrue(appService.getApplicationNames().contains("SecondJarApp"));
- assertTrue(appService.getApplicationNames().contains("ThirdJarApp"));
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricObjectNameFactoryTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricObjectNameFactoryTest.java b/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricObjectNameFactoryTest.java
deleted file mode 100644
index 11db97b..0000000
--- a/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricObjectNameFactoryTest.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
-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 edgent.test.providers.direct.metrics;
-
-import edgent.test.metrics.MetricObjectNameFactoryTest;
-import edgent.test.providers.direct.DirectTestSetup;
-
-public class DirectMetricObjectNameFactoryTest extends MetricObjectNameFactoryTest implements DirectTestSetup {
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricsEverywhereTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricsEverywhereTest.java b/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricsEverywhereTest.java
deleted file mode 100644
index 2d06252..0000000
--- a/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricsEverywhereTest.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
-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 edgent.test.providers.direct.metrics;
-
-import edgent.test.metrics.MetricsEverywhereTest;
-import edgent.test.providers.direct.DirectTestSetup;
-
-public class DirectMetricsEverywhereTest extends MetricsEverywhereTest implements DirectTestSetup {
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricsOffTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricsOffTest.java b/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricsOffTest.java
deleted file mode 100644
index e470d0a..0000000
--- a/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricsOffTest.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
-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 edgent.test.providers.direct.metrics;
-
-import edgent.test.metrics.MetricsOffTest;
-import edgent.test.providers.direct.DirectTestSetup;
-
-public class DirectMetricsOffTest extends MetricsOffTest implements DirectTestSetup {
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricsOnTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricsOnTest.java b/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricsOnTest.java
deleted file mode 100644
index 6ce76b8..0000000
--- a/providers/direct/src/test/java/edgent/test/providers/direct/metrics/DirectMetricsOnTest.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
-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 edgent.test.providers.direct.metrics;
-
-import edgent.test.metrics.MetricsOnTest;
-import edgent.test.providers.direct.DirectTestSetup;
-
-public class DirectMetricsOnTest extends MetricsOnTest implements DirectTestSetup {
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/edgent/test/providers/direct/metrics/package-info.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/edgent/test/providers/direct/metrics/package-info.java b/providers/direct/src/test/java/edgent/test/providers/direct/metrics/package-info.java
deleted file mode 100644
index 2870cdd..0000000
--- a/providers/direct/src/test/java/edgent/test/providers/direct/metrics/package-info.java
+++ /dev/null
@@ -1,19 +0,0 @@
-/*
-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 edgent.test.providers.direct.metrics;
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectPlumbingTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectPlumbingTest.java b/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectPlumbingTest.java
deleted file mode 100644
index 764ca17..0000000
--- a/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectPlumbingTest.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
-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 edgent.test.providers.direct.topology;
-
-import edgent.test.providers.direct.DirectTestSetup;
-import edgent.test.topology.PlumbingTest;
-
-public class DirectPlumbingTest extends PlumbingTest implements DirectTestSetup {
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectTStreamTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectTStreamTest.java b/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectTStreamTest.java
deleted file mode 100644
index 75cbfd2..0000000
--- a/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectTStreamTest.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
-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 edgent.test.providers.direct.topology;
-
-import edgent.test.providers.direct.DirectTestSetup;
-import edgent.test.topology.TStreamTest;
-
-public class DirectTStreamTest extends TStreamTest implements DirectTestSetup {
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectTopologyTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectTopologyTest.java b/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectTopologyTest.java
deleted file mode 100644
index 1e1d202..0000000
--- a/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectTopologyTest.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
-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 edgent.test.providers.direct.topology;
-
-import edgent.test.providers.direct.DirectTestSetup;
-import edgent.test.topology.TopologyTest;
-
-public class DirectTopologyTest extends TopologyTest implements DirectTestSetup {
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectWindowTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectWindowTest.java b/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectWindowTest.java
deleted file mode 100644
index 8901aee..0000000
--- a/providers/direct/src/test/java/edgent/test/providers/direct/topology/DirectWindowTest.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
-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 edgent.test.providers.direct.topology;
-
-import edgent.test.providers.direct.DirectTestSetup;
-import edgent.test.topology.TWindowTest;
-
-public class DirectWindowTest extends TWindowTest implements DirectTestSetup {
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/DirectJobTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/DirectJobTest.java b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/DirectJobTest.java
new file mode 100644
index 0000000..14f9704
--- /dev/null
+++ b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/DirectJobTest.java
@@ -0,0 +1,342 @@
+/*
+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.edgent.test.providers.direct;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.edgent.execution.Configs;
+import org.apache.edgent.execution.Job;
+import org.apache.edgent.graph.Vertex;
+import org.apache.edgent.oplet.Oplet;
+import org.apache.edgent.oplet.core.PeriodicSource;
+import org.apache.edgent.oplet.core.Pipe;
+import org.apache.edgent.providers.direct.DirectProvider;
+import org.apache.edgent.test.topology.TopologyAbstractTest;
+import org.apache.edgent.topology.TStream;
+import org.apache.edgent.topology.Topology;
+import org.junit.Test;
+
+import com.google.gson.JsonObject;
+
+public class DirectJobTest extends TopologyAbstractTest implements DirectTestSetup {
+ @Test
+ public void jobName0() throws Exception {
+ String[] data = new String[] {};
+ String topologyName = "topoName";
+ Topology t = newTopology(topologyName);
+ t.strings(data);
+ Job job = awaitCompleteExecution(t);
+ assertTrue(job.getName().startsWith(topologyName));
+ }
+
+ @Test
+ public void jobName1() throws Exception {
+ String[] data = new String[] {};
+ String topologyName = "topoName";
+ Topology t = newTopology(topologyName);
+ t.strings(data);
+ JsonObject config = new JsonObject();
+ config.addProperty(Configs.JOB_NAME, (String)null);
+ Job job = awaitCompleteExecution(t, config);
+ assertTrue(job.getName().startsWith(topologyName));
+ }
+
+ @Test
+ public void jobName2() throws Exception {
+ String[] data = new String[] {};
+ String jobName = "testJob";
+ Topology t = newTopology();
+ t.strings(data);
+ JsonObject config = new JsonObject();
+ config.addProperty(Configs.JOB_NAME, jobName);
+ Job job = awaitCompleteExecution(t, config);
+ assertEquals(jobName, job.getName());
+ }
+
+ @Test
+ public void jobDone0() throws Exception {
+ String[] data = new String[] {};
+ Topology t = newTopology();
+ @SuppressWarnings("unused")
+ TStream<String> s = t.strings(data);
+
+ Job job = awaitCompleteExecution(t);
+ assertEquals("job.getCurrentState() must be RUNNING", Job.State.RUNNING, job.getCurrentState());
+ assertEquals("job.getCurrentState() must be HEALTHY", Job.Health.HEALTHY, job.getHealth());
+ assertEquals("", job.getLastError());
+
+ job.stateChange(Job.Action.CLOSE);
+ assertEquals("job.getCurrentState() must be CLOSED", Job.State.CLOSED, job.getCurrentState());
+ assertEquals("job.getCurrentState() must be HEALTHY", Job.Health.HEALTHY, job.getHealth());
+ assertEquals("", job.getLastError());
+ }
+
+ @Test
+ public void jobDone1() throws Exception {
+ String[] data = new String[] {"a", "b", "c"};
+ Topology t = newTopology();
+ @SuppressWarnings("unused")
+ TStream<String> s = t.strings(data);
+
+ Job job = awaitCompleteExecution(t);
+ assertEquals("job.getCurrentState() must be RUNNING", Job.State.RUNNING, job.getCurrentState());
+ assertEquals("job.getCurrentState() must be HEALTHY", Job.Health.HEALTHY, job.getHealth());
+ assertEquals("", job.getLastError());
+ job.stateChange(Job.Action.CLOSE);
+ assertEquals("job.getCurrentState() must be CLOSED", Job.State.CLOSED, job.getCurrentState());
+ assertEquals("job.getCurrentState() must be HEALTHY", Job.Health.HEALTHY, job.getHealth());
+ assertEquals("", job.getLastError());
+ }
+
+ @Test
+ public void jobDone2() throws Exception {
+ final int NUM_TUPLES = 1000000;
+ Integer[] data = new Integer[NUM_TUPLES];
+ AtomicInteger numTuples = new AtomicInteger();
+
+ for (int i = 0; i < data.length; i++) {
+ data[i] = new Integer(i);
+ }
+ Topology t = newTopology();
+ TStream<Integer> ints = t.collection(Arrays.asList(data));
+ ints.sink(tuple -> numTuples.incrementAndGet());
+
+ Job job = awaitCompleteExecution(t);
+ Thread.sleep(1500); // wait for numTuples visibility
+ assertEquals(NUM_TUPLES, numTuples.get());
+ assertEquals("job.getCurrentState() must be RUNNING", Job.State.RUNNING, job.getCurrentState());
+ assertEquals("job.getCurrentState() must be HEALTHY", Job.Health.HEALTHY, job.getHealth());
+ assertEquals("", job.getLastError());
+ job.stateChange(Job.Action.CLOSE);
+ assertEquals("job.getCurrentState() must be CLOSED", Job.State.CLOSED, job.getCurrentState());
+ assertEquals("job.getCurrentState() must be HEALTHY", Job.Health.HEALTHY, job.getHealth());
+ assertEquals("", job.getLastError());
+ }
+
+ @Test
+ public void jobPeriodicSource() throws Exception {
+ Topology t = newTopology();
+ AtomicInteger n = new AtomicInteger(0);
+ @SuppressWarnings("unused")
+ TStream<Integer> ints = t.poll(() -> n.incrementAndGet(), 100, TimeUnit.MILLISECONDS);
+
+ Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t);
+ Job job = fj.get();
+ assertEquals(Job.State.RUNNING, job.getCurrentState());
+ Thread.sleep(TimeUnit.SECONDS.toMillis(2));
+ assertTrue(n.get() > 0); // At least one tuple was processed
+ job.stateChange(Job.Action.CLOSE);
+ assertEquals(Job.State.CLOSED, job.getCurrentState());
+ }
+
+ @Test
+ public void jobPeriodicSourceCancellation() throws Exception {
+ Topology t = newTopology();
+ AtomicInteger n = new AtomicInteger(0);
+ @SuppressWarnings("unused")
+ TStream<Integer> ints = t.poll(() -> n.incrementAndGet(), 500, TimeUnit.MILLISECONDS);
+
+ // Get the source oplet
+ Collection<Vertex<? extends Oplet<?, ?>, ?, ?>> vertices = t.graph().getVertices();
+ PeriodicSource<?> src = null;
+ for (Vertex<? extends Oplet<?, ?>, ?, ?> v : vertices) {
+ Oplet<?,?> op = v.getInstance();
+ assertTrue(op instanceof PeriodicSource);
+ src = (PeriodicSource<?>) op;
+ assertEquals(500, src.getPeriod());
+ assertSame(TimeUnit.MILLISECONDS, src.getUnit());
+ }
+
+ // Submit job
+ Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t);
+ Job job = fj.get();
+ assertEquals(Job.State.RUNNING, job.getCurrentState());
+ Thread.sleep(TimeUnit.SECONDS.toMillis(1));
+ int tupleCount = n.get();
+ assertTrue("Expected more tuples than "+ tupleCount, tupleCount > 0); // At least one tuple was processed
+ assertEquals(Job.Health.HEALTHY, job.getHealth());
+ assertEquals("", job.getLastError());
+
+ // Changing the period cancels the source's task and schedules new one
+ src.setPeriod(100);
+
+ Thread.sleep(TimeUnit.SECONDS.toMillis(1));
+
+ // More tuples processed after resetting the period
+ assertTrue("Expected more tuples than "+ n.get(), n.get() > 3*tupleCount);
+
+ job.stateChange(Job.Action.CLOSE);
+ assertEquals(Job.State.CLOSED, job.getCurrentState());
+ assertEquals(Job.Health.HEALTHY, job.getHealth());
+ assertEquals("", job.getLastError());
+ }
+
+ @Test
+ public void jobProcessSource() throws Exception {
+ Topology t = newTopology();
+ AtomicInteger n = new AtomicInteger(0);
+ @SuppressWarnings("unused")
+ TStream<Integer> ints = t.generate(() -> n.incrementAndGet());
+
+ Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t);
+ Job job = fj.get();
+ assertEquals(Job.State.RUNNING, job.getCurrentState());
+ assertEquals(Job.Health.HEALTHY, job.getHealth());
+ assertEquals("", job.getLastError());
+ Thread.sleep(TimeUnit.SECONDS.toMillis(2));
+ assertTrue(n.get() > 0); // At least one tuple was processed
+ job.stateChange(Job.Action.CLOSE);
+ assertEquals(Job.State.CLOSED, job.getCurrentState());
+ assertEquals(Job.Health.HEALTHY, job.getHealth());
+ assertEquals("", job.getLastError());
+ }
+
+ @Test(expected = TimeoutException.class)
+ public void jobTimesOut() throws Exception {
+ Topology t = newTopology();
+ AtomicInteger n = new AtomicInteger(0);
+ @SuppressWarnings("unused")
+ TStream<Integer> ints = t.poll(() -> n.incrementAndGet(), 100, TimeUnit.MILLISECONDS);
+
+ Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t);
+ Job job = fj.get();
+ assertEquals(Job.State.RUNNING, job.getCurrentState());
+ try {
+ job.complete(700, TimeUnit.MILLISECONDS);
+ } finally {
+ assertTrue(n.get() > 0); // At least one tuple was processed
+ assertEquals(Job.State.RUNNING, job.getCurrentState());
+ assertEquals(Job.Health.HEALTHY, job.getHealth());
+ assertEquals("", job.getLastError());
+ }
+ }
+
+ @Test(expected = ExecutionException.class)
+ public void jobPeriodicSourceError() throws Exception {
+ Topology t = newTopology();
+ AtomicInteger n = new AtomicInteger(0);
+ TStream<Integer> ints = t.poll(() -> n.incrementAndGet(), 100, TimeUnit.MILLISECONDS);
+ ints.pipe(new FailedOplet<Integer>(5, 0));
+
+ Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t);
+ Job job = fj.get();
+ assertEquals(Job.State.RUNNING, job.getCurrentState());
+ try {
+ job.complete(10, TimeUnit.SECONDS);
+ } finally {
+ // RUNNING even though execution error
+ assertEquals(Job.State.RUNNING, job.getCurrentState());
+ assertEquals(Job.Health.UNHEALTHY, job.getHealth());
+ assertEquals("java.lang.RuntimeException: Expected Test Exception", job.getLastError());
+ }
+ }
+
+ @Test(expected = ExecutionException.class)
+ public void jobProcessSourceError() throws Exception {
+ Topology t = newTopology();
+ AtomicInteger n = new AtomicInteger(0);
+ TStream<Integer> ints = t.generate(() -> n.incrementAndGet());
+ ints.pipe(new FailedOplet<Integer>(12, 100));
+
+ Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t);
+ Job job = fj.get();
+ assertEquals(Job.State.RUNNING, job.getCurrentState());
+ try {
+ job.complete(10, TimeUnit.SECONDS);
+ } finally {
+ // RUNNING even though execution error
+ assertEquals(Job.State.RUNNING, job.getCurrentState());
+ assertEquals(Job.Health.UNHEALTHY, job.getHealth());
+ assertEquals("java.lang.RuntimeException: Expected Test Exception", job.getLastError());
+ }
+ }
+
+ private Job awaitCompleteExecution(Topology t) throws InterruptedException, ExecutionException {
+ return awaitCompleteExecution(t, null);
+ }
+
+ private Job awaitCompleteExecution(Topology t, JsonObject config) throws InterruptedException, ExecutionException {
+ Future<Job> fj = ((DirectProvider)getTopologyProvider()).submit(t, config);
+ Job job = fj.get();
+ job.complete();
+ return job;
+ }
+
+ /**
+ * Test oplet which fails after receiving a given number of tuples.
+ * @param <T>
+ */
+ @SuppressWarnings("serial")
+ private static class FailedOplet<T> extends Pipe<T,T> {
+ private final int threshold;
+ private final int sleepMillis;
+
+ /**
+ * Create test oplet.
+ *
+ * @param afterTuples number of tuples to receive before failing
+ * @param sleepMillis milliseconds of sleep before processing each tuple
+ */
+ public FailedOplet(int afterTuples, int sleepMillis) {
+ if (afterTuples < 0)
+ throw new IllegalArgumentException("afterTuples="+afterTuples);
+ if (sleepMillis < 0)
+ throw new IllegalArgumentException("sleepMillis="+sleepMillis);
+ this.threshold = afterTuples;
+ this.sleepMillis = sleepMillis;
+ }
+
+ @Override
+ public void close() throws Exception {
+ }
+ @Override
+ public void accept(T tuple) {
+ if (sleepMillis > 0) {
+ sleep(sleepMillis);
+ }
+ injectError(threshold);
+ submit(tuple);
+ }
+
+ private AtomicInteger count = new AtomicInteger(0);
+ protected void injectError(int errorAt) {
+ if (count.getAndIncrement() == errorAt)
+ throw new RuntimeException("Expected Test Exception");
+ }
+
+ protected static void sleep(long millis) {
+ try {
+ Thread.sleep(millis);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/DirectTestSetup.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/DirectTestSetup.java b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/DirectTestSetup.java
new file mode 100644
index 0000000..c06ddd1
--- /dev/null
+++ b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/DirectTestSetup.java
@@ -0,0 +1,38 @@
+/*
+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.edgent.test.providers.direct;
+
+import org.apache.edgent.execution.Job;
+import org.apache.edgent.execution.Submitter;
+import org.apache.edgent.providers.direct.DirectProvider;
+import org.apache.edgent.test.topology.TopologyTestSetup;
+import org.apache.edgent.topology.Topology;
+
+public interface DirectTestSetup extends TopologyTestSetup {
+ @Override
+ default DirectProvider createTopologyProvider() {
+ return new DirectProvider();
+ }
+
+ @Override
+ default Submitter<Topology, Job> createSubmitter() {
+ return (DirectProvider) getTopologyProvider();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/appservice/AppServiceTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/appservice/AppServiceTest.java b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/appservice/AppServiceTest.java
new file mode 100644
index 0000000..e246983
--- /dev/null
+++ b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/appservice/AppServiceTest.java
@@ -0,0 +1,72 @@
+/*
+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.edgent.test.providers.direct.appservice;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.net.URL;
+
+import org.apache.edgent.providers.direct.DirectProvider;
+import org.apache.edgent.runtime.appservice.AppService;
+import org.apache.edgent.topology.services.ApplicationService;
+import org.junit.Test;
+
+public class AppServiceTest {
+
+
+ @Test
+ public void testRegisterTopology() {
+ DirectProvider direct = new DirectProvider();
+ ApplicationService appService = AppService.createAndRegister(direct, direct);
+
+ assertTrue(appService.getApplicationNames().isEmpty());
+
+ appService.registerTopology("FirstApp", (t,c) -> t.strings("a"));
+ assertEquals(1, appService.getApplicationNames().size());
+ assertTrue(appService.getApplicationNames().contains("FirstApp"));
+
+ appService.registerTopology("SecondApp", (t,c) -> t.strings("b"));
+ assertEquals(2, appService.getApplicationNames().size());
+ assertTrue(appService.getApplicationNames().contains("FirstApp"));
+ assertTrue(appService.getApplicationNames().contains("SecondApp"));
+ }
+
+ @Test
+ public void testRegisterJar() throws Exception {
+ DirectProvider direct = new DirectProvider();
+ ApplicationService appService = AppService.createAndRegister(direct, direct);
+
+ String qd = System.getProperty("edgent.test.root.dir");
+ assertNotNull(qd);
+ File testAppsJar = new File(qd, "api/topology/test.classes/edgent.api.topology.APPS.TEST.jar");
+ assertTrue(testAppsJar.exists());
+
+ URL testAppsJarURL = testAppsJar.toURI().toURL();
+
+ appService.registerJar(testAppsJarURL.toExternalForm(), null);
+
+ assertEquals(3, appService.getApplicationNames().size());
+ assertTrue(appService.getApplicationNames().contains("FirstJarApp"));
+ assertTrue(appService.getApplicationNames().contains("SecondJarApp"));
+ assertTrue(appService.getApplicationNames().contains("ThirdJarApp"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricObjectNameFactoryTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricObjectNameFactoryTest.java b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricObjectNameFactoryTest.java
new file mode 100644
index 0000000..0f77091
--- /dev/null
+++ b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricObjectNameFactoryTest.java
@@ -0,0 +1,25 @@
+/*
+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.edgent.test.providers.direct.metrics;
+
+import org.apache.edgent.test.metrics.MetricObjectNameFactoryTest;
+import org.apache.edgent.test.providers.direct.DirectTestSetup;
+
+public class DirectMetricObjectNameFactoryTest extends MetricObjectNameFactoryTest implements DirectTestSetup {
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricsEverywhereTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricsEverywhereTest.java b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricsEverywhereTest.java
new file mode 100644
index 0000000..c7ece97
--- /dev/null
+++ b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricsEverywhereTest.java
@@ -0,0 +1,25 @@
+/*
+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.edgent.test.providers.direct.metrics;
+
+import org.apache.edgent.test.metrics.MetricsEverywhereTest;
+import org.apache.edgent.test.providers.direct.DirectTestSetup;
+
+public class DirectMetricsEverywhereTest extends MetricsEverywhereTest implements DirectTestSetup {
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricsOffTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricsOffTest.java b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricsOffTest.java
new file mode 100644
index 0000000..adcf48f
--- /dev/null
+++ b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricsOffTest.java
@@ -0,0 +1,25 @@
+/*
+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.edgent.test.providers.direct.metrics;
+
+import org.apache.edgent.test.metrics.MetricsOffTest;
+import org.apache.edgent.test.providers.direct.DirectTestSetup;
+
+public class DirectMetricsOffTest extends MetricsOffTest implements DirectTestSetup {
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricsOnTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricsOnTest.java b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricsOnTest.java
new file mode 100644
index 0000000..b592746
--- /dev/null
+++ b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/DirectMetricsOnTest.java
@@ -0,0 +1,25 @@
+/*
+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.edgent.test.providers.direct.metrics;
+
+import org.apache.edgent.test.metrics.MetricsOnTest;
+import org.apache.edgent.test.providers.direct.DirectTestSetup;
+
+public class DirectMetricsOnTest extends MetricsOnTest implements DirectTestSetup {
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/package-info.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/package-info.java b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/package-info.java
new file mode 100644
index 0000000..d97e17d
--- /dev/null
+++ b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/metrics/package-info.java
@@ -0,0 +1,19 @@
+/*
+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.edgent.test.providers.direct.metrics;
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/topology/DirectPlumbingTest.java
----------------------------------------------------------------------
diff --git a/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/topology/DirectPlumbingTest.java b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/topology/DirectPlumbingTest.java
new file mode 100644
index 0000000..6f74690
--- /dev/null
+++ b/providers/direct/src/test/java/org/apache/edgent/test/providers/direct/topology/DirectPlumbingTest.java
@@ -0,0 +1,25 @@
+/*
+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.edgent.test.providers.direct.topology;
+
+import org.apache.edgent.test.providers.direct.DirectTestSetup;
+import org.apache.edgent.test.topology.PlumbingTest;
+
+public class DirectPlumbingTest extends PlumbingTest implements DirectTestSetup {
+}