Support reactive in org.springframework.cloud.function.adapter.gcp.GcfJarLauncher #1259
Description
I have a question, I've been trying to use Mono<Message> in a function, but I get the following error:
2025-04-07T21:17:47.156-06:00 INFO 53937 --- [demo-cloud-function-http-spring-boot] [ main] c.g.c.functions.invoker.runner.Invoker : Serving function...
2025-04-07T21:17:47.156-06:00 INFO 53937 --- [demo-cloud-function-http-spring-boot] [ main] c.g.c.functions.invoker.runner.Invoker : Function: org.springframework.cloud.function.adapter.gcp.GcfJarLauncher
2025-04-07T21:17:47.156-06:00 INFO 53937 --- [demo-cloud-function-http-spring-boot] [ main] c.g.c.functions.invoker.runner.Invoker : URL: http://localhost:8080/
2025-04-07T21:18:09.902-06:00 ERROR 53937 --- [demo-cloud-function-http-spring-boot] [qtp644163395-41] com.google.cloud.functions.invoker : Failed to execute org.springframework.cloud.function.adapter.gcp.GcfJarLauncher
java.lang.ClassCastException: class reactor.core.publisher.MonoMapFuseable cannot be cast to class org.springframework.messaging.Message (reactor.core.publisher.MonoMapFuseable and org.springframework.messaging.Message are in unnamed module of loader com.google.cloud.functions.invoker.runner.Invoker$FunctionClassLoader @5d28ac23)
at org.springframework.cloud.function.adapter.gcp.FunctionInvoker.service(FunctionInvoker.java:120) ~[spring-cloud-function-adapter-gcp-4.1.5.jar:4.1.5]
at org.springframework.cloud.function.adapter.gcp.GcfJarLauncher.service(GcfJarLauncher.java:53) ~[spring-cloud-function-adapter-gcp-4.1.5.jar:4.1.5]
at com.google.cloud.functions.invoker.HttpFunctionExecutor.service(HttpFunctionExecutor.java:68) ~[na:na]
at javax.servlet.http.HttpServlet.service(HttpServlet.java:750) ~[java-function-invoker-1.3.1.jar:1.3.1]
at org.eclipse.jetty.servlet.ServletHolder.handle(ServletHolder.java:799) ~[na:na]
at org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:554) ~[na:na]
at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233) ~[na:na]
at org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1440) ~[na:na]
at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:188) ~[na:na]
at org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:505) ~[na:na]
at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:186) ~[na:na]
at org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1355) ~[na:na]
at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141) ~[na:na]
at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127) ~[na:na]
at com.google.cloud.functions.invoker.runner.Invoker$NotFoundHandler.handle(Invoker.java:478) ~[na:na]
at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127) ~[na:na]
at org.eclipse.jetty.server.Server.handle(Server.java:516) ~[na:na]
at org.eclipse.jetty.server.HttpChannel.lambda$handle$1(HttpChannel.java:487) ~[na:na]
at org.eclipse.jetty.server.HttpChannel.dispatch(HttpChannel.java:732) ~[na:na]
at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:479) ~[na:na]
at org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:277) ~[na:na]
at org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:311) ~[na:na]
at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:105) ~[na:na]
at org.eclipse.jetty.io.ChannelEndPoint$1.run(ChannelEndPoint.java:104) ~[na:na]
at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:883) ~[na:na]
at org.eclipse.jetty.util.thread.QueuedThreadPool$Runner.run(QueuedThreadPool.java:1034) ~[na:na]
at java.base/java.lang.Thread.run(Thread.java:840) ~[na:na]
my code is:
package com.geezylucas.democloudfunctionhttpspringboot.config;
import com.geezylucas.democloudfunctionhttpspringboot.dto.ProductResponseDTO;
import com.geezylucas.democloudfunctionhttpspringboot.dto.PubSubBodyDTO;
import com.google.cloud.spring.pubsub.core.publisher.PubSubPublisherTemplate;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.function.adapter.gcp.FunctionInvoker;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.function.Function;
@Slf4j
@Configuration
@RequiredArgsConstructor
public class FunctionConfig {
private final WebClient webClient;
private final PubSubPublisherTemplate pubSubPublisherTemplate;
@Bean
public Function<Mono<PubSubBodyDTO>, Mono<Message<String>>> function() {
return input -> input
.doOnNext(it -> log.info("Function invoked with message: {}", it))
.map(pubSubBodyDTO -> new String(Base64.getDecoder().decode(pubSubBodyDTO.message().data()), StandardCharsets.UTF_8))
.doOnNext(s -> log.info("Received Pub/Sub message: {}", s))
.flatMap(s -> webClient
.get()
.uri(uriBuilder -> uriBuilder
.path("/products/{id}")
.build(s))
.retrieve()
.bodyToMono(ProductResponseDTO.class)
)
.flatMap(productResponseDTO -> Mono.fromFuture(pubSubPublisherTemplate.publish("my-function-topic", productResponseDTO.title())))
.map(title -> MessageBuilder
.withPayload(title)
.setHeader(FunctionInvoker.HTTP_STATUS_CODE, 200)
.build()
);
}
}my pom.xml is:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>3.3.10</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.geezylucas</groupId>
<artifactId>demo-cloud-function-http-spring-boot</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>demo-cloud-function-http-spring-boot</name>
<description>demo-cloud-function-http-spring-boot</description>
<properties>
<java.version>17</java.version>
<spring-cloud-gcp.version>5.11.1</spring-cloud-gcp.version>
<spring-cloud.version>2023.0.5</spring-cloud.version>
<spring-cloud-function-adapter-gcp.version>4.1.5</spring-cloud-function-adapter-gcp.version>
<function-maven-plugin.version>0.11.1</function-maven-plugin.version>
<java-function-invoker.version>1.4.1</java-function-invoker.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-function-webflux</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-function-adapter-gcp</artifactId>
</dependency>
<dependency>
<groupId>com.google.cloud</groupId>
<artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.google.cloud.functions.invoker</groupId>
<artifactId>java-function-invoker</artifactId>
<version>${java-function-invoker.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>com.google.cloud</groupId>
<artifactId>spring-cloud-gcp-dependencies</artifactId>
<version>${spring-cloud-gcp.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<plugins>
<plugin>
<artifactId>maven-deploy-plugin</artifactId>
<configuration>
<skip>true</skip>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<annotationProcessorPaths>
<path>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
</path>
<path>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</path>
</annotationProcessorPaths>
</configuration>
</plugin>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<outputDirectory>target/deploy</outputDirectory>
<excludes>
<exclude>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</exclude>
</excludes>
</configuration>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-function-adapter-gcp</artifactId>
<version>${spring-cloud-function-adapter-gcp.version}</version>
</dependency>
</dependencies>
</plugin>
<plugin>
<groupId>com.google.cloud.functions</groupId>
<artifactId>function-maven-plugin</artifactId>
<version>${function-maven-plugin.version}</version>
<configuration>
<functionTarget>org.springframework.cloud.function.adapter.gcp.GcfJarLauncher</functionTarget>
<port>8080</port>
</configuration>
</plugin>
</plugins>
</build>
</project>I'm wondering if spring-cloud-function-adapter-gcp has support for using the reactive paradigm
Thanks for your answers.