From 61ae36350c5f99200e81311f394e3a007ea694df Mon Sep 17 00:00:00 2001
From: Jan Lukavsky <je.ik@seznam.cz>
Date: Mon, 9 Oct 2023 15:39:26 +0200
Subject: [PATCH] [proxima-tools] add type hints to WindowedStream

---
 .../proxima/tools/groovy/WindowedStream.java  | 31 ++++++++++++++-----
 1 file changed, 23 insertions(+), 8 deletions(-)

diff --git a/tools/src/main/java/cz/o2/proxima/tools/groovy/WindowedStream.java b/tools/src/main/java/cz/o2/proxima/tools/groovy/WindowedStream.java
index cb75b4d42..69355c0d7 100644
--- a/tools/src/main/java/cz/o2/proxima/tools/groovy/WindowedStream.java
+++ b/tools/src/main/java/cz/o2/proxima/tools/groovy/WindowedStream.java
@@ -541,36 +541,51 @@ WindowedStream<T> distinct(
   // overrides of Stream methods with fixed return types
 
   @Override
-  default <X> WindowedStream<X> flatMap(Closure<Iterable<X>> mapper) {
+  default <X> WindowedStream<X> flatMap(
+      @ClosureParams(value = FromString.class, options = "T") Closure<Iterable<X>> mapper) {
     return flatMap(null, mapper);
   }
 
   @Override
-  <X> WindowedStream<X> flatMap(@Nullable String name, Closure<Iterable<X>> mapper);
+  <X> WindowedStream<X> flatMap(
+      @Nullable String name,
+      @ClosureParams(value = FromString.class, options = "T") Closure<Iterable<X>> mapper);
 
   @Override
-  default <X> WindowedStream<X> map(Closure<X> mapper) {
+  default <X> WindowedStream<X> map(
+      @ClosureParams(value = FromString.class, options = "T") Closure<X> mapper) {
+
     return map(null, mapper);
   }
 
   @Override
-  <X> WindowedStream<X> map(@Nullable String name, Closure<X> mapper);
+  <X> WindowedStream<X> map(
+      @Nullable String name,
+      @ClosureParams(value = FromString.class, options = "T") Closure<X> mapper);
 
   @Override
-  default WindowedStream<T> filter(Closure<Boolean> predicate) {
+  default WindowedStream<T> filter(
+      @ClosureParams(value = FromString.class, options = "T") Closure<Boolean> predicate) {
+
     return filter(null, predicate);
   }
 
   @Override
-  WindowedStream<T> filter(@Nullable String name, Closure<Boolean> predicate);
+  WindowedStream<T> filter(
+      @Nullable String name,
+      @ClosureParams(value = FromString.class, options = "T") Closure<Boolean> predicate);
 
   @Override
-  default WindowedStream<T> assignEventTime(Closure<Long> assigner) {
+  default WindowedStream<T> assignEventTime(
+      @ClosureParams(value = FromString.class, options = "T") Closure<Long> assigner) {
+
     return assignEventTime(null, assigner);
   }
 
   @Override
-  WindowedStream<T> assignEventTime(@Nullable String name, Closure<Long> assigner);
+  WindowedStream<T> assignEventTime(
+      @Nullable String name,
+      @ClosureParams(value = FromString.class, options = "T") Closure<Long> assigner);
 
   @Override
   default WindowedStream<Pair<Object, T>> withWindow() {