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:44 UTC
[43/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/api/oplet/src/main/java/org/apache/edgent/oplet/core/Split.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/core/Split.java b/api/oplet/src/main/java/org/apache/edgent/oplet/core/Split.java
new file mode 100644
index 0000000..9c21549
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/core/Split.java
@@ -0,0 +1,84 @@
+/*
+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.oplet.core;
+
+import static org.apache.edgent.function.Functions.closeFunction;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.edgent.function.Consumer;
+import org.apache.edgent.function.ToIntFunction;
+import org.apache.edgent.oplet.OpletContext;
+
+/**
+ * Split a stream into multiple streams depending
+ * on the result of a splitter function.
+ * <BR>
+ * For each tuple a function is called:
+ * <UL>
+ * <LI>If the return is negative the tuple is dropped.</LI>
+ * <LI>Otherwise the return value is modded by the number of
+ * output ports and the result is the output port index
+ * the tuple is submitted to.</LI>
+ * </UL>
+ *
+ * @param <T> Type of the tuple.
+ */
+public class Split<T> extends AbstractOplet<T, T> implements Consumer<T> {
+
+ private static final long serialVersionUID = 1L;
+ private final ToIntFunction<T> splitter;
+ private List<? extends Consumer<T>> destinations;
+ private int n;
+
+ public Split(ToIntFunction<T> splitter) {
+ this.splitter = splitter;
+ }
+
+
+ @Override
+ public void initialize(OpletContext<T, T> context) {
+ super.initialize(context);
+
+ destinations = context.getOutputs();
+ n = destinations.size();
+ }
+
+ @Override
+ public void start() {
+ }
+
+ @Override
+ public List<Consumer<T>> getInputs() {
+ return Collections.singletonList(this);
+ }
+
+ @Override
+ public void accept(T tuple) {
+ int s = splitter.applyAsInt(tuple);
+ if (s >= 0)
+ destinations.get(s % n).accept(tuple);
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeFunction(splitter);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/core/Union.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/core/Union.java b/api/oplet/src/main/java/org/apache/edgent/oplet/core/Union.java
new file mode 100644
index 0000000..580b627
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/core/Union.java
@@ -0,0 +1,50 @@
+/*
+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.oplet.core;
+
+import org.apache.edgent.oplet.OpletContext;
+
+/**
+ * Union oplet, merges multiple input ports
+ * into a single output port.
+ *
+ * Processing for each input is identical
+ * and just submits the tuple to the single output.
+ *
+ * @param <T> Tuple type
+ */
+public final class Union<T> extends FanIn<T, T> {
+
+ @Override
+ public void start() {
+ }
+
+
+ @Override
+ public void initialize(OpletContext<T, T> context) {
+ super.initialize(context);
+
+ // forward every tuple to the output port
+ setReceiver((tuple, iportIndex) -> tuple);
+ }
+
+ @Override
+ public void close() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/core/mbeans/package-info.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/core/mbeans/package-info.java b/api/oplet/src/main/java/org/apache/edgent/oplet/core/mbeans/package-info.java
new file mode 100644
index 0000000..04fa1e6
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/core/mbeans/package-info.java
@@ -0,0 +1,22 @@
+/*
+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.
+*/
+/**
+ * Management beans for core oplets.
+ */
+package org.apache.edgent.oplet.core.mbeans;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/core/package-info.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/core/package-info.java b/api/oplet/src/main/java/org/apache/edgent/oplet/core/package-info.java
new file mode 100644
index 0000000..5e59bdd
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/core/package-info.java
@@ -0,0 +1,22 @@
+/*
+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.
+*/
+/**
+ * Core primitive oplets.
+ */
+package org.apache.edgent.oplet.core;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Events.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Events.java b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Events.java
new file mode 100644
index 0000000..ad789a5
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Events.java
@@ -0,0 +1,80 @@
+/*
+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.oplet.functional;
+
+
+import static org.apache.edgent.function.Functions.closeFunction;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+
+import org.apache.edgent.function.Consumer;
+import org.apache.edgent.oplet.core.Source;
+
+/**
+ * Generate tuples from events.
+ * This oplet implements {@link Consumer} which
+ * can be called directly from an event handler,
+ * listener or callback.
+ *
+ * @param <T> Data container type for output tuples.
+ */
+public class Events<T> extends Source<T>implements Consumer<T> {
+
+ private static final long serialVersionUID = 1L;
+ private Consumer<Consumer<T>> eventSetup;
+
+ public Events(Consumer<Consumer<T>> eventSetup) {
+ this.eventSetup = eventSetup;
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeFunction(eventSetup);
+ }
+
+ @Override
+ public void start() {
+ // Allocate a thread so the job containing this oplet
+ // doesn't look "complete" and shutdown.
+ Thread endlessEventSource = getOpletContext()
+ .getService(ThreadFactory.class)
+ .newThread(() -> {
+ try {
+ Thread.sleep(Long.MAX_VALUE);
+ }
+ catch (InterruptedException e) {
+ // cancelled; we're done.
+ }
+ });
+ endlessEventSource.setDaemon(false);
+ endlessEventSource.start();
+
+ // It's possible for uses to do things like a blocking connect
+ // to an external system from eventSetup.accept() so run it as
+ // a task to avoid start() / submit-job timeouts.
+ getOpletContext().getService(ScheduledExecutorService.class)
+ .submit(() -> eventSetup.accept(this));
+ }
+
+ @Override
+ public void accept(T tuple) {
+ submit(tuple);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Filter.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Filter.java b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Filter.java
new file mode 100644
index 0000000..ee7997a
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Filter.java
@@ -0,0 +1,44 @@
+/*
+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.oplet.functional;
+
+import static org.apache.edgent.function.Functions.closeFunction;
+
+import org.apache.edgent.function.Predicate;
+import org.apache.edgent.oplet.core.Pipe;
+
+public class Filter<T> extends Pipe<T, T> {
+ private static final long serialVersionUID = 1L;
+ private Predicate<T> filter;
+
+ public Filter(Predicate<T> filter) {
+ this.filter = filter;
+ }
+
+ @Override
+ public void accept(T tuple) {
+ if (filter.test(tuple))
+ submit(tuple);
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeFunction(filter);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/functional/FlatMap.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/functional/FlatMap.java b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/FlatMap.java
new file mode 100644
index 0000000..21ff0b3
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/FlatMap.java
@@ -0,0 +1,66 @@
+/*
+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.oplet.functional;
+
+import static org.apache.edgent.function.Functions.closeFunction;
+
+import org.apache.edgent.function.Function;
+import org.apache.edgent.oplet.core.Pipe;
+
+/**
+ *
+ * Map an input tuple to 0-N output tuples.
+ *
+ * Uses a function that returns an iterable
+ * to map the input tuple. The return value
+ * of the function's apply method is
+ * iterated through with each returned
+ * value being submitted as an output tuple.
+ *
+ *
+ * @param <I>
+ * Data container type for input tuples.
+ * @param <O>
+ * Data container type for output tuples.
+ */
+public class FlatMap<I, O> extends Pipe<I, O> {
+ private static final long serialVersionUID = 1L;
+
+ private Function<I, Iterable<O>> function;
+
+ public FlatMap(Function<I, Iterable<O>> function) {
+ this.function = function;
+ }
+
+ @Override
+ public void accept(I tuple) {
+ Iterable<O> outputs = function.apply(tuple);
+ if (outputs != null) {
+ for (O output : outputs) {
+ if (output != null)
+ submit(output);
+ }
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeFunction(function);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Map.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Map.java b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Map.java
new file mode 100644
index 0000000..11cb73b
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Map.java
@@ -0,0 +1,54 @@
+/*
+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.oplet.functional;
+
+import static org.apache.edgent.function.Functions.closeFunction;
+
+import org.apache.edgent.function.Function;
+import org.apache.edgent.oplet.core.Pipe;
+
+/**
+ * Map an input tuple to 0-1 output tuple
+ *
+ *
+ * @param <I>
+ * Data container type for input tuples.
+ * @param <O>
+ * Data container type for output tuples.
+ */
+public class Map<I, O> extends Pipe<I, O> {
+ private static final long serialVersionUID = 1L;
+ private Function<I, O> function;
+
+ public Map(Function<I, O> function) {
+ this.function = function;
+ }
+
+ @Override
+ public void accept(I tuple) {
+ O output = function.apply(tuple);
+ if (output != null)
+ submit(output);
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeFunction(function);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Peek.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Peek.java b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Peek.java
new file mode 100644
index 0000000..26bfaa9
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/Peek.java
@@ -0,0 +1,58 @@
+/*
+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.oplet.functional;
+
+
+import static org.apache.edgent.function.Functions.closeFunction;
+
+import org.apache.edgent.function.Consumer;
+
+/**
+ * Functional peek oplet.
+ *
+ * Each peek calls {@code peeker.accept(tuple)}.
+ *
+ * @param <T> Tuple type.
+ */
+public class Peek<T> extends org.apache.edgent.oplet.core.Peek<T> {
+ private static final long serialVersionUID = 1L;
+ private final Consumer<T> peeker;
+
+ /**
+ * Peek oplet using a function to peek.
+ * @param peeker Function that peeks at the tuple.
+ */
+ public Peek(Consumer<T> peeker) {
+ this.peeker = peeker;
+ }
+
+ protected Consumer<T> getPeeker() {
+ return peeker;
+ }
+
+ @Override
+ protected void peek(T tuple) {
+ peeker.accept(tuple);
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeFunction(peeker);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/functional/SupplierPeriodicSource.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/functional/SupplierPeriodicSource.java b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/SupplierPeriodicSource.java
new file mode 100644
index 0000000..82fb13c
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/SupplierPeriodicSource.java
@@ -0,0 +1,54 @@
+/*
+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.oplet.functional;
+
+import static org.apache.edgent.function.Functions.closeFunction;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.edgent.function.Supplier;
+import org.apache.edgent.oplet.OpletContext;
+import org.apache.edgent.oplet.core.PeriodicSource;
+
+public class SupplierPeriodicSource<T> extends PeriodicSource<T> {
+
+ private Supplier<T> data;
+
+ public SupplierPeriodicSource(long period, TimeUnit unit, Supplier<T> data) {
+ super(period, unit);
+ this.data = data;
+ }
+
+ @Override
+ public void initialize(OpletContext<Void, T> context) {
+ super.initialize(context);
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeFunction(data);
+ }
+
+ @Override
+ public void fetchTuples() {
+ T tuple = data.get();
+ if (tuple != null)
+ submit(tuple);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/functional/SupplierSource.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/functional/SupplierSource.java b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/SupplierSource.java
new file mode 100644
index 0000000..15f473e
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/SupplierSource.java
@@ -0,0 +1,58 @@
+/*
+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.oplet.functional;
+
+import static org.apache.edgent.function.Functions.closeFunction;
+
+import org.apache.edgent.function.Supplier;
+import org.apache.edgent.oplet.OpletContext;
+import org.apache.edgent.oplet.core.ProcessSource;
+
+public class SupplierSource<T> extends ProcessSource<T> {
+
+ private Supplier<Iterable<T>> data;
+
+ public SupplierSource() {
+ }
+
+ public SupplierSource(Supplier<Iterable<T>> data) {
+ this.data = data;
+ }
+
+ @Override
+ public void initialize(OpletContext<Void, T> context) {
+ super.initialize(context);
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeFunction(data);
+ }
+
+ @Override
+ public void process() {
+ for (T tuple : data.get()) {
+ if (tuple != null)
+ submit(tuple);
+
+ if (Thread.currentThread().isInterrupted())
+ break;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/functional/package-info.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/functional/package-info.java b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/package-info.java
new file mode 100644
index 0000000..a0724d8
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/functional/package-info.java
@@ -0,0 +1,22 @@
+/*
+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.
+*/
+/**
+ * Oplets that process tuples using functions.
+ */
+package org.apache.edgent.oplet.functional;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/package-info.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/package-info.java b/api/oplet/src/main/java/org/apache/edgent/oplet/package-info.java
new file mode 100644
index 0000000..bc13fa3
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/package-info.java
@@ -0,0 +1,28 @@
+/*
+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.
+*/
+/**
+ * Oplets API.
+ * <P>
+ * An oplet is a stream processor that can have 0-N input ports and 0-M output ports.
+ * Tuples on streams connected to an oplet's input port are delivered to the oplet for processing.
+ * The oplet submits tuples to its output ports which results in the tuples
+ * being present on the connected streams.
+ * </P>
+ */
+package org.apache.edgent.oplet;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/Barrier.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/Barrier.java b/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/Barrier.java
new file mode 100644
index 0000000..f134ca4
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/Barrier.java
@@ -0,0 +1,113 @@
+/*
+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.oplet.plumbing;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+
+import org.apache.edgent.function.BiFunction;
+import org.apache.edgent.oplet.OpletContext;
+import org.apache.edgent.oplet.core.FanIn;
+
+/**
+ * A tuple synchronization barrier.
+ * <P>
+ * {@code Barrier} has n input ports with tuple type {@code T}
+ * and one output port with tuple type {@code List<T>}.
+ * Once the oplet receives one tuple on each of its input ports,
+ * it generates an output tuple containing one tuple from each input port.
+ * It then awaits receiving the next collection of tuples.
+ * Input port 0's tuple is in list[0], port 1's tuple in list[1], and so on.
+ * </P><P>
+ * Each input port has an associated queue of size {@code queueCapacity}.
+ * The input port's {@code Consumer<T>.accept()} will block if it's queue is full.
+ * </P>
+ *
+ * @param <T> Type of the tuple.
+ */
+public class Barrier<T> extends FanIn<T, List<T>> {
+
+ private final int queueCapacity;
+ private Thread thread;
+ private List<LinkedBlockingQueue<T>> iportQueues;
+
+ /**
+ * Create a new instance.
+ * @param queueCapacity size of each input port's blocking queue
+ */
+ public Barrier(int queueCapacity) {
+ this.queueCapacity = queueCapacity;
+ }
+
+ @Override
+ public void initialize(OpletContext<T, List<T>> context) {
+ super.initialize(context);
+
+ thread = context.getService(ThreadFactory.class).newThread(() -> run());
+
+ int numIports = getOpletContext().getInputCount();
+ iportQueues = new ArrayList<>(numIports);
+ for (int i = 0; i < numIports; i++)
+ iportQueues.add(new LinkedBlockingQueue<>(queueCapacity));
+
+ setReceiver(receiver());
+ }
+
+ @Override
+ public void start() {
+ thread.start();
+ }
+
+ protected BiFunction<T,Integer,List<T>> receiver() {
+ return (tuple, iportIndex) -> {
+ accept(tuple, iportIndex);
+ return null;
+ };
+ }
+
+ protected void accept(T tuple, int iportIndex) {
+ try {
+ iportQueues.get(iportIndex).put(tuple);
+ } catch(InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void run() {
+ while (!Thread.interrupted()) {
+ try {
+ List<T> list = new ArrayList<>(iportQueues.size());
+ for (LinkedBlockingQueue<T> iport : iportQueues) {
+ list.add(iport.take());
+ }
+ submit(list);
+ } catch (InterruptedException e) {
+ break;
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/Isolate.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/Isolate.java b/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/Isolate.java
new file mode 100644
index 0000000..45ab0c7
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/Isolate.java
@@ -0,0 +1,97 @@
+/*
+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.oplet.plumbing;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+
+import org.apache.edgent.oplet.OpletContext;
+import org.apache.edgent.oplet.core.Pipe;
+
+/**
+ * Isolate upstream processing from downstream
+ * processing guaranteeing tuple order.
+ * Input tuples are placed at the tail of a queue
+ * and dedicated thread removes them from the
+ * head and is used for downstream processing.
+ *
+ * @param <T> Type of the tuple.
+ */
+public class Isolate<T> extends Pipe<T,T> {
+ private static final long serialVersionUID = 1L;
+
+ private Thread thread;
+ private final LinkedBlockingQueue<T> tuples;
+
+ /**
+ * Create a new Isolate oplet.
+ * <BR>
+ * Same as Isolate(Integer.MAX_VALUE).
+ */
+ public Isolate() {
+ this(Integer.MAX_VALUE);
+ }
+
+ /**
+ * Create a new Isolate oplet.
+ * @param queueCapacity size of the queue between the input stream
+ * and the output stream.
+ * {@link #accept(Object) accept} blocks when the queue is full.
+ */
+ public Isolate(int queueCapacity) {
+ tuples = new LinkedBlockingQueue<>(queueCapacity);
+ }
+
+ @Override
+ public void initialize(OpletContext<T, T> context) {
+ super.initialize(context);
+ thread = context.getService(ThreadFactory.class).newThread(() -> run());
+ }
+
+ @Override
+ public void start() {
+ super.start();
+ thread.start();
+ }
+
+ @Override
+ public void accept(T tuple) {
+ try {
+ tuples.put(tuple);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void run() {
+ while (!Thread.interrupted()) {
+ try {
+ submit(tuples.take());
+ } catch (InterruptedException e) {
+ break;
+ }
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/PressureReliever.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/PressureReliever.java b/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/PressureReliever.java
new file mode 100644
index 0000000..fb9bc61
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/PressureReliever.java
@@ -0,0 +1,167 @@
+/*
+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.oplet.plumbing;
+
+import java.util.LinkedList;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.edgent.function.BiConsumer;
+import org.apache.edgent.function.Function;
+import org.apache.edgent.function.Functions;
+import org.apache.edgent.oplet.OpletContext;
+import org.apache.edgent.oplet.core.Pipe;
+import org.apache.edgent.window.Partition;
+import org.apache.edgent.window.PartitionedState;
+import org.apache.edgent.window.Policies;
+import org.apache.edgent.window.Window;
+import org.apache.edgent.window.Windows;
+
+/**
+ * Relieve pressure on upstream oplets by discarding tuples.
+ * This oplet ensures that upstream processing is not
+ * constrained by any delay in downstream processing,
+ * for example by a sink oplet not being able to connect
+ * to its external system.
+ * When downstream processing cannot keep up with the input rate
+ * this oplet maintains a defined window of the most recent
+ * tuples and discards any earlier tuples using arrival order.
+ * <P>
+ * A window partition is maintained for each key seen
+ * on the input stream. Any tuple arriving on the input
+ * stream is inserted into the window. Asynchronously
+ * tuples are taken from the window using FIFO and
+ * submitted downstream. The submission of tuples maintains
+ * order within a partition but not across partitions.
+ * </P>
+ * <P>
+ * Tuples are <B>discarded and not</B> submitted to the
+ * output port if the downstream processing cannot keep up
+ * the incoming tuple rate.
+ * </P>
+ * <UL>
+ * <LI>For a {@link #PressureReliever(int, Function) count}
+ * {@code PressureReliever} up to last (most recent) {@code N} tuples
+ * are maintained in a window partition.
+ * <BR> Asynchronous tuple submission removes the last (oldest) tuple in the partition
+ * before submitting it downstream.
+ * <BR> If when an input tuple is processed the window partition contains N tuples, then
+ * the first (oldest) tuple in the partition is discarded before the input tuple is inserted into the window.
+ * </UL>
+ * <P>
+ * Insertion of the oplet into a stream disconnects the
+ * upstream processing from the downstream processing,
+ * so that downstream processing is executed on a different
+ * thread to the thread that processed the input tuple.
+ * </P>
+ *
+ * @param <T> Tuple type.
+ * @param <K> Key type.
+ */
+public class PressureReliever<T, K> extends Pipe<T, T> {
+ private static final long serialVersionUID = 1L;
+
+ private ScheduledExecutorService executor;
+ private final Window<T, K, LinkedList<T>> window;
+
+ /**
+ * Pressure reliever that maintains up to {@code count} most recent tuples per key.
+ *
+ * @param count Number of tuples to maintain where downstream processing cannot keep up.
+ * @param keyFunction Key function for tuples.
+ */
+ public PressureReliever(int count, Function<T, K> keyFunction) {
+ window = Windows.window(
+ Policies.alwaysInsert(),
+ Policies.countContentsPolicy(count),
+ Policies.evictOldest(),
+ new FirstSubmitter(),
+ keyFunction,
+ () -> new LinkedList<T>());
+
+ // No processing of the window takes place
+ window.registerPartitionProcessor((tuples, k) -> { });
+ }
+
+ @Override
+ public void initialize(OpletContext<T, T> context) {
+ super.initialize(context);
+ executor = context.getService(ScheduledExecutorService.class);
+ }
+
+ @Override
+ public void accept(T tuple) {
+ window.insert(tuple);
+ }
+
+ @Override
+ public void close() throws Exception {
+ }
+
+ private class FirstSubmitter extends PartitionedState<K, AtomicBoolean>
+ implements BiConsumer<Partition<T, K, LinkedList<T>>, T> {
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ FirstSubmitter() {
+ super(() -> new AtomicBoolean());
+ }
+
+ /**
+ * Process the window (to consume the oldest tuple in the partition)
+ * only if a tuple from this partition is not already being consumed.
+ *
+ * @param t
+ * @param v
+ */
+ @Override
+ public void accept(Partition<T, K, LinkedList<T>> partition, T tuple) {
+ submitNextTuple(partition);
+ }
+
+ private void submitNextTuple(Partition<T, K, LinkedList<T>> partition) {
+ final K key = partition.getKey();
+ final AtomicBoolean latch = getState(key);
+ if (!latch.compareAndSet(false, true))
+ return;
+
+ final T firstTuple;
+ synchronized (partition) {
+ final LinkedList<T> contents = partition.getContents();
+ if (contents.isEmpty()) {
+ latch.set(false);
+ return;
+ }
+
+ firstTuple = contents.removeFirst();
+ }
+
+ Runnable submit = Functions.delayedConsume(getDestination(), firstTuple);
+ submit = Functions.runWithFinal(submit, () -> {
+ latch.set(false);
+ submitNextTuple(partition);
+ });
+
+ executor.execute(submit);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/UnorderedIsolate.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/UnorderedIsolate.java b/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/UnorderedIsolate.java
new file mode 100644
index 0000000..5108301
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/UnorderedIsolate.java
@@ -0,0 +1,55 @@
+/*
+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.oplet.plumbing;
+
+import java.util.concurrent.ScheduledExecutorService;
+
+import org.apache.edgent.function.Functions;
+import org.apache.edgent.oplet.OpletContext;
+import org.apache.edgent.oplet.core.Pipe;
+
+/**
+ * Isolate upstream processing from downstream
+ * processing without guaranteeing tuple order.
+ * An executor is used for downstream processing
+ * thus tuple order cannot be guaranteed as the
+ * scheduler does not guarantee execution order.
+ *
+ * @param <T> Type of the tuple.
+ */
+public class UnorderedIsolate<T> extends Pipe<T,T> {
+ private static final long serialVersionUID = 1L;
+
+ private ScheduledExecutorService executor;
+
+ @Override
+ public void initialize(OpletContext<T, T> context) {
+ super.initialize(context);
+ executor = context.getService(ScheduledExecutorService.class);
+ }
+
+ @Override
+ public void accept(T tuple) {
+ executor.execute(Functions.delayedConsume(getDestination(), tuple));
+ }
+
+ @Override
+ public void close() throws Exception {
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/package-info.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/package-info.java b/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/package-info.java
new file mode 100644
index 0000000..ee11cbd
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/plumbing/package-info.java
@@ -0,0 +1,22 @@
+/*
+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.
+*/
+/**
+ * Oplets that control the flow of tuples.
+ */
+package org.apache.edgent.oplet.plumbing;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/window/Aggregate.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/window/Aggregate.java b/api/oplet/src/main/java/org/apache/edgent/oplet/window/Aggregate.java
new file mode 100644
index 0000000..0e717c5
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/window/Aggregate.java
@@ -0,0 +1,81 @@
+/*
+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.oplet.window;
+
+import static org.apache.edgent.function.Functions.closeFunction;
+
+import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
+
+import org.apache.edgent.function.BiConsumer;
+import org.apache.edgent.function.BiFunction;
+import org.apache.edgent.oplet.OpletContext;
+import org.apache.edgent.oplet.core.Pipe;
+import org.apache.edgent.window.Window;
+
+/**
+ * Aggregate a window.
+ * Window contents are aggregated by a
+ * {@link BiFunction aggregator function}
+ * passing the list of tuples in the window and
+ * the partition key. The returned value
+ * is submitted to the sole output port
+ * if it is not {@code null}.
+ *
+ * @param <T> Type of the input tuples.
+ * @param <U> Type of the output tuples.
+ * @param <K> Type of the partition key.
+ */
+public class Aggregate<T,U,K> extends Pipe<T, U> {
+ private static final long serialVersionUID = 1L;
+ private final Window<T,K, ? extends List<T>> window;
+ /**
+ * The aggregator provided by the user.
+ */
+ private final BiFunction<List<T>,K, U> aggregator;
+
+ public Aggregate(Window<T,K, ? extends List<T>> window, BiFunction<List<T>,K, U> aggregator){
+ this.aggregator = aggregator;
+ BiConsumer<List<T>, K> partProcessor = (tuples, key) -> {
+ U aggregateTuple = aggregator.apply(tuples, key);
+ if (aggregateTuple != null)
+ submit(aggregateTuple);
+ };
+
+ window.registerPartitionProcessor(partProcessor);
+ this.window=window;
+ }
+
+ @Override
+ public void initialize(OpletContext<T,U> context) {
+ super.initialize(context);
+ window.registerScheduledExecutorService(this.getOpletContext().getService(ScheduledExecutorService.class));
+ }
+
+ @Override
+ public void accept(T tuple) {
+ window.insert(tuple);
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeFunction(aggregator);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/main/java/org/apache/edgent/oplet/window/package-info.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/org/apache/edgent/oplet/window/package-info.java b/api/oplet/src/main/java/org/apache/edgent/oplet/window/package-info.java
new file mode 100644
index 0000000..e2a6a38
--- /dev/null
+++ b/api/oplet/src/main/java/org/apache/edgent/oplet/window/package-info.java
@@ -0,0 +1,22 @@
+/*
+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.
+*/
+/**
+ * Oplets using windows.
+ */
+package org.apache.edgent.oplet.window;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/test/java/edgent/test/oplet/functional/FilterTest.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/test/java/edgent/test/oplet/functional/FilterTest.java b/api/oplet/src/test/java/edgent/test/oplet/functional/FilterTest.java
deleted file mode 100644
index 8de6841..0000000
--- a/api/oplet/src/test/java/edgent/test/oplet/functional/FilterTest.java
+++ /dev/null
@@ -1,34 +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.oplet.functional;
-
-import static org.junit.Assert.assertTrue;
-
-import org.junit.Test;
-
-import edgent.oplet.core.Pipe;
-import edgent.oplet.functional.Filter;
-
-public class FilterTest {
-
- @Test
- public void testHierachy() {
- assertTrue(Pipe.class.isAssignableFrom(Filter.class));
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/test/java/edgent/test/oplet/operators/PipeTest.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/test/java/edgent/test/oplet/operators/PipeTest.java b/api/oplet/src/test/java/edgent/test/oplet/operators/PipeTest.java
deleted file mode 100644
index bb617ad..0000000
--- a/api/oplet/src/test/java/edgent/test/oplet/operators/PipeTest.java
+++ /dev/null
@@ -1,60 +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.oplet.operators;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertTrue;
-
-import java.util.List;
-
-import org.junit.Test;
-
-import edgent.function.Consumer;
-import edgent.oplet.Oplet;
-import edgent.oplet.core.AbstractOplet;
-import edgent.oplet.core.Pipe;
-
-public class PipeTest {
-
- @Test
- public void testHierachy() {
- assertTrue(Oplet.class.isAssignableFrom(Pipe.class));
- assertTrue(AbstractOplet.class.isAssignableFrom(Pipe.class));
- }
-
- @Test
- public void testDestination() {
- Pipe<String, Integer> pipe = new Pipe<String, Integer>() {
- private static final long serialVersionUID = 1L;
-
- @Override
- public void accept(String tuple) {
- }
-
- @Override
- public void close() throws Exception {
- };
- };
-
- List<? extends Consumer<String>> inputs = pipe.getInputs();
- assertEquals(1, inputs.size());
- assertSame(pipe, inputs.get(0));
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/test/java/org/apache/edgent/test/oplet/functional/FilterTest.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/test/java/org/apache/edgent/test/oplet/functional/FilterTest.java b/api/oplet/src/test/java/org/apache/edgent/test/oplet/functional/FilterTest.java
new file mode 100644
index 0000000..bcd3840
--- /dev/null
+++ b/api/oplet/src/test/java/org/apache/edgent/test/oplet/functional/FilterTest.java
@@ -0,0 +1,33 @@
+/*
+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.oplet.functional;
+
+import static org.junit.Assert.assertTrue;
+
+import org.apache.edgent.oplet.core.Pipe;
+import org.apache.edgent.oplet.functional.Filter;
+import org.junit.Test;
+
+public class FilterTest {
+
+ @Test
+ public void testHierachy() {
+ assertTrue(Pipe.class.isAssignableFrom(Filter.class));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/oplet/src/test/java/org/apache/edgent/test/oplet/operators/PipeTest.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/test/java/org/apache/edgent/test/oplet/operators/PipeTest.java b/api/oplet/src/test/java/org/apache/edgent/test/oplet/operators/PipeTest.java
new file mode 100644
index 0000000..d1077f5
--- /dev/null
+++ b/api/oplet/src/test/java/org/apache/edgent/test/oplet/operators/PipeTest.java
@@ -0,0 +1,59 @@
+/*
+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.oplet.operators;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.util.List;
+
+import org.apache.edgent.function.Consumer;
+import org.apache.edgent.oplet.Oplet;
+import org.apache.edgent.oplet.core.AbstractOplet;
+import org.apache.edgent.oplet.core.Pipe;
+import org.junit.Test;
+
+public class PipeTest {
+
+ @Test
+ public void testHierachy() {
+ assertTrue(Oplet.class.isAssignableFrom(Pipe.class));
+ assertTrue(AbstractOplet.class.isAssignableFrom(Pipe.class));
+ }
+
+ @Test
+ public void testDestination() {
+ Pipe<String, Integer> pipe = new Pipe<String, Integer>() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void accept(String tuple) {
+ }
+
+ @Override
+ public void close() throws Exception {
+ };
+ };
+
+ List<? extends Consumer<String>> inputs = pipe.getInputs();
+ assertEquals(1, inputs.size());
+ assertSame(pipe, inputs.get(0));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/topology/build.xml
----------------------------------------------------------------------
diff --git a/api/topology/build.xml b/api/topology/build.xml
index 9e404d5..f716582 100644
--- a/api/topology/build.xml
+++ b/api/topology/build.xml
@@ -46,10 +46,10 @@
<target name="test.compile" depends="edgent.common.test.compile,test.jar"/>
<target name="test.jar" depends="edgent.common.test.compile">
<jar destfile="${test.classes}/${ant.project.name}.APPS.TEST.jar">
- <service type="edgent.topology.services.TopologyBuilder">
- <provider classname="edgent.test.topology.services.TestApplications$AppOne" />
- <provider classname="edgent.test.topology.services.TestApplications$AppTwo" />
- <provider classname="edgent.test.topology.services.TestApplications$AppThree" />
+ <service type="org.apache.edgent.topology.services.TopologyBuilder">
+ <provider classname="org.apache.edgent.test.topology.services.TestApplications$AppOne" />
+ <provider classname="org.apache.edgent.test.topology.services.TestApplications$AppTwo" />
+ <provider classname="org.apache.edgent.test.topology.services.TestApplications$AppThree" />
</service>
<fileset dir="${test.classes}" includes="**/TestApplications*"/>
</jar>
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/topology/src/main/java/edgent/topology/TSink.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/edgent/topology/TSink.java b/api/topology/src/main/java/edgent/topology/TSink.java
deleted file mode 100644
index 8af8d18..0000000
--- a/api/topology/src/main/java/edgent/topology/TSink.java
+++ /dev/null
@@ -1,41 +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.topology;
-
-/**
- * Termination point (sink) for a stream.
- *
- * @param <T> Tuple type
- */
-public interface TSink<T> extends TopologyElement {
- /**
- * Get the stream feeding this sink.
- * The returned reference may be used for
- * further processing of the feeder stream.
- * <BR>
- * For example, {@code s.print().filter(...)}
- * <BR>
- * Here the filter is applied
- * to {@code s} so that {@code s} feeds
- * the {@code print()} and {@code filter()}.
- *
- * @return stream feeding this sink.
- */
- public TStream<T> getFeed();
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a129aa16/api/topology/src/main/java/edgent/topology/TStream.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/edgent/topology/TStream.java b/api/topology/src/main/java/edgent/topology/TStream.java
deleted file mode 100644
index e747624..0000000
--- a/api/topology/src/main/java/edgent/topology/TStream.java
+++ /dev/null
@@ -1,543 +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.topology;
-
-import java.util.EnumMap;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-
-import edgent.execution.services.ControlService;
-import edgent.function.BiFunction;
-import edgent.function.Consumer;
-import edgent.function.Function;
-import edgent.function.Predicate;
-import edgent.function.ToIntFunction;
-import edgent.function.UnaryOperator;
-import edgent.oplet.core.FanIn;
-import edgent.oplet.core.Pipe;
-import edgent.oplet.core.Sink;
-
-/**
- * A {@code TStream} is a declaration of a continuous sequence of tuples. A
- * connected topology of streams and functional transformations is built using
- * {@link Topology}. <BR>
- * Generic methods on this interface provide the ability to
- * {@link #filter(Predicate) filter}, {@link #map(Function)
- * map (or transform)} or {@link #sink(Consumer) sink} this declared stream using a
- * function.
- * <P>
- * {@code TStream} is not a runtime representation of a stream,
- * it is a declaration used in building a topology.
- * The actual runtime stream is created once the topology
- * is {@link edgent.execution.Submitter#submit(Object) submitted}
- * to a runtime.
- *
- * </P>
- * @param <T>
- * Tuple type.
- */
-public interface TStream<T> extends TopologyElement {
-
- /**
- * TYPE is used to identify {@link ControlService} mbeans registered for
- * for a TStream.
- * The value is {@value}
- */
- public static final String TYPE = "stream";
- // N.B. to avoid build problems due to topology <=> oplet,
- // other code contain a copy of this value (ugh) as TSTREAM_TYPE
-
- /**
- * Declare a new stream that filters tuples from this stream. Each tuple
- * {@code t} on this stream will appear in the returned stream if
- * {@link Predicate#test(Object) filter.test(t)} returns {@code true}. If
- * {@code filter.test(t)} returns {@code false} then then {@code t} will not
- * appear in the returned stream.
- * <P>
- * Examples of filtering out all empty strings from stream {@code s} of type
- * {@code String}
- * </P>
- *
- * <pre>
- * <code>
- * TStream<String> s = ...
- * TStream<String> filtered = s.filter(t -> !t.isEmpty());
- *
- * </code>
- * </pre>
- *
- * @param predicate
- * Filtering logic to be executed against each tuple.
- * @return Filtered stream
- */
- TStream<T> filter(Predicate<T> predicate);
-
- /**
- * Declare a new stream that maps (or transforms) each tuple from this stream into one
- * (or zero) tuple of a different type {@code U}. For each tuple {@code t}
- * on this stream, the returned stream will contain a tuple that is the
- * result of {@code mapper.apply(t)} when the return is not {@code null}.
- * If {@code mapper.apply(t)} returns {@code null} then no tuple
- * is submitted to the returned stream for {@code t}.
- *
- * <P>
- * Examples of transforming a stream containing numeric values as
- * {@code String} objects into a stream of {@code Double} values.
- * </P>
- *
- * <pre>
- * <code>
- * // Using lambda expression
- * TStream<String> strings = ...
- * TStream<Double> doubles = strings.map(v -> Double.valueOf(v));
- *
- * // Using method reference
- * TStream<String> strings = ...
- * TStream<Double> doubles = strings.map(Double::valueOf);
- *
- * </code>
- * </pre>
- *
- * @param <U> Tuple type of output stream
- * @param mapper
- * Mapping logic to be executed against each tuple.
- * @return Stream that will contain tuples of type {@code U} mapped from this
- * stream's tuples.
- */
- <U> TStream<U> map(Function<T, U> mapper);
-
- /**
- * Declare a new stream that maps tuples from this stream into one or
- * more (or zero) tuples of a different type {@code U}. For each tuple
- * {@code t} on this stream, the returned stream will contain all non-null tuples in
- * the {@code Iterator<U>} that is the result of {@code mapper.apply(t)}.
- * Tuples will be added to the returned stream in the order the iterator
- * returns them.
- *
- * <BR>
- * If the return is null or an empty iterator then no tuples are added to
- * the returned stream for input tuple {@code t}.
- * <P>
- * Examples of mapping a stream containing lines of text into a stream
- * of words split out from each line. The order of the words in the stream
- * will match the order of the words in the lines.
- * </P>
- *
- * <pre>
- * <code>
- * TStream<String> lines = ...
- * TStream<String> words = lines.flatMap(
- * line -> Arrays.asList(line.split(" ")));
- *
- * </code>
- * </pre>
- *
- * @param <U> Type of mapped input tuples.
- * @param mapper
- * Mapper logic to be executed against each tuple.
- * @return Stream that will contain tuples of type {@code U} mapped and flattened from this
- * stream's tuples.
- */
- <U> TStream<U> flatMap(Function<T, Iterable<U>> mapper);
-
- /**
- * Split a stream's tuples among {@code n} streams as specified by
- * {@code splitter}.
- *
- * <P>
- * For each tuple on the stream, {@code splitter.applyAsInt(tuple)} is
- * called. The return value {@code r} determines the destination stream:
- * </P>
- *
- * <pre>
- * if r < 0 the tuple is discarded
- * else it is sent to the stream at position (r % n) in the returned array.
- * </pre>
- *
- * <P>
- * Each split {@code TStream} is exposed by the API. The user has full
- * control over the each stream's processing pipeline. Each stream's
- * pipeline must be declared explicitly. Each stream can have different
- * processing pipelines.
- * </P>
- * <P>
- * An N-way {@code split()} is logically equivalent to a collection of N
- * {@code filter()} invocations, each with a predicate to select the tuples
- * for its stream. {@code split()} is more efficient. Each tuple is analyzed
- * only once by a single {@code splitter} instance to identify the
- * destination stream. For example, these are logically equivalent:
- * </P>
- * <pre>
- * List<TStream<String>> streams = stream.split(2, tuple -> tuple.length());
- *
- * TStream<String> stream0 = stream.filter(tuple -> (tuple.length() % 2) == 0);
- * TStream<String> stream1 = stream.filter(tuple -> (tuple.length() % 2) == 1);
- * </pre>
- * <P>
- * Example of splitting a stream of log records by their level attribute:
- * </P>
- *
- * <pre>
- * <code>
- * TStream<LogRecord> lrs = ...
- * List<<TStream<LogRecord>> splits = lrr.split(3, lr -> {
- if (SEVERE.equals(lr.getLevel()))
- return 0;
- else if (WARNING.equals(lr.getLevel()))
- return 1;
- else
- return 2;
- });
- * splits.get(0). ... // SEVERE log record processing pipeline
- * splits.get(1). ... // WARNING log record processing pipeline
- * splits.get(2). ... // "other" log record processing pipeline
- * </code>
- * </pre>
- *
- * @param n
- * the number of output streams
- * @param splitter
- * the splitter function
- * @return List of {@code n} streams
- *
- * @throws IllegalArgumentException
- * if {@code n <= 0}
- */
- List<TStream<T>> split(int n, ToIntFunction<T> splitter);
-
- /**
- * Split a stream's tuples among {@code enumClass.size} streams as specified by
- * {@code splitter}.
- *
- * @param <E> Enum type
- * @param enumClass
- * enum data to split
- * @param splitter
- * the splitter function
- * @return EnumMap<E,TStream<T>>
- * @throws IllegalArgumentException
- * if {@code enumclass.size <= 0}
- */
- <E extends Enum<E>> EnumMap<E,TStream<T>> split(Class<E> enumClass, Function<T, E> splitter);
-
- /**
- * Declare a stream that contains the same contents as this stream while
- * peeking at each element using {@code peeker}. <BR>
- * For each tuple {@code t} on this stream, {@code peeker.accept(t)} will be
- * called.
- *
- * @param peeker
- * Function to be called for each tuple.
- * @return {@code this}
- */
- TStream<T> peek(Consumer<T> peeker);
-
- /**
- * Sink (terminate) this stream using a function. For each tuple {@code t} on this stream
- * {@link Consumer#accept(Object) sinker.accept(t)} will be called. This is
- * typically used to send information to external systems, such as databases
- * or dashboards.
- * <p>
- * If {@code sinker} implements {@link AutoCloseable}, its {@code close()}
- * method will be called when the topology's execution is terminated.
- * </P>
- * <P>
- * Example of terminating a stream of {@code String} tuples by printing them
- * to {@code System.out}.
- * </P>
- *
- * <pre>
- * <code>
- * TStream<String> values = ...
- * values.sink(t -> System.out.println(tuple));
- * </code>
- * </pre>
- *
- * @param sinker
- * Logic to be executed against each tuple on this stream.
- * @return sink element representing termination of this stream.
- */
- TSink<T> sink(Consumer<T> sinker);
-
- /**
- * Sink (terminate) this stream using a oplet.
- * This provides a richer api for a sink than
- * {@link #sink(Consumer)} with a full life-cycle of
- * the oplet as well as easy access to
- * {@link edgent.execution.services.RuntimeServices runtime services}.
- *
- * @param oplet Oplet processes each tuple without producing output.
- * @return sink element representing termination of this stream.
- */
- TSink<T> sink(Sink<T> oplet);
-
- /**
- * Declare a stream that contains the output of the specified {@link Pipe}
- * oplet applied to this stream.
- *
- * @param <U> Tuple type of the returned stream.
- * @param pipe The {@link Pipe} oplet.
- *
- * @return Declared stream that contains the tuples emitted by the pipe
- * oplet.
- */
- <U> TStream<U> pipe(Pipe<T, U> pipe);
-
- /**
- * Declare a stream that contains the output of the specified
- * {@link FanIn} oplet applied to this stream and {@code others}.
- *
- * @param <U> Tuple type of the returned streams.
- * @param fanin The {@link FanIn} oplet.
- * @param others The other input streams.
- * Must not be empty or contain duplicates or {@code this}
- *
- * @return a stream that contains the tuples emitted by the oplet.
- * @see #union(Set)
- * @see #pipe(Pipe)
- * @see #sink(Sink)
- */
- <U> TStream<U> fanin(FanIn<T,U> fanin, List<TStream<T>> others);
-
- /**
- * Declare a new stream that modifies each tuple from this stream into one
- * (or zero) tuple of the same type {@code T}. For each tuple {@code t}
- * on this stream, the returned stream will contain a tuple that is the
- * result of {@code modifier.apply(t)} when the return is not {@code null}.
- * The function may return the same reference as its input {@code t} or
- * a different object of the same type.
- * If {@code modifier.apply(t)} returns {@code null} then no tuple
- * is submitted to the returned stream for {@code t}.
- *
- * <P>
- * Example of modifying a stream {@code String} values by adding the suffix '{@code extra}'.
- * </P>
- *
- * <pre>
- * <code>
- * TStream<String> strings = ...
- * TStream<String> modifiedStrings = strings.modify(t -> t.concat("extra"));
- * </code>
- * </pre>
- *
- * <P>
- * This method is equivalent to
- * {@code map(Function<T,T> modifier}).
- * </P>
- *
- * @param modifier
- * Modifier logic to be executed against each tuple.
- * @return Stream that will contain tuples of type {@code T} modified from this
- * stream's tuples.
- */
- TStream<T> modify(UnaryOperator<T> modifier);
-
- /**
- * Convert this stream to a stream of {@code String} tuples by calling
- * {@code toString()} on each tuple. This is equivalent to
- * {@code map(Object::toString)}.
- *
- * @return Declared stream that will contain each the string representation
- * of each tuple on this stream.
- */
- TStream<String> asString();
-
- /**
- * Utility method to print the contents of this stream
- * to {@code System.out} at runtime. Each tuple is printed
- * using {@code System.out.println(tuple)}.
- * @return {@code TSink} for the sink processing.
- */
- TSink<T> print();
-
- /**
- * Declare a partitioned window that continually represents the last {@code count}
- * tuples on this stream for each partition. Each partition independently maintains the last
- * {@code count} tuples for each key seen on this stream.
- * If no tuples have been seen on the stream for a key then the corresponding partition will be empty.
- * <BR>
- * The window is partitioned by each tuple's key, obtained by {@code keyFunction}.
- * For each tuple on the stream {@code keyFunction.apply(tuple)} is called
- * and the returned value is the tuple's key. For any two tuples {@code ta,tb} in a partition
- * {@code keyFunction.apply(ta).equals(keyFunction.apply(tb))} is true.
- * <BR>
- * The key function must return keys that implement {@code equals()} and {@code hashCode()} correctly.
- * <P>
- * To create a window partitioned using the tuple as the key use {@link edgent.function.Functions#identity() identity()}
- * as the key function.
- * </P>
- * <P>
- * To create an unpartitioned window use a key function that returns a constant,
- * by convention {@link edgent.function.Functions#unpartitioned() unpartitioned()} is recommended.
- * </P>
- *
- * @param <K> Key type.
- *
- * @param count Number of tuples to maintain in each partition.
- * @param keyFunction Function that defines the key for each tuple.
- * @return Window on this stream representing the last {@code count} tuples for each partition.
- */
- <K> TWindow<T, K> last(int count, Function<T, K> keyFunction);
-
- /**
- * Declare a partitioned window that continually represents the last {@code time} seconds of
- * tuples on this stream for each partition. If no tuples have been
- * seen on the stream for a key in the last {@code time} seconds then the partition will be empty.
- * Each partition independently maintains the last
- * {@code count} tuples for each key seen on this stream.
- * <BR>
- * The window is partitioned by each tuple's key, obtained by {@code keyFunction}.
- * For each tuple on the stream {@code keyFunction.apply(tuple)} is called
- * and the returned value is the tuple's key. For any two tuples {@code ta,tb} in a partition
- * {@code keyFunction.apply(ta).equals(keyFunction.apply(tb))} is true.
- * <BR>
- * The key function must return keys that implement {@code equals()} and {@code hashCode()} correctly.
- * <P>
- * To create a window partitioned using the tuple as the key use {@link edgent.function.Functions#identity() identity()}
- * as the key function.
- * </P>
- * <P>
- * To create an unpartitioned window use a key function that returns a constant,
- * by convention {@link edgent.function.Functions#unpartitioned() unpartitioned()} is recommended.
- * </P>
- *
- * @param <K> Key type.
- *
- * @param time Time to retain a tuple in a partition.
- * @param unit Unit for {@code time}.
- * @param keyFunction Function that defines the key for each tuple.
- * @return Partitioned window on this stream representing the last {@code count} tuple.
- */
- <K> TWindow<T, K> last(long time, TimeUnit unit, Function<T, K> keyFunction);
-
- /**
- * Declare a stream that will contain all tuples from this stream and
- * {@code other}. A stream cannot be unioned with itself, in this case
- * {@code this} will be returned.
- *
- * @param other the other stream
- * @return A stream that is the union of {@code this} and {@code other}.
- */
- TStream<T> union(TStream<T> other);
-
- /**
- * Declare a stream that will contain all tuples from this stream and all the
- * streams in {@code others}. A stream cannot be unioned with itself, in
- * this case the union will only contain tuples from this stream once. If
- * {@code others} is empty or only contains {@code this} then {@code this}
- * is returned.
- *
- * @param others
- * Stream to union with this stream.
- * @return A stream that is the union of {@code this} and {@code others}.
- */
- TStream<T> union(Set<TStream<T>> others);
-
- /**
- * Adds the specified tags to the stream. Adding the same tag to
- * a stream multiple times will not change the result beyond the
- * initial application.
- *
- * @param values
- * Tag values.
- * @return The tagged stream.
- */
- TStream<T> tag(String... values);
-
- /**
- * Returns the set of tags associated with this stream.
- *
- * @return set of tags
- */
- Set<String> getTags();
-
- /**
- * Set an alias for the stream.
- * <p>
- * The alias must be unique within the topology.
- * The alias may be used in various contexts:
- * </p>
- * <ul>
- * <li>Runtime control services for the stream are registered with this alias.</li>
- * </ul>
- *
- * @param alias an alias for the stream.
- * @return this
- * @throws IllegalStateException if the an alias has already been set.
- * @see ControlService
- */
- TStream<T> alias(String alias);
-
- /**
- * Returns the stream's alias if any.
- * @return the alias. null if one has not be set.
- */
- String getAlias();
-
- /**
- * Join this stream with a partitioned window of type {@code U} with key type {@code K}.
- * For each tuple on this stream, it is joined with the contents of {@code window}
- * for the key {@code keyer.apply(tuple)}. Each tuple is
- * passed into {@code joiner} and the return value is submitted to the
- * returned stream. If call returns null then no tuple is submitted.
- *
- * @param <J> Tuple type of result stream
- * @param <U> Tuple type of window to join with
- * @param <K> Key type
- * @param keyer Key function for this stream to match the window's key.
- * @param window Keyed window to join this stream with.
- * @param joiner Join function.
- * @return A stream that is the results of joining this stream with
- * {@code window}.
- */
- <J, U, K> TStream<J> join(Function<T, K> keyer, TWindow<U, K> window, BiFunction<T, List<U>, J> joiner);
-
- /**
- * Join this stream with the last tuple seen on a stream of type {@code U}
- * with partitioning.
- * For each tuple on this
- * stream, it is joined with the last tuple seen on {@code lastStream}
- * with a matching key (of type {@code K}).
- * <BR>
- * Each tuple {@code t} on this stream will match the last tuple
- * {@code u} on {@code lastStream} if
- * {@code keyer.apply(t).equals(lastStreamKeyer.apply(u))}
- * is true.
- * <BR>
- * The assumption is made that
- * the key classes correctly implement the contract for {@code equals} and
- * {@code hashCode()}.
- * <P>Each tuple is
- * passed into {@code joiner} and the return value is submitted to the
- * returned stream. If call returns null then no tuple is submitted.
- * </P>
- * @param <J> Tuple type of result stream
- * @param <U> Tuple type of stream to join with
- * @param <K> Key type
- * @param keyer Key function for this stream
- * @param lastStream Stream to join with.
- * @param lastStreamKeyer Key function for {@code lastStream}
- * @param joiner Join function.
- * @return A stream that is the results of joining this stream with
- * {@code lastStream}.
- */
- <J, U, K> TStream<J> joinLast(Function<T, K> keyer, TStream<U> lastStream, Function<U, K> lastStreamKeyer, BiFunction<T, U, J> joiner);
-
-}