You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2017/02/07 01:28:15 UTC
[02/12] storm git commit: [STORM-1961] Stream api for storm core use
cases
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/operations/ValueJoiner.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/operations/ValueJoiner.java b/storm-core/src/jvm/org/apache/storm/streams/operations/ValueJoiner.java
new file mode 100644
index 0000000..003d097
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/operations/ValueJoiner.java
@@ -0,0 +1,36 @@
+/**
+ * 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.storm.streams.operations;
+
+/**
+ * An interface for joining two values to produce a result.
+ *
+ * @param <V1> the type of the first value
+ * @param <V2> the type of the second value
+ * @param <R> the result type
+ */
+public interface ValueJoiner<V1, V2, R> extends Operation {
+ /**
+ * Joins two values and produces a result.
+ *
+ * @param value1 the first value
+ * @param value2 the second value
+ * @return the result
+ */
+ R apply(V1 value1, V2 value2);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/operations/aggregators/Count.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/operations/aggregators/Count.java b/storm-core/src/jvm/org/apache/storm/streams/operations/aggregators/Count.java
new file mode 100644
index 0000000..fd02d2a
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/operations/aggregators/Count.java
@@ -0,0 +1,37 @@
+/**
+ * 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.storm.streams.operations.aggregators;
+
+import org.apache.storm.streams.operations.Aggregator;
+
+/**
+ * Computes the count of values.
+ *
+ * @param <T> the value type
+ */
+public class Count<T> implements Aggregator<T, Long> {
+ @Override
+ public Long init() {
+ return 0L;
+ }
+
+ @Override
+ public Long apply(T value, Long aggregate) {
+ return aggregate + 1;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/operations/aggregators/Sum.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/operations/aggregators/Sum.java b/storm-core/src/jvm/org/apache/storm/streams/operations/aggregators/Sum.java
new file mode 100644
index 0000000..e232075
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/operations/aggregators/Sum.java
@@ -0,0 +1,35 @@
+/**
+ * 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.storm.streams.operations.aggregators;
+
+import org.apache.storm.streams.operations.Aggregator;
+
+/**
+ * Computes the long sum of the input values
+ */
+public class Sum implements Aggregator<Number, Long> {
+ @Override
+ public Long init() {
+ return 0L;
+ }
+
+ @Override
+ public Long apply(Number value, Long aggregate) {
+ return value.longValue() + aggregate;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/PairValueMapper.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/PairValueMapper.java b/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/PairValueMapper.java
new file mode 100644
index 0000000..1490e51
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/PairValueMapper.java
@@ -0,0 +1,51 @@
+/**
+ * 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.storm.streams.operations.mappers;
+
+import org.apache.storm.streams.Pair;
+import org.apache.storm.streams.operations.PairFunction;
+import org.apache.storm.tuple.Tuple;
+
+/**
+ * Extracts a typed key-value pair from a tuple.
+ *
+ * @param <K> the key type
+ * @param <V> the value type
+ */
+public class PairValueMapper<K, V> implements TupleValueMapper<Pair<K, V>>, PairFunction<Tuple, K, V> {
+ private final int keyIndex;
+ private final int valueIndex;
+
+ /**
+ * Constructs a new {@link PairValueMapper} that constructs a pair
+ * from a tuple based on the key and value index.
+ *
+ * @param keyIndex the key index
+ * @param valueIndex the value index
+ */
+ public PairValueMapper(int keyIndex, int valueIndex) {
+ this.keyIndex = keyIndex;
+ this.valueIndex = valueIndex;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Pair<K, V> apply(Tuple input) {
+ return Pair.of((K) input.getValue(keyIndex), (V) input.getValue(valueIndex));
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/TupleValueMapper.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/TupleValueMapper.java b/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/TupleValueMapper.java
new file mode 100644
index 0000000..6ffbd88
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/TupleValueMapper.java
@@ -0,0 +1,30 @@
+/**
+ * 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.storm.streams.operations.mappers;
+
+import org.apache.storm.streams.operations.Function;
+import org.apache.storm.tuple.Tuple;
+
+/**
+ * A generic interface for mapping a {@link Tuple} to typed
+ * values.
+ *
+ * @param <T> the result type
+ */
+public interface TupleValueMapper<T> extends Function<Tuple, T> {
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/ValueMapper.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/ValueMapper.java b/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/ValueMapper.java
new file mode 100644
index 0000000..0468d0d
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/ValueMapper.java
@@ -0,0 +1,45 @@
+/**
+ * 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.storm.streams.operations.mappers;
+
+import org.apache.storm.tuple.Tuple;
+
+/**
+ * Extracts a single typed value from a tuple.
+ *
+ * @param <T> the value type
+ */
+public class ValueMapper<T> implements TupleValueMapper<T> {
+ private final int index;
+
+ /**
+ * Constructs a new {@link ValueMapper} that extracts
+ * the value at index 'i' from a tuple.
+ *
+ * @param index the index
+ */
+ public ValueMapper(int index) {
+ this.index = index;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public T apply(Tuple input) {
+ return (T) input.getValue(index);
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/ValuesMapper.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/ValuesMapper.java b/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/ValuesMapper.java
new file mode 100644
index 0000000..f1d8e3b
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/operations/mappers/ValuesMapper.java
@@ -0,0 +1,48 @@
+/**
+ * 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.storm.streams.operations.mappers;
+
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+
+/**
+ * Constructs a {@link Values} from a {@link Tuple}
+ * based on indicies.
+ */
+public class ValuesMapper implements TupleValueMapper<Values> {
+ private final int[] indices;
+
+ /**
+ * Constructs a new {@link ValuesMapper} that extracts
+ * value from a {@link Tuple} at specified indices
+ *
+ * @param indices the indices
+ */
+ public ValuesMapper(int... indices) {
+ this.indices = indices;
+ }
+
+ @Override
+ public Values apply(Tuple input) {
+ Values values = new Values();
+ for (int i : indices) {
+ values.add(input.getValue(i));
+ }
+ return values;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/AggregateByKeyProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/AggregateByKeyProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/AggregateByKeyProcessor.java
new file mode 100644
index 0000000..d53a52d
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/AggregateByKeyProcessor.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.storm.streams.processors;
+
+import org.apache.storm.streams.Pair;
+import org.apache.storm.streams.operations.Aggregator;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class AggregateByKeyProcessor<K, V, R> extends BaseProcessor<Pair<K, V>> implements BatchProcessor {
+ private final Aggregator<V, R> aggregator;
+ private final Map<K, R> state = new HashMap<>();
+
+ public AggregateByKeyProcessor(Aggregator<V, R> aggregator) {
+ this.aggregator = aggregator;
+ }
+
+ @Override
+ public void execute(Pair<K, V> input) {
+ K key = input.getFirst();
+ V val = input.getSecond();
+ R agg = state.get(key);
+ if (agg == null) {
+ agg = aggregator.init();
+ }
+ state.put(key, aggregator.apply(val, agg));
+ mayBeForwardAggUpdate(Pair.of(key, state.get(key)));
+ }
+
+ @Override
+ public void finish() {
+ for (Map.Entry<K, R> entry : state.entrySet()) {
+ context.forward(Pair.of(entry.getKey(), entry.getValue()));
+ }
+ state.clear();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/AggregateProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/AggregateProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/AggregateProcessor.java
new file mode 100644
index 0000000..c5a1906
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/AggregateProcessor.java
@@ -0,0 +1,45 @@
+/**
+ * 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.storm.streams.processors;
+
+import org.apache.storm.streams.operations.Aggregator;
+
+public class AggregateProcessor<T, R> extends BaseProcessor<T> implements BatchProcessor {
+ private final Aggregator<T, R> aggregator;
+ private R state;
+
+ public AggregateProcessor(Aggregator<T, R> aggregator) {
+ this.aggregator = aggregator;
+ }
+
+ @Override
+ public void execute(T input) {
+ if (state == null) {
+ state = aggregator.init();
+ }
+ R curAggregate = (state != null) ? state : aggregator.init();
+ state = aggregator.apply(input, curAggregate);
+ mayBeForwardAggUpdate(state);
+ }
+
+ @Override
+ public void finish() {
+ context.forward(state);
+ state = null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/BaseProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/BaseProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/BaseProcessor.java
new file mode 100644
index 0000000..3ea469c
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/BaseProcessor.java
@@ -0,0 +1,107 @@
+/**
+ * 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.storm.streams.processors;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.apache.storm.streams.WindowNode.PUNCTUATION;
+
+/**
+ * Base implementation of the {@link Processor} interface that provides
+ * convenience methods {@link #execute(Object)} and {@link #finish()}.
+ *
+ * @param <T>
+ */
+abstract class BaseProcessor<T> implements Processor<T> {
+ private final Set<String> punctuationState = new HashSet<>();
+ protected ProcessorContext context;
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void init(ProcessorContext context) {
+ this.context = context;
+ }
+
+ /**
+ * {@inheritDoc}
+ * Processors that do not care about the source stream should
+ * override {@link BaseProcessor#execute(Object)} instead.
+ */
+ @Override
+ public void execute(T input, String streamId) {
+ execute(input);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void punctuate(String stream) {
+ if ((stream == null) || shouldPunctuate(stream)) {
+ finish();
+ context.forward(PUNCTUATION);
+ punctuationState.clear();
+ }
+ }
+
+ /**
+ * Execute some operation on the input value. Sub classes can override
+ * this when then don't care about the source stream from where the
+ * input is received.
+ *
+ * @param input the input
+ */
+ protected void execute(T input) {
+ // NOOP
+ }
+
+ /**
+ * This is triggered to signal the end of the current
+ * batch of values. Sub classes can override this to
+ * emit the result of a batch of values, for e.g. to emit
+ * the result of an aggregate or join operation on a batch
+ * of values. If a processor does per-value operation like
+ * filter, map etc, they can choose to ignore this.
+ */
+ protected void finish() {
+ // NOOP
+ }
+
+ /**
+ * Forwards the result update to downstream processors. Processors
+ * that operate on a batch of tuples, like aggregation, join etc
+ * can use this to emit the partial results on each input
+ * if they are operating in non-windowed mode.
+ *
+ * @param result the result
+ * @param <R> the result type
+ */
+ protected final <R> void mayBeForwardAggUpdate(R result) {
+ if (!context.isWindowed()) {
+ context.forward(result);
+ }
+ }
+
+ private boolean shouldPunctuate(String parentStream) {
+ punctuationState.add(parentStream);
+ return punctuationState.equals(context.getWindowedParentStreams());
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/BatchProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/BatchProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/BatchProcessor.java
new file mode 100644
index 0000000..f9a39b4
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/BatchProcessor.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.storm.streams.processors;
+
+/**
+ * Top level marker interface for processors that computes
+ * results for a batch of tuples like Aggregate, Join etc.
+ */
+public interface BatchProcessor {
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/BranchProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/BranchProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/BranchProcessor.java
new file mode 100644
index 0000000..f8bc739
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/BranchProcessor.java
@@ -0,0 +1,41 @@
+/**
+ * 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.storm.streams.processors;
+
+import org.apache.storm.streams.operations.Predicate;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class BranchProcessor<T> extends BaseProcessor<T> {
+ private final Map<Predicate<T>, String> predicateToStream = new HashMap<>();
+
+ public void addPredicate(Predicate<T> predicate, String stream) {
+ predicateToStream.put(predicate, stream);
+ }
+
+ @Override
+ public void execute(T input) {
+ for (Map.Entry<Predicate<T>, String> entry : predicateToStream.entrySet()) {
+ if (entry.getKey().test(input)) {
+ context.forward(input, entry.getValue());
+ break;
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/ChainedProcessorContext.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/ChainedProcessorContext.java b/storm-core/src/jvm/org/apache/storm/streams/processors/ChainedProcessorContext.java
new file mode 100644
index 0000000..c92d992
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/ChainedProcessorContext.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.storm.streams.processors;
+
+import org.apache.storm.streams.ProcessorNode;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * A composite context that holds a chain of {@link ProcessorContext}
+ */
+public class ChainedProcessorContext implements ProcessorContext {
+ private final ProcessorNode processorNode;
+ private final List<? extends ProcessorContext> contexts;
+
+ public ChainedProcessorContext(ProcessorNode processorNode, List<? extends ProcessorContext> contexts) {
+ this.processorNode = processorNode;
+ this.contexts = new ArrayList<>(contexts);
+ }
+
+ public ChainedProcessorContext(ProcessorNode processorNode, ProcessorContext... contexts) {
+ this(processorNode, Arrays.asList(contexts));
+ }
+
+ @Override
+ public <T> void forward(T input) {
+ for (ProcessorContext context : contexts) {
+ context.forward(input);
+ }
+ }
+
+ @Override
+ public <T> void forward(T input, String stream) {
+ for (ProcessorContext context : contexts) {
+ context.forward(input, stream);
+ }
+ }
+
+ @Override
+ public boolean isWindowed() {
+ return processorNode.isWindowed();
+ }
+
+ @Override
+ public Set<String> getWindowedParentStreams() {
+ return processorNode.getWindowedParentStreams();
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/EmittingProcessorContext.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/EmittingProcessorContext.java b/storm-core/src/jvm/org/apache/storm/streams/processors/EmittingProcessorContext.java
new file mode 100644
index 0000000..d841080
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/EmittingProcessorContext.java
@@ -0,0 +1,170 @@
+/**
+ * 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.storm.streams.processors;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Collections2;
+import org.apache.storm.streams.Pair;
+import org.apache.storm.streams.ProcessorNode;
+import org.apache.storm.streams.RefCountedTuple;
+import org.apache.storm.streams.StreamUtil;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.storm.streams.WindowNode.PUNCTUATION;
+
+/**
+ * A context that emits the results to downstream processors which are in
+ * another bolt.
+ */
+public class EmittingProcessorContext implements ProcessorContext {
+ private static final Logger LOG = LoggerFactory.getLogger(EmittingProcessorContext.class);
+ private final ProcessorNode processorNode;
+ private final String outputStreamId;
+ private final OutputCollector collector;
+ private final Fields outputFields;
+ private final Values punctuation;
+ private final List<RefCountedTuple> anchors = new ArrayList<>();
+ private boolean emitPunctuation = true;
+ private long eventTimestamp;
+ private String timestampField;
+
+ public EmittingProcessorContext(ProcessorNode processorNode, OutputCollector collector, String outputStreamId) {
+ this.processorNode = processorNode;
+ this.outputStreamId = outputStreamId;
+ this.collector = collector;
+ outputFields = processorNode.getOutputFields();
+ punctuation = createPunctuation();
+ }
+
+ @Override
+ public <T> void forward(T input) {
+ if (input instanceof Pair) {
+ Pair<?, ?> value = (Pair<?, ?>) input;
+ emit(new Values(value.getFirst(), value.getSecond()));
+ } else if (PUNCTUATION.equals(input)) {
+ if (emitPunctuation) {
+ emit(punctuation);
+ } else {
+ LOG.debug("Not emitting punctuation since emitPunctuation is false");
+ }
+ maybeAck();
+ } else {
+ emit(new Values(input));
+ }
+ }
+
+ @Override
+ public <T> void forward(T input, String stream) {
+ if (stream.equals(outputStreamId)) {
+ forward(input);
+ }
+ }
+
+ @Override
+ public boolean isWindowed() {
+ return processorNode.isWindowed();
+ }
+
+ @Override
+ public Set<String> getWindowedParentStreams() {
+ return processorNode.getWindowedParentStreams();
+ }
+
+ public void setEmitPunctuation(boolean emitPunctuation) {
+ this.emitPunctuation = emitPunctuation;
+ }
+
+ public void setTimestampField(String fieldName) {
+ timestampField = fieldName;
+ }
+
+ public void setAnchor(RefCountedTuple anchor) {
+ if (processorNode.isWindowed() && processorNode.isBatch()) {
+ anchor.increment();
+ anchors.add(anchor);
+ } else {
+ if (anchors.isEmpty()) {
+ anchors.add(anchor);
+ } else {
+ anchors.set(0, anchor);
+ }
+ /*
+ * track punctuation in non-batch mode so that the
+ * punctuation is acked after all the processors have emitted the punctuation downstream.
+ */
+ if (StreamUtil.isPunctuation(anchor.tuple().getValue(0))) {
+ anchor.increment();
+ }
+ }
+ }
+
+ public void setEventTimestamp(long timestamp) {
+ this.eventTimestamp = timestamp;
+ }
+
+ private Values createPunctuation() {
+ Values values = new Values();
+ for (int i = 0; i < outputFields.size(); i++) {
+ values.add(PUNCTUATION);
+ }
+ return values;
+ }
+
+ private void maybeAck() {
+ if (!anchors.isEmpty()) {
+ for (RefCountedTuple anchor : anchors) {
+ anchor.decrement();
+ if (anchor.shouldAck()) {
+ LOG.debug("Acking {} ", anchor);
+ collector.ack(anchor.tuple());
+ anchor.setAcked();
+ }
+ }
+ anchors.clear();
+ }
+ }
+
+ private Collection<Tuple> tuples(Collection<RefCountedTuple> anchors) {
+ return anchors.stream().map(RefCountedTuple::tuple).collect(Collectors.toList());
+ }
+
+ private void emit(Values values) {
+ if (timestampField != null) {
+ values.add(eventTimestamp);
+ }
+ if (anchors.isEmpty()) {
+ // for windowed bolt, windowed output collector will do the anchoring/acking
+ LOG.debug("Emit un-anchored, outputStreamId: {}, values: {}", outputStreamId, values);
+ collector.emit(outputStreamId, values);
+ } else {
+ LOG.debug("Emit, outputStreamId: {}, anchors: {}, values: {}", outputStreamId, anchors, values);
+ collector.emit(outputStreamId, tuples(anchors), values);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/FilterProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/FilterProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/FilterProcessor.java
new file mode 100644
index 0000000..f245a44
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/FilterProcessor.java
@@ -0,0 +1,35 @@
+/**
+ * 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.storm.streams.processors;
+
+import org.apache.storm.streams.operations.Predicate;
+
+public class FilterProcessor<T> extends BaseProcessor<T> {
+ private final Predicate<T> predicate;
+
+ public FilterProcessor(Predicate<T> predicate) {
+ this.predicate = predicate;
+ }
+
+ @Override
+ public void execute(T input) {
+ if (predicate.test(input)) {
+ context.forward(input);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/FlatMapProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/FlatMapProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/FlatMapProcessor.java
new file mode 100644
index 0000000..fdc86fe
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/FlatMapProcessor.java
@@ -0,0 +1,35 @@
+/**
+ * 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.storm.streams.processors;
+
+import org.apache.storm.streams.operations.FlatMapFunction;
+
+public class FlatMapProcessor<T, R> extends BaseProcessor<T> {
+ private final FlatMapFunction<T, R> function;
+
+ public FlatMapProcessor(FlatMapFunction<T, R> function) {
+ this.function = function;
+ }
+
+ @Override
+ public void execute(T input) {
+ for (R res : function.apply(input)) {
+ context.forward(res);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/FlatMapValuesProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/FlatMapValuesProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/FlatMapValuesProcessor.java
new file mode 100644
index 0000000..92953a3
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/FlatMapValuesProcessor.java
@@ -0,0 +1,36 @@
+/**
+ * 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.storm.streams.processors;
+
+import org.apache.storm.streams.Pair;
+import org.apache.storm.streams.operations.FlatMapFunction;
+
+public class FlatMapValuesProcessor<K, V, R> extends BaseProcessor<Pair<K, V>> {
+ private final FlatMapFunction<V, R> function;
+
+ public FlatMapValuesProcessor(FlatMapFunction<V, R> function) {
+ this.function = function;
+ }
+
+ @Override
+ protected void execute(Pair<K, V> input) {
+ for (R res : function.apply(input.getSecond())) {
+ context.forward(Pair.of(input.getFirst(), res));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/ForEachProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/ForEachProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/ForEachProcessor.java
new file mode 100644
index 0000000..323b532
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/ForEachProcessor.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.storm.streams.processors;
+
+import org.apache.storm.streams.operations.Consumer;
+
+public class ForEachProcessor<T> extends BaseProcessor<T> {
+ private final Consumer<T> consumer;
+
+ public ForEachProcessor(Consumer<T> consumer) {
+ this.consumer = consumer;
+ }
+
+ @Override
+ public void execute(T input) {
+ consumer.accept(input);
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/ForwardingProcessorContext.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/ForwardingProcessorContext.java b/storm-core/src/jvm/org/apache/storm/streams/processors/ForwardingProcessorContext.java
new file mode 100644
index 0000000..49eb836
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/ForwardingProcessorContext.java
@@ -0,0 +1,102 @@
+/**
+ * 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.storm.streams.processors;
+
+import com.google.common.collect.Multimap;
+import org.apache.storm.streams.ProcessorNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Set;
+
+import static org.apache.storm.streams.WindowNode.PUNCTUATION;
+
+/**
+ * A context that emits the results to downstream processors which are in
+ * the same bolt.
+ */
+public class ForwardingProcessorContext implements ProcessorContext {
+ private static final Logger LOG = LoggerFactory.getLogger(ForwardingProcessorContext.class);
+ private final ProcessorNode processorNode;
+ private final Multimap<String, ProcessorNode> streamToChildren;
+ private final Set<String> streams;
+
+ public ForwardingProcessorContext(ProcessorNode processorNode, Multimap<String, ProcessorNode> streamToChildren) {
+ this.processorNode = processorNode;
+ this.streamToChildren = streamToChildren;
+ this.streams = streamToChildren.keySet();
+ }
+
+ @Override
+ public <T> void forward(T input) {
+ if (PUNCTUATION.equals(input)) {
+ finishAllStreams();
+ } else {
+ executeAllStreams(input);
+ }
+ }
+
+ @Override
+ public <T> void forward(T input, String stream) {
+ if (PUNCTUATION.equals(input)) {
+ finish(stream);
+ } else {
+ execute(input, stream);
+ }
+ }
+
+ @Override
+ public boolean isWindowed() {
+ return processorNode.isWindowed();
+ }
+
+ @Override
+ public Set<String> getWindowedParentStreams() {
+ return processorNode.getWindowedParentStreams();
+ }
+
+ private void finishAllStreams() {
+ for (String stream : streams) {
+ finish(stream);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private <T> void finish(String stream) {
+ for (ProcessorNode node : streamToChildren.get(stream)) {
+ LOG.debug("Punctuating processor: {}", node);
+ Processor<T> processor = (Processor<T>) node.getProcessor();
+ processor.punctuate(stream);
+ }
+ }
+
+ private <T> void executeAllStreams(T input) {
+ for (String stream : streams) {
+ execute(input, stream);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private <T> void execute(T input, String stream) {
+ for (ProcessorNode node : streamToChildren.get(stream)) {
+ LOG.debug("Forward input: {} to processor node: {}", input, node);
+ Processor<T> processor = (Processor<T>) node.getProcessor();
+ processor.execute(input, stream);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/JoinProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/JoinProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/JoinProcessor.java
new file mode 100644
index 0000000..d56cfea
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/JoinProcessor.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.storm.streams.processors;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.Multimap;
+import org.apache.storm.streams.Pair;
+import org.apache.storm.streams.Tuple3;
+import org.apache.storm.streams.operations.ValueJoiner;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Provides equi-join implementation based on simple hash-join.
+ */
+public class JoinProcessor<K, R, V1, V2> extends BaseProcessor<Pair<K, ?>> implements BatchProcessor {
+ private final ValueJoiner<V1, V2, R> valueJoiner;
+ private final String leftStream;
+ private final String rightStream;
+ private final List<Pair<K, V1>> leftRows = new ArrayList<>();
+ private final List<Pair<K, V2>> rightRows = new ArrayList<>();
+
+ public JoinProcessor(String leftStream, String rightStream, ValueJoiner<V1, V2, R> valueJoiner) {
+ this.valueJoiner = valueJoiner;
+ this.leftStream = leftStream;
+ this.rightStream = rightStream;
+ }
+
+ @Override
+ public void execute(Pair<K, ?> input, String sourceStream) {
+ K key = input.getFirst();
+ if (sourceStream.equals(leftStream)) {
+ V1 val = (V1) input.getSecond();
+ Pair<K, V1> pair = Pair.of(key, val);
+ leftRows.add(pair);
+ if (!context.isWindowed()) {
+ joinAndForward(Collections.singletonList(pair), rightRows);
+ }
+ } else if (sourceStream.equals(rightStream)) {
+ V2 val = (V2) input.getSecond();
+ Pair<K, V2> pair = Pair.of(key, val);
+ rightRows.add(pair);
+ if (!context.isWindowed()) {
+ joinAndForward(leftRows, Collections.singletonList(pair));
+ }
+ }
+ }
+
+ @Override
+ public void finish() {
+ joinAndForward(leftRows, rightRows);
+ leftRows.clear();
+ rightRows.clear();
+ }
+
+ public String getLeftStream() {
+ return leftStream;
+ }
+
+ public String getRightStream() {
+ return rightStream;
+ }
+
+ private void joinAndForward(List<Pair<K, V1>> leftRows, List<Pair<K, V2>> rightRows) {
+ if (leftRows.size() <= rightRows.size()) {
+ for (Tuple3<K, V1, V2> res : join(getJoinTable(leftRows), rightRows)) {
+ context.forward(Pair.of(res._1, valueJoiner.apply(res._2, res._3)));
+ }
+ } else {
+ for (Tuple3<K, V2, V1> res : join(getJoinTable(rightRows), leftRows)) {
+ context.forward(Pair.of(res._1, valueJoiner.apply(res._3, res._2)));
+ }
+ }
+ }
+
+ private <T1, T2> List<Tuple3<K, T1, T2>> join(Multimap<K, T1> tab, List<Pair<K, T2>> rows) {
+ List<Tuple3<K, T1, T2>> res = new ArrayList<>();
+ for (Pair<K, T2> row : rows) {
+ for (T1 mapValue : tab.get(row.getFirst())) {
+ if (mapValue != null) {
+ res.add(new Tuple3<>(row.getFirst(), mapValue, row.getSecond()));
+ }
+ }
+ }
+ return res;
+ }
+
+ private <T> Multimap<K, T> getJoinTable(List<Pair<K, T>> rows) {
+ Multimap<K, T> m = ArrayListMultimap.create();
+ for (Pair<K, T> v : rows) {
+ m.put(v.getFirst(), v.getSecond());
+ }
+ return m;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/MapProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/MapProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/MapProcessor.java
new file mode 100644
index 0000000..e9f89a9
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/MapProcessor.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.storm.streams.processors;
+
+import org.apache.storm.streams.operations.Function;
+
+public class MapProcessor<T, R> extends BaseProcessor<T> {
+ private final Function<T, R> function;
+
+ public MapProcessor(Function<T, R> function) {
+ this.function = function;
+ }
+
+ @Override
+ public void execute(T input) {
+ context.forward(function.apply(input));
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/MapValuesProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/MapValuesProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/MapValuesProcessor.java
new file mode 100644
index 0000000..a963e2e
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/MapValuesProcessor.java
@@ -0,0 +1,34 @@
+/**
+ * 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.storm.streams.processors;
+
+import org.apache.storm.streams.Pair;
+import org.apache.storm.streams.operations.Function;
+
+public class MapValuesProcessor<K, V, R> extends BaseProcessor<Pair<K, V>> {
+ private final Function<V, R> function;
+
+ public MapValuesProcessor(Function<V, R> function) {
+ this.function = function;
+ }
+
+ @Override
+ public void execute(Pair<K, V> input) {
+ context.forward(Pair.of(input.getFirst(), function.apply(input.getSecond())));
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/PeekProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/PeekProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/PeekProcessor.java
new file mode 100644
index 0000000..3014a98
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/PeekProcessor.java
@@ -0,0 +1,34 @@
+/**
+ * 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.storm.streams.processors;
+
+import org.apache.storm.streams.operations.Consumer;
+
+public class PeekProcessor<T> extends BaseProcessor<T> {
+ private final Consumer<T> consumer;
+
+ public PeekProcessor(Consumer<T> consumer) {
+ this.consumer = consumer;
+ }
+
+ @Override
+ public void execute(T input) {
+ consumer.accept(input);
+ context.forward(input);
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/Processor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/Processor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/Processor.java
new file mode 100644
index 0000000..5551682
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/Processor.java
@@ -0,0 +1,51 @@
+/**
+ * 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.storm.streams.processors;
+
+import java.io.Serializable;
+
+/**
+ * A processor processes a stream of elements and produces some result.
+ *
+ * @param <T> the type of the input that is processed
+ */
+public interface Processor<T> extends Serializable {
+ /**
+ * Initializes the processor. This is typically invoked from the underlying
+ * storm bolt's prepare method.
+ *
+ * @param context the processor context
+ */
+ void init(ProcessorContext context);
+
+ /**
+ * Executes some operation on the input and possibly emits some result.
+ *
+ * @param input the input to be processed
+ * @param streamId the source stream id from where the input is received
+ */
+ void execute(T input, String streamId);
+
+ /**
+ * Punctuation marks end of a batch which can be used to compute and pass
+ * the results of one stage in the pipeline to the next. For e.g. emit the results of an aggregation.
+ *
+ * @param stream the stream id on which the punctuation arrived
+ */
+ void punctuate(String stream);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/ProcessorContext.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/ProcessorContext.java b/storm-core/src/jvm/org/apache/storm/streams/processors/ProcessorContext.java
new file mode 100644
index 0000000..2a10a22
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/ProcessorContext.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.storm.streams.processors;
+
+import java.io.Serializable;
+import java.util.Set;
+
+/**
+ * Context information passed to the {@link Processor}.
+ */
+public interface ProcessorContext extends Serializable {
+ /**
+ * Forwards the input to all downstream processors.
+ *
+ * @param input the input
+ * @param <T> the type of the input
+ */
+ <T> void forward(T input);
+
+ /**
+ * Forwards the input to downstream processors at specified stream.
+ *
+ * @param input the input
+ * @param stream the stream to forward
+ * @param <T> the type of the input
+ */
+ <T> void forward(T input, String stream);
+
+ /**
+ * Returns if the processing is in a windowed context and should wait for
+ * punctuation before emitting results.
+ *
+ * @return if this is a windowed context or not
+ */
+ boolean isWindowed();
+
+ /**
+ * Returns the windowed parent streams. These are the streams
+ * where punctuations arrive.
+ *
+ * @return the windowed parent streams
+ */
+ Set<String> getWindowedParentStreams();
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/ReduceByKeyProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/ReduceByKeyProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/ReduceByKeyProcessor.java
new file mode 100644
index 0000000..eba0de6
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/ReduceByKeyProcessor.java
@@ -0,0 +1,52 @@
+/**
+ * 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.storm.streams.processors;
+
+import org.apache.storm.streams.Pair;
+import org.apache.storm.streams.operations.Reducer;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class ReduceByKeyProcessor<K, V> extends BaseProcessor<Pair<K, V>> implements BatchProcessor {
+ private final Reducer<V> reducer;
+ private final Map<K, V> state = new HashMap<>();
+
+ public ReduceByKeyProcessor(Reducer<V> reducer) {
+ this.reducer = reducer;
+ }
+
+ @Override
+ public void execute(Pair<K, V> input) {
+ K key = input.getFirst();
+ V val = input.getSecond();
+ V agg = state.get(key);
+ agg = (agg == null) ? val : reducer.apply(agg, val);
+ state.put(key, agg);
+ mayBeForwardAggUpdate(Pair.of(key, agg));
+ }
+
+ @Override
+ public void finish() {
+ for (Map.Entry<K, V> entry : state.entrySet()) {
+ context.forward(Pair.of(entry.getKey(), entry.getValue()));
+ }
+ state.clear();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/ReduceProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/ReduceProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/ReduceProcessor.java
new file mode 100644
index 0000000..d64e114
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/ReduceProcessor.java
@@ -0,0 +1,41 @@
+/**
+ * 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.storm.streams.processors;
+
+import org.apache.storm.streams.operations.Reducer;
+
+public class ReduceProcessor<T> extends BaseProcessor<T> implements BatchProcessor {
+ private final Reducer<T> reducer;
+ private T agg;
+
+ public ReduceProcessor(Reducer<T> reducer) {
+ this.reducer = reducer;
+ }
+
+ @Override
+ public void execute(T input) {
+ agg = (agg == null) ? input : reducer.apply(agg, input);
+ mayBeForwardAggUpdate(agg);
+ }
+
+ @Override
+ public void finish() {
+ context.forward(agg);
+ agg = null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/StateQueryProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/StateQueryProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/StateQueryProcessor.java
new file mode 100644
index 0000000..e577a65
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/StateQueryProcessor.java
@@ -0,0 +1,48 @@
+/**
+ * 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.storm.streams.processors;
+
+import org.apache.storm.state.KeyValueState;
+import org.apache.storm.streams.Pair;
+import org.apache.storm.streams.StreamState;
+
+public class StateQueryProcessor<K, V> extends BaseProcessor<K> implements StatefulProcessor<K, V> {
+ private final StreamState<K, V> streamState;
+ private KeyValueState<K, V> keyValueState;
+
+ public StateQueryProcessor(StreamState<K, V> streamState) {
+ this.streamState = streamState;
+ }
+
+ @Override
+ public void initState(KeyValueState<K, V> keyValueState) {
+ this.keyValueState = keyValueState;
+ }
+
+ public StreamState<K, V> getStreamState() {
+ return streamState;
+ }
+
+ @Override
+ protected void execute(K key) {
+ V value = keyValueState.get(key);
+ if (value != null) {
+ context.forward(Pair.of(key, value));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/StatefulProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/StatefulProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/StatefulProcessor.java
new file mode 100644
index 0000000..acc4188
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/StatefulProcessor.java
@@ -0,0 +1,36 @@
+/**
+ * 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.storm.streams.processors;
+
+import org.apache.storm.state.KeyValueState;
+
+/**
+ * Top level interface for processors that does stateful processing.
+ *
+ * @param <K> the key type
+ * @param <V> the value type
+ */
+public interface StatefulProcessor<K, V> {
+ /**
+ * Initialize the state of the processor with the given
+ * {@link KeyValueState}
+ *
+ * @param keyValueState the key-value state
+ */
+ void initState(KeyValueState<K, V> keyValueState);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/processors/UpdateStateByKeyProcessor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/processors/UpdateStateByKeyProcessor.java b/storm-core/src/jvm/org/apache/storm/streams/processors/UpdateStateByKeyProcessor.java
new file mode 100644
index 0000000..9e068a0
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/processors/UpdateStateByKeyProcessor.java
@@ -0,0 +1,49 @@
+/**
+ * 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.storm.streams.processors;
+
+import org.apache.storm.state.KeyValueState;
+import org.apache.storm.streams.Pair;
+import org.apache.storm.streams.operations.Aggregator;
+
+public class UpdateStateByKeyProcessor<K, V, R> extends BaseProcessor<Pair<K, V>> implements StatefulProcessor<K, R> {
+ private final Aggregator<V, R> aggregator;
+ private KeyValueState<K, R> keyValueState;
+
+ public UpdateStateByKeyProcessor(Aggregator<V, R> aggregator) {
+ this.aggregator = aggregator;
+ }
+
+ @Override
+ public void initState(KeyValueState<K, R> keyValueState) {
+ this.keyValueState = keyValueState;
+ }
+
+ @Override
+ protected void execute(Pair<K, V> input) {
+ K key = input.getFirst();
+ V val = input.getSecond();
+ R agg = keyValueState.get(key);
+ if (agg == null) {
+ agg = aggregator.init();
+ }
+ R newAgg = aggregator.apply(val, agg);
+ keyValueState.put(key, newAgg);
+ context.forward(Pair.of(key, newAgg));
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/windowing/BaseWindow.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/windowing/BaseWindow.java b/storm-core/src/jvm/org/apache/storm/streams/windowing/BaseWindow.java
new file mode 100644
index 0000000..4526fa5
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/windowing/BaseWindow.java
@@ -0,0 +1,64 @@
+/**
+ * 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.storm.streams.windowing;
+
+import static org.apache.storm.topology.base.BaseWindowedBolt.Duration;
+
+public abstract class BaseWindow<L, I> implements Window<L, I> {
+ protected String timestampField;
+ protected String lateTupleStream;
+ protected Duration lag;
+
+ @Override
+ public String getTimestampField() {
+ return timestampField;
+ }
+
+ @Override
+ public String getLateTupleStream() {
+ return lateTupleStream;
+ }
+
+ @Override
+ public Duration getLag() {
+ return lag;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ BaseWindow<?, ?> that = (BaseWindow<?, ?>) o;
+
+ if (timestampField != null ? !timestampField.equals(that.timestampField) : that.timestampField != null)
+ return false;
+ if (lateTupleStream != null ? !lateTupleStream.equals(that.lateTupleStream) : that.lateTupleStream != null)
+ return false;
+ return lag != null ? lag.equals(that.lag) : that.lag == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = timestampField != null ? timestampField.hashCode() : 0;
+ result = 31 * result + (lateTupleStream != null ? lateTupleStream.hashCode() : 0);
+ result = 31 * result + (lag != null ? lag.hashCode() : 0);
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/windowing/SlidingWindows.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/windowing/SlidingWindows.java b/storm-core/src/jvm/org/apache/storm/streams/windowing/SlidingWindows.java
new file mode 100644
index 0000000..275d3c3
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/windowing/SlidingWindows.java
@@ -0,0 +1,151 @@
+/**
+ * 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.storm.streams.windowing;
+
+import org.apache.storm.topology.base.BaseWindowedBolt;
+import org.apache.storm.topology.base.BaseWindowedBolt.Count;
+import org.apache.storm.topology.base.BaseWindowedBolt.Duration;
+
+/**
+ * A sliding window specification based on a window length and sliding interval.
+ *
+ * @param <L> The type of the window length (e.g. Duration or Count)
+ * @param <I> The type of the sliding interval (e.g. Duration or Count)
+ */
+public class SlidingWindows<L, I> extends BaseWindow<L, I> {
+ private final L windowLength;
+ private final I slidingInterval;
+
+ private SlidingWindows(L windowLength, I slidingInterval) {
+ this.windowLength = windowLength;
+ this.slidingInterval = slidingInterval;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public L getWindowLength() {
+ return windowLength;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public I getSlidingInterval() {
+ return slidingInterval;
+ }
+
+ /**
+ * Count based sliding window configuration.
+ *
+ * @param windowLength the number of tuples in the window
+ * @param slidingInterval the number of tuples after which the window slides
+ */
+ public static SlidingWindows<Count, Count> of(Count windowLength, Count slidingInterval) {
+ return new SlidingWindows<>(windowLength, slidingInterval);
+ }
+
+ /**
+ * Time duration based sliding window configuration.
+ *
+ * @param windowLength the time duration of the window
+ * @param slidingInterval the time duration after which the window slides
+ */
+ public static SlidingWindows<Duration, Duration> of(Duration windowLength, Duration slidingInterval) {
+ return new SlidingWindows<>(windowLength, slidingInterval);
+ }
+
+ /**
+ * Tuple count and time duration based sliding window configuration.
+ *
+ * @param windowLength the number of tuples in the window
+ * @param slidingInterval the time duration after which the window slides
+ */
+ public static SlidingWindows<Count, Duration> of(Count windowLength, Duration slidingInterval) {
+ return new SlidingWindows<>(windowLength, slidingInterval);
+ }
+
+ /**
+ * Time duration and count based sliding window configuration.
+ *
+ * @param windowLength the time duration of the window
+ * @param slidingInterval the number of tuples after which the window slides
+ */
+ public static SlidingWindows<Duration, Count> of(Duration windowLength, Count slidingInterval) {
+ return new SlidingWindows<>(windowLength, slidingInterval);
+ }
+
+ /**
+ * The name of the field in the tuple that contains the timestamp when the event occurred as a long value.
+ * This is used of event-time based processing. If this config is set and the field is not present in the incoming tuple,
+ * an {@link IllegalArgumentException} will be thrown.
+ *
+ * @param fieldName the name of the field that contains the timestamp
+ */
+ public SlidingWindows<L, I> withTimestampField(String fieldName) {
+ timestampField = fieldName;
+ return this;
+ }
+
+ /**
+ * Specify a stream id on which late tuples are going to be emitted. They are going to be accessible via the
+ * {@link org.apache.storm.topology.WindowedBoltExecutor#LATE_TUPLE_FIELD} field.
+ * It must be defined on a per-component basis, and in conjunction with the
+ * {@link BaseWindowedBolt#withTimestampField}, otherwise {@link IllegalArgumentException} will be thrown.
+ *
+ * @param streamId the name of the stream used to emit late tuples on
+ */
+ public SlidingWindows<L, I> withLateTupleStream(String streamId) {
+ lateTupleStream = streamId;
+ return this;
+ }
+
+ /**
+ * Specify the maximum time lag of the tuple timestamp in milliseconds. It means that the tuple timestamps
+ * cannot be out of order by more than this amount.
+ *
+ * @param duration the max lag duration
+ */
+ public SlidingWindows<L, I> withLag(Duration duration) {
+ lag = duration;
+ return this;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ if (!super.equals(o)) return false;
+
+ SlidingWindows<?, ?> that = (SlidingWindows<?, ?>) o;
+
+ if (windowLength != null ? !windowLength.equals(that.windowLength) : that.windowLength != null) return false;
+ return slidingInterval != null ? slidingInterval.equals(that.slidingInterval) : that.slidingInterval == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = super.hashCode();
+ result = 31 * result + (windowLength != null ? windowLength.hashCode() : 0);
+ result = 31 * result + (slidingInterval != null ? slidingInterval.hashCode() : 0);
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/windowing/TumblingWindows.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/windowing/TumblingWindows.java b/storm-core/src/jvm/org/apache/storm/streams/windowing/TumblingWindows.java
new file mode 100644
index 0000000..71ed617
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/windowing/TumblingWindows.java
@@ -0,0 +1,119 @@
+/**
+ * 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.storm.streams.windowing;
+
+import org.apache.storm.topology.base.BaseWindowedBolt;
+
+import static org.apache.storm.topology.base.BaseWindowedBolt.Count;
+import static org.apache.storm.topology.base.BaseWindowedBolt.Duration;
+
+/**
+ * A tumbling window specification. The window tumbles after the specified window length.
+ *
+ * @param <L> the type of the length of the window (e.g Count, Duration)
+ */
+public class TumblingWindows<L> extends BaseWindow<L, L> {
+ private final L windowLength;
+
+ private TumblingWindows(L windowLength) {
+ this.windowLength = windowLength;
+ }
+
+ @Override
+ public L getWindowLength() {
+ return windowLength;
+ }
+
+ @Override
+ public L getSlidingInterval() {
+ return windowLength;
+ }
+
+ /**
+ * A count based tumbling window.
+ *
+ * @param count the number of tuples after which the window tumbles
+ */
+ public static TumblingWindows<Count> of(Count count) {
+ return new TumblingWindows<>(count);
+ }
+
+ /**
+ * A time duration based tumbling window.
+ *
+ * @param duration the time duration after which the window tumbles
+ */
+ public static TumblingWindows<Duration> of(Duration duration) {
+ return new TumblingWindows<>(duration);
+ }
+
+ /**
+ * The name of the field in the tuple that contains the timestamp when the event occurred as a long value.
+ * This is used of event-time based processing. If this config is set and the field is not present in the incoming tuple,
+ * an {@link IllegalArgumentException} will be thrown.
+ *
+ * @param fieldName the name of the field that contains the timestamp
+ */
+ public TumblingWindows<L> withTimestampField(String fieldName) {
+ timestampField = fieldName;
+ return this;
+ }
+
+ /**
+ * Specify a stream id on which late tuples are going to be emitted. They are going to be accessible via the
+ * {@link org.apache.storm.topology.WindowedBoltExecutor#LATE_TUPLE_FIELD} field.
+ * It must be defined on a per-component basis, and in conjunction with the
+ * {@link BaseWindowedBolt#withTimestampField}, otherwise {@link IllegalArgumentException} will be thrown.
+ *
+ * @param streamId the name of the stream used to emit late tuples on
+ */
+ public TumblingWindows<L> withLateTupleStream(String streamId) {
+ lateTupleStream = streamId;
+ return this;
+ }
+
+ /**
+ * Specify the maximum time lag of the tuple timestamp in milliseconds. It means that the tuple timestamps
+ * cannot be out of order by more than this amount.
+ *
+ * @param duration the max lag duration
+ */
+ public TumblingWindows<L> withLag(Duration duration) {
+ lag = duration;
+ return this;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ if (!super.equals(o)) return false;
+
+ TumblingWindows<?> that = (TumblingWindows<?>) o;
+
+ return windowLength != null ? windowLength.equals(that.windowLength) : that.windowLength == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = super.hashCode();
+ result = 31 * result + (windowLength != null ? windowLength.hashCode() : 0);
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/streams/windowing/Window.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/streams/windowing/Window.java b/storm-core/src/jvm/org/apache/storm/streams/windowing/Window.java
new file mode 100644
index 0000000..40378e3
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/streams/windowing/Window.java
@@ -0,0 +1,70 @@
+/**
+ * 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.storm.streams.windowing;
+
+import java.io.Serializable;
+
+import static org.apache.storm.topology.base.BaseWindowedBolt.Duration;
+
+/**
+ * The window specification within {@link org.apache.storm.streams.Stream}.
+ *
+ * @param <L> the type of window length parameter (E.g. Count, Duration)
+ * @param <I> the type of the sliding interval parameter (E.g. Count, Duration)
+ */
+public interface Window<L, I> extends Serializable {
+
+ /**
+ * The length of the window.
+ *
+ * @return the window length
+ */
+ L getWindowLength();
+
+ /**
+ * The sliding interval of the window.
+ *
+ * @return the sliding interval
+ */
+ I getSlidingInterval();
+
+ /**
+ * The name of the field in the tuple that contains the timestamp when the event occurred as a long value.
+ * This is used of event-time based processing. If this config is set and the field is not present in the incoming tuple,
+ * an {@link IllegalArgumentException} will be thrown.
+ *
+ * @return the timestamp field.
+ */
+ String getTimestampField();
+
+ /**
+ * The name of the stream where late arriving tuples should be emitted. If this is not provided,
+ * the late tuples would be discarded.
+ *
+ * @return the name of the stream used to emit late tuples on
+ */
+ String getLateTupleStream();
+
+ /**
+ * The maximum time lag of the tuple timestamp in milliseconds. It means that the tuple timestamps
+ * cannot be out of order by more than this amount.
+ *
+ * @return the lag
+ */
+ Duration getLag();
+}