Samza operator v2

samza-api/src/main/java/org/apache/samza/operators/MessageStream.java

@@ -27,7 +27,6 @@
 import org.apache.samza.operators.functions.SinkFunction;
 import org.apache.samza.operators.windows.Window;
 import org.apache.samza.operators.windows.WindowOutput;
-import org.apache.samza.operators.windows.WindowState;
 
 import java.util.Collection;
 
@@ -37,7 +36,7 @@
  * <p>
  * A {@link MessageStream} can be transformed into another {@link MessageStream} by applying the transforms in this API.
  *
- * @param <M>  type of {@link MessageEnvelope}s in this stream
+ * @param <M> type of {@link MessageEnvelope}s in this stream
  */
 @InterfaceStability.Unstable
 public interface MessageStream<M extends MessageEnvelope> {
@@ -90,14 +89,16 @@ public interface MessageStream<M extends MessageEnvelope> {
    * Use the {@link org.apache.samza.operators.windows.Windows} helper methods to create the appropriate windows.
    *
    * @param window  the {@link Window} to group and process {@link MessageEnvelope}s from this {@link MessageStream}
+   * @param <K>  the type of key the {@link Window} is computed on.
    * @param <WK>  the type of key in the {@link WindowOutput} from the {@link Window}
    * @param <WV>  the type of value in the {@link WindowOutput} from the {@link Window}
-   * @param <WS>  the type of window state kept in the {@link Window}
    * @param <WM>  the type of {@link WindowOutput} in the transformed {@link MessageStream}
    * @return  the transformed {@link MessageStream}
    */
-  <WK, WV, WS extends WindowState<WV>, WM extends WindowOutput<WK, WV>> MessageStream<WM> window(
-      Window<M, WK, WV, WM> window);
+
+  <K, WK, WV, WM extends WindowOutput<WK, WV>> MessageStream<WM> window(
+      Window<M, K, WK, WV, WM> window);
+
 
   /**
    * Joins this {@link MessageStream} with another {@link MessageStream} using the provided pairwise {@link JoinFunction}.

samza-api/src/main/java/org/apache/samza/operators/windows/BaseWindow.java

@@ -0,0 +1,100 @@
+/*
+ * 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.samza.operators.windows;
+
+import org.apache.samza.operators.data.MessageEnvelope;
+
+import java.util.List;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+
+/**
+ *  The base class of all types of {@link Window}s. Sub-classes can specify the default triggering semantics
+ *  for the {@link Window}, semantics for emission of early or late results and whether to accumulate or discard
+ *  previous results.
+ */
+
+public class BaseWindow<M extends MessageEnvelope, K, WV> implements Window<M, K, WindowKey<K>, WV, WindowOutput<WindowKey<K>, WV>> {
+
+  /**
+   * Defines the default triggering semantics for the {@link Window}.
+   */
+  private final List<TriggersBuilder.Trigger> defaultTriggers;
+
+
+  /**
+   * Defines the triggering semantics for emission of early or late results.
+   */
+  private List<TriggersBuilder.Trigger> earlyTriggers;
+  private List<TriggersBuilder.Trigger> lateTriggers;
+
+  /**
+   * Defines the fold function that is applied each time a {@link MessageEnvelope} is added to this window.
+   */
+  private final BiFunction<M, WV, WV> aggregator;
+
+  /*
+   * Defines the function that extracts the event time from a {@link MessageEnvelope}
+   */
+  private final Function<M, Long> eventTimeExtractor;
+
+  /*
+   * Defines the function that extracts the key from a {@link MessageEnvelope}
+   */
+  private final Function<M, K> keyExtractor;
+
+
+  public BaseWindow(Function<M, K> keyExtractor, BiFunction<M, WV, WV> aggregator, Function<M, Long> eventTimeExtractor, List<TriggersBuilder.Trigger> defaultTriggers) {
+    this.aggregator = aggregator;
+    this.eventTimeExtractor = eventTimeExtractor;
+    this.keyExtractor = keyExtractor;
+    this.defaultTriggers = defaultTriggers;
+  }
+
+  @Override
+  public Window<M, K, WindowKey<K>, WV, WindowOutput<WindowKey<K>, WV>> setTriggers(TriggersBuilder.Triggers wndTrigger) {
+    this.earlyTriggers = wndTrigger.getEarlyTriggers();
+    this.lateTriggers = wndTrigger.getLateTriggers();
+    return this;
+  }
+
+  public List<TriggersBuilder.Trigger> getDefaultTriggers() {
+    return defaultTriggers;
+  }
+
+  public List<TriggersBuilder.Trigger> getEarlyTriggers() {
+    return earlyTriggers;
+  }
+
+  public List<TriggersBuilder.Trigger> getLateTriggers() {
+    return lateTriggers;
+  }
+
+  public BiFunction<M, WV, WV> getAggregator() {
+    return aggregator;
+  }
+
+  public Function<M, Long> getEventTimeExtractor() {
+    return eventTimeExtractor;
+  }
+
+  public Function<M, K> getKeyExtractor() {
+    return keyExtractor;
+  }
+}