Fix docs with regard to Flux.interval
Also provide ability to run with `Flux.interval(Duration)` in compiled scripts Fixes gh-174
This commit is contained in:
@@ -182,7 +182,7 @@ curl -X POST -H "Content-Type: text/plain" -d foo localhost:9002/print
|
|||||||
First register a streaming words supplier:
|
First register a streaming words supplier:
|
||||||
|
|
||||||
----
|
----
|
||||||
./registerSupplier.sh -n wordstream -f "()->Flux.intervalMillis(1000).map(i->\"message-\"+i)"
|
./registerSupplier.sh -n wordstream -f "()->Flux.interval(Duration.ofMillis(1000)).map(i->\"message-\"+i)"
|
||||||
----
|
----
|
||||||
|
|
||||||
Then start the source (supplier), processor (function), and sink (consumer) apps
|
Then start the source (supplier), processor (function), and sink (consumer) apps
|
||||||
|
|||||||
@@ -122,7 +122,7 @@ curl -X POST -H "Content-Type: text/plain" -d foo localhost:9002/print
|
|||||||
First register a streaming words supplier:
|
First register a streaming words supplier:
|
||||||
|
|
||||||
----
|
----
|
||||||
./registerSupplier.sh -n wordstream -f "()->Flux.intervalMillis(1000).map(i->\"message-\"+i)"
|
./registerSupplier.sh -n wordstream -f "()->Flux.interval(Duration.ofMillis(1000)).map(i->\"message-\"+i)"
|
||||||
----
|
----
|
||||||
|
|
||||||
Then start the source (supplier), processor (function), and sink (consumer) apps
|
Then start the source (supplier), processor (function), and sink (consumer) apps
|
||||||
|
|||||||
@@ -45,7 +45,7 @@ while getopts ":i:s:f:c:o:p:d:" opt; do
|
|||||||
esac
|
esac
|
||||||
done
|
done
|
||||||
|
|
||||||
java -jar ../spring-cloud-function-samples/spring-cloud-function-sample-compiler/target/function-sample-compiler-1.0.0.BUILD-SNAPSHOT.jar\
|
java -jar ../spring-cloud-function-samples/function-sample-compiler/target/function-sample-compiler-1.0.0.BUILD-SNAPSHOT.jar\
|
||||||
--management.security.enabled=false\
|
--management.security.enabled=false\
|
||||||
--server.port=$PORT\
|
--server.port=$PORT\
|
||||||
--spring.cloud.function.stream.endpoint=$FUNC\
|
--spring.cloud.function.stream.endpoint=$FUNC\
|
||||||
|
|||||||
@@ -53,6 +53,7 @@ public abstract class AbstractFunctionCompiler<F> {
|
|||||||
private static String SOURCE_CODE_TEMPLATE = "package "
|
private static String SOURCE_CODE_TEMPLATE = "package "
|
||||||
+ AbstractFunctionCompiler.class.getPackage().getName() + ";\n"
|
+ AbstractFunctionCompiler.class.getPackage().getName() + ";\n"
|
||||||
+ "import java.util.*;\n" // Helpful to include this
|
+ "import java.util.*;\n" // Helpful to include this
|
||||||
|
+ "import java.time.*;\n" // Helpful to include this
|
||||||
+ "import java.util.function.*;\n"
|
+ "import java.util.function.*;\n"
|
||||||
+ "import reactor.core.publisher.Flux;\n"
|
+ "import reactor.core.publisher.Flux;\n"
|
||||||
+ "public class %s implements CompilationResultFactory<%s> {\n"
|
+ "public class %s implements CompilationResultFactory<%s> {\n"
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ package org.springframework.cloud.function.compiler;
|
|||||||
import java.util.function.Supplier;
|
import java.util.function.Supplier;
|
||||||
|
|
||||||
import org.junit.Test;
|
import org.junit.Test;
|
||||||
|
import reactor.core.publisher.Flux;
|
||||||
|
|
||||||
import org.springframework.cloud.function.core.FunctionFactoryUtils;
|
import org.springframework.cloud.function.core.FunctionFactoryUtils;
|
||||||
|
|
||||||
@@ -47,4 +48,13 @@ public class SupplierCompilerTests {
|
|||||||
assertThat(compiled.getResult().get()).isEqualTo("foo");
|
assertThat(compiled.getResult().get()).isEqualTo("foo");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void supppliesFluxStreamString() {
|
||||||
|
CompiledFunctionFactory<Supplier<Flux<String>>> compiled = new SupplierCompiler<Flux<String>>(
|
||||||
|
String.class.getName()).compile("foos",
|
||||||
|
"() -> Flux.interval(Duration.ofMillis(1000)).map(Object::toString)",
|
||||||
|
"Flux<String>");
|
||||||
|
assertThat(FunctionFactoryUtils.isFluxSupplier(compiled.getFactoryMethod())).isTrue();
|
||||||
|
assertThat(compiled.getResult().get().blockFirst()).isEqualTo("0");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user