init

fn-rabbitmq/src/main/java/net/yaoyi/pipeline/rabbitmq/RabbitConsumer.java

@@ -2,6 +2,7 @@ package net.yaoyi.pipeline.rabbitmq;
 
 import com.aliyun.fc.runtime.Context;
 import com.aliyun.fc.runtime.PojoRequestHandler;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import lombok.extern.slf4j.Slf4j;
@@ -21,6 +22,8 @@ public class RabbitConsumer implements PojoRequestHandler<Map<String, Object>, V
 
     private static final String PIPELINE_API = "http://1861991840278303.eventbridge.cn-beijing.aliyuncs.com/webhook/putEvents?token=62ed79329bb74d0d8ede5cc5d0487a45897a6ff99f5d4d61946daf9f6e9c7fbc766b9d55ef2241edb2c2cb3aee651ac101409d2783f341aa8e8d27fdb9869d86";
 
+    private static final int DEFAULT_WAIT_SECONDS = 2 * 60;
+
     private static final OkHttpClient httpClient = new OkHttpClient.Builder()
             .connectTimeout(10, TimeUnit.SECONDS)
             .readTimeout(30, TimeUnit.SECONDS)
@@ -31,19 +34,19 @@ public class RabbitConsumer implements PojoRequestHandler<Map<String, Object>, V
     @Override
     public Void handleRequest(Map<String, Object> input, Context context) {
         log.info("函数计算启动, Request ID: {}, data:{}", context.getRequestId(), JacksonUtils.toJson(input));
-        Map<String, Object> payload = (Map<String, Object>) input.get("payload");
-        String queueName = payload.get("queueName").toString();
+        JsonNode payload = JacksonUtils.toJsonNode(input.get("payload").toString());
+        String queueName = payload.get("queueName").asText();
         try (Connection connection = RabbitMQHelper.createConnection();
              Channel channel = RabbitMQHelper.createChannel(connection, queueName)) {
 
             startConsumer(channel,queueName);
             log.info("RabbitMQ 消费者启动成功");
 
-            Object val = payload.get("waitSeconds");
-            if (val instanceof Integer) {
-                Thread.sleep(((int) val) * 1000);
+            long waitSeconds ;
+            if (payload.has("waitSeconds") && (waitSeconds = payload.get("waitSeconds").asLong()) > 0) {
+                Thread.sleep(waitSeconds * 1000);
             } else {
-                Thread.sleep(2 * 60 * 1000);
+                Thread.sleep(DEFAULT_WAIT_SECONDS * 1000);
             }
         } catch (Exception e) {
             log.error("启动 RabbitMQ 消费者失败", e);