HTTPingresstypeApacheFlinkStateFun

基于 Flink StateFun 架构中的 HTTP Ingress 类型，可以通过以下代码示例解决该问题。

import io.github.fstatefun.sdk.io.IngressIdentifier;
import io.github.fstatefun.sdk.statefun.StatefulFunction;
import io.github.fstatefun.sdk.statefun.StatefulFunctionModule;
import io.github.fstatefun.sdk.statefun.ValueSpec;
import io.github.fstatefun.sdk.statefun.flink.EgressBuilder;
import io.github.fstatefun.sdk.statefun.flink.IngressBuilder;
import io.github.fstatefun.sdk.statefun.flink.core.StatefulFunctions;

import java.util.Map;

public class MyFunction implements StatefulFunction {

    public static final IngressIdentifier<String> INGRESS_ID =
        new IngressIdentifier<>(String.class, "namespace", "ingress-name");

    public void invoke(Context context, Object input) {

        if (input instanceof HttpRequest) {
            HttpRequest request = (HttpRequest) input;
            // 处理 HTTP 响应
            HttpResponse response = HttpResponse.ok();
            context.send("egress-name", response.toByteString());
        }
    }

    public static final class Module implements StatefulFunctionModule {

        @Override
        public void configure(Map<String, String> globalConfiguration, Binder binder) {

            StatefulFunctions functions = new StatefulFunctions();

            //创建 IngressBuilder
            IngressBuilder<String> ingressBuilder = IngressBuilder.<String>newBuilder(INGRESS_ID)
                .withTypeString("http")
                .withDeserializer(BytesDeserializer.INSTANCE);

            // 创建 EgressBuilder
            EgressBuilder<MyProtos.MyType> egressBuilder = FlinkEgressBuilder.<MyProtos.MyType>forEgress("egress-name")
                .withSerializer(MySerializer.INSTANCE);

            // 绑定函数
            functions.bindIngress(ingressBuilder.withConsumer(MyFunction::new));
            functions.bindEgress(egressBuilder);

            // 绑定值类型
            binder.bindIngressRouter(INGRESS_ID, new MyRouter());
            binder.bindFunctionProvider(MyFunction.class, unused -> MyFunction::new);
            binder.bindEgressCollector("egress-name", new MyTypeEgress());
            binder.bind