RabbitMQ 订阅 Demo

Node.js

依赖：npm install amqplib。运行：node consumer.js。

// consumer.js — PetaX RabbitMQ 订阅 Demo (Node.js, amqplib)
// 运行：npm install amqplib && node consumer.js
// 注意：字段形状是基于 HTTP 回调 DTO 的假设，未经真实 PetaX 消息验证，
// 尤其 notify_issue_result 的 id 字段可能是 issueId 而非 roundId，接入前务必用真实消息核对。
const fs = require('fs');
const amqp = require('amqplib');

const HOST = process.env.RABBITMQ_HOST || 'localhost';
const PORT = process.env.RABBITMQ_PORT || '5671';
const VHOST = process.env.RABBITMQ_VHOST || '/';
const USERNAME = process.env.RABBITMQ_USERNAME || 'guest';
const PASSWORD = process.env.RABBITMQ_PASSWORD || 'guest';
const CA_CERT = process.env.RABBITMQ_CA_CERT || 'ca_certificate.pem';
const CLIENT_CERT = process.env.RABBITMQ_CLIENT_CERT || 'client_certificate.pem';
const CLIENT_KEY = process.env.RABBITMQ_CLIENT_KEY || 'client_key.pem';
const SERVER_NAME = process.env.RABBITMQ_SERVER_NAME || '';

// 三个队列名默认值是官方文档已公开的示例格式；真实队列名由 PetaX 分配，
// 务必以 PetaX 提供的为准，不要假设跟这里的示例值相同。
const QUEUE_ROLES = {
  [process.env.RABBITMQ_QUEUE_SETTLED_WIN || 'staging.10102.merchant102_server.settled_win.q']: 'settled_ticket',
  [process.env.RABBITMQ_QUEUE_SETTLED_LOSS || 'staging.10102.merchant102_server.settled_loss.q']: 'settled_ticket',
  [process.env.RABBITMQ_QUEUE_ISSUE_RESULT || 'staging.10102.merchant102_server.notify_issue_result.q']: 'settled_round',
};

// 仅用于 Demo：记录已经"模拟失败过一次"的 key（ticketId 或 roundId），
// 第二次投递（redelivered=true）就放行成功。真实商户不需要这个逻辑，
// 这里只是让 demo 能演示 nack+requeue 而不会无限热循环重投同一条坏消息。
const failedOnce = new Set();

function handleMessage(channel, msg, role) {
  const redelivered = msg.fields.redelivered;
  let payload;
  try {
    payload = JSON.parse(msg.content.toString());
  } catch (e) {
    // 1. 解析失败 = 永久错误：ack 丢弃，避免毒消息卡队列。
    console.log(`[bad-json] discarding, redelivered=${redelivered}`);
    channel.ack(msg);
    return;
  }

  let key;
  if (role === 'settled_ticket') {
    // settled_win/settled_loss 假定按 SettledTicketRequest 字段解析。
    key = payload.ticketId;
    console.log(`[settled_ticket] ticketId=${key} status=${payload.status} ccy=${payload.ccy} redelivered=${redelivered}`);
  } else {
    // notify_issue_result 假定按 SettledRoundRequest 字段解析；roundId 字段名
    // 是本仓库运行手册标注的最大不确定性来源，真实接入前务必用真实消息核对。
    key = payload.roundId;
    console.log(`[settled_round] roundId=${key} result=${payload.result} redelivered=${redelivered}`);
  }

  if (!key) {
    console.log('[warn] missing key field, discarding (ack)');
    channel.ack(msg);
    return;
  }

  // 2. 第一次见到这个 key：模拟一次瞬时处理失败，nack + requeue。
  //    只失败一次，避免同一条消息被无限重投形成热循环。
  if (!failedOnce.has(key)) {
    failedOnce.add(key);
    console.log(`[simulate-failure] ${key} -> nack+requeue`);
    channel.nack(msg, false, true);
    return;
  }

  // 3. 第二次投递（redelivered=true）：放行成功，ack。
  console.log(`[success] ${key} -> ack`);
  channel.ack(msg);
}

async function main() {
  // 注意：amqplib 的 connect(url, socketOptions) 只从第一个参数里挑
  // host/port/servername 拷进 socket 层，ca/cert/key 等 TLS 材料必须放进
  // 第二个参数（socketOptions），否则会静默走系统信任库、连不上 mTLS。
  const connection = await amqp.connect(
    {
      protocol: 'amqps',
      hostname: HOST,
      port: parseInt(PORT, 10),
      vhost: VHOST,
      username: USERNAME,
      password: PASSWORD,
    },
    {
      ca: [fs.readFileSync(CA_CERT)],
      cert: fs.readFileSync(CLIENT_CERT),
      key: fs.readFileSync(CLIENT_KEY),
      servername: SERVER_NAME || undefined,
    }
  );
  const channel = await connection.createChannel();
  await channel.prefetch(parseInt(process.env.RABBITMQ_PREFETCH || '10', 10));

  for (const [queue, role] of Object.entries(QUEUE_ROLES)) {
    // checkQueue 只检查队列存在，不主动创建——真实队列由 PetaX 预先建好。
    await channel.checkQueue(queue);
    channel.consume(queue, (msg) => handleMessage(channel, msg, role), { noAck: false });
  }

  console.log(`listening on ${Object.keys(QUEUE_ROLES).join(', ')}`);
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

Java

依赖：Maven（pom.xml 见下）+ 需要把 PetaX 提供的证书/私钥转换成 PKCS12（openssl pkcs12 -export ...，命令见代码注释）。运行：mvn compile exec:java。

<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.example.petax</groupId>
  <artifactId>rabbitmq-subscribe-demo</artifactId>
  <version>1.0.0</version>
  <packaging>jar</packaging>
  <properties>
    <maven.compiler.source>17</maven.compiler.source>
    <maven.compiler.target>17</maven.compiler.target>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>
  <dependencies>
    <dependency>
      <groupId>com.rabbitmq</groupId>
      <artifactId>amqp-client</artifactId>
      <version>5.21.0</version>
    </dependency>
    <dependency>
      <groupId>com.fasterxml.jackson.core</groupId>
      <artifactId>jackson-databind</artifactId>
      <version>2.17.2</version>
    </dependency>
  </dependencies>
  <build>
    <plugins>
      <plugin>
        <groupId>org.codehaus.mojo</groupId>
        <artifactId>exec-maven-plugin</artifactId>
        <version>3.1.0</version>
        <configuration>
          <mainClass>com.example.petax.RabbitmqConsumer</mainClass>
        </configuration>
      </plugin>
    </plugins>
  </build>
</project>

Maven 标准目录结构，消费者源码放在 src/main/java/com/example/petax/RabbitmqConsumer.java：

package com.example.petax;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLParameters;
import javax.net.ssl.SSLSocket;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManagerFactory;
import java.io.FileInputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.security.KeyStore;
import java.security.cert.Certificate;
import java.security.cert.CertificateFactory;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

// consumer.java — PetaX RabbitMQ 订阅 Demo (Java, com.rabbitmq:amqp-client)
// 运行：mvn compile exec:java
// 注意：字段形状是基于 HTTP 回调 DTO 的假设，未经真实 PetaX 消息验证，
// 尤其 notify_issue_result 的 id 字段可能是 issueId 而非 roundId，接入前务必用真实消息核对。
public class RabbitmqConsumer {

    private static final ObjectMapper MAPPER = new ObjectMapper();

    // 仅用于 Demo：记录已经"模拟失败过一次"的 key（ticketId 或 roundId），
    // 第二次投递（redelivered=true）就放行成功。真实商户不需要这个逻辑，
    // 这里只是让 demo 能演示 nack+requeue 而不会无限热循环重投同一条坏消息。
    private static final Set<String> FAILED_ONCE = ConcurrentHashMap.newKeySet();

    public static void main(String[] args) throws Exception {
        String host = getenv("RABBITMQ_HOST", "localhost");
        int port = Integer.parseInt(getenv("RABBITMQ_PORT", "5671"));
        String vhost = getenv("RABBITMQ_VHOST", "/");
        String username = getenv("RABBITMQ_USERNAME", "guest");
        String password = getenv("RABBITMQ_PASSWORD", "guest");
        String caCert = getenv("RABBITMQ_CA_CERT", "ca_certificate.pem");
        String serverName = getenv("RABBITMQ_SERVER_NAME", "");
        // Java 的 TLS API 需要 PKCS12 keystore，不能直接读 PEM 私钥；
        // 用 openssl 把 PetaX 给的证书/私钥转换一次：
        //   openssl pkcs12 -export -in client_certificate.pem -inkey client_key.pem -out client.p12 -passout pass:
        String clientKeystore = getenv("RABBITMQ_CLIENT_KEYSTORE", "client.p12");

        // 三个队列名默认值是官方文档已公开的示例格式；真实队列名由 PetaX 分配，
        // 务必以 PetaX 提供的为准，不要假设跟这里的示例值相同。
        Map<String, String> queueRoles = Map.of(
                getenv("RABBITMQ_QUEUE_SETTLED_WIN", "staging.10102.merchant102_server.settled_win.q"), "settled_ticket",
                getenv("RABBITMQ_QUEUE_SETTLED_LOSS", "staging.10102.merchant102_server.settled_loss.q"), "settled_ticket",
                getenv("RABBITMQ_QUEUE_ISSUE_RESULT", "staging.10102.merchant102_server.notify_issue_result.q"), "settled_round"
        );

        CertificateFactory cf = CertificateFactory.getInstance("X.509");
        Certificate ca;
        try (var in = new FileInputStream(caCert)) {
            ca = cf.generateCertificate(in);
        }
        KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
        trustStore.load(null, null);
        trustStore.setCertificateEntry("ca", ca);
        TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
        tmf.init(trustStore);

        KeyStore keyStore = KeyStore.getInstance("PKCS12");
        try (var in = new FileInputStream(clientKeystore)) {
            keyStore.load(in, "".toCharArray());
        }
        KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
        kmf.init(keyStore, "".toCharArray());

        SSLContext sslContext = SSLContext.getInstance("TLS");
        sslContext.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(host);
        factory.setPort(port);
        factory.setVirtualHost(vhost);
        factory.setUsername(username);
        factory.setPassword(password);
        // 如果用 IP 连接，但证书 CN/SAN 是 rabbitmq，可通过 RABBITMQ_SERVER_NAME
        // 指定 TLS 校验和 SNI 使用的服务名；TCP 仍连接 RABBITMQ_HOST。
        factory.setSocketFactory(new ServerNameSslSocketFactory((SSLSocketFactory) sslContext.getSocketFactory(), serverName));
        factory.enableHostnameVerification();

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.basicQos(Integer.parseInt(getenv("RABBITMQ_PREFETCH", "10")));

        for (Map.Entry<String, String> entry : queueRoles.entrySet()) {
            String queue = entry.getKey();
            String role = entry.getValue();
            // queueDeclarePassive 只检查队列存在，不主动创建——真实队列由 PetaX 预先建好。
            channel.queueDeclarePassive(queue);

            DeliverCallback callback = (consumerTag, delivery) -> {
                boolean redelivered = delivery.getEnvelope().isRedeliver();
                long tag = delivery.getEnvelope().getDeliveryTag();
                JsonNode payload;
                try {
                    payload = MAPPER.readTree(delivery.getBody());
                } catch (Exception e) {
                    // 1. 解析失败 = 永久错误：ack 丢弃，避免毒消息卡队列。
                    System.out.println("[bad-json] discarding, redelivered=" + redelivered);
                    channel.basicAck(tag, false);
                    return;
                }

                String key;
                if ("settled_ticket".equals(role)) {
                    // settled_win/settled_loss 假定按 SettledTicketRequest 字段解析。
                    key = payload.path("ticketId").asText(null);
                    System.out.println("[settled_ticket] ticketId=" + key + " status=" + payload.path("status").asText()
                            + " ccy=" + payload.path("ccy").asText() + " redelivered=" + redelivered);
                } else {
                    // notify_issue_result 假定按 SettledRoundRequest 字段解析；roundId
                    // 字段名是本仓库运行手册标注的最大不确定性来源，真实接入前务必用真实消息核对。
                    key = payload.path("roundId").asText(null);
                    System.out.println("[settled_round] roundId=" + key + " result=" + payload.path("result").asText()
                            + " redelivered=" + redelivered);
                }

                if (key == null) {
                    System.out.println("[warn] missing key field, discarding (ack)");
                    channel.basicAck(tag, false);
                    return;
                }

                // 2. 第一次见到这个 key：模拟一次瞬时处理失败，nack + requeue。
                //    只失败一次，避免同一条消息被无限重投形成热循环。
                if (FAILED_ONCE.add(key)) {
                    System.out.println("[simulate-failure] " + key + " -> nack+requeue");
                    channel.basicNack(tag, false, true);
                    return;
                }

                // 3. 第二次投递（redelivered=true）：放行成功，ack。
                System.out.println("[success] " + key + " -> ack");
                channel.basicAck(tag, false);
            };
            channel.basicConsume(queue, false, callback, consumerTag -> {});
        }

        System.out.println("listening on " + queueRoles.keySet());
        Thread.currentThread().join();
    }

    private static class ServerNameSslSocketFactory extends SSLSocketFactory {
        private final SSLSocketFactory delegate;
        private final String serverName;

        ServerNameSslSocketFactory(SSLSocketFactory delegate, String serverName) {
            this.delegate = delegate;
            this.serverName = serverName == null ? "" : serverName;
        }

        private String peerHost(String host) {
            return serverName.isBlank() ? host : serverName;
        }

        private Socket wrap(Socket raw, String host, int port) throws IOException {
            SSLSocket socket = (SSLSocket) delegate.createSocket(raw, peerHost(host), port, true);
            SSLParameters params = socket.getSSLParameters();
            params.setEndpointIdentificationAlgorithm("HTTPS");
            socket.setSSLParameters(params);
            return socket;
        }

        @Override public String[] getDefaultCipherSuites() { return delegate.getDefaultCipherSuites(); }
        @Override public String[] getSupportedCipherSuites() { return delegate.getSupportedCipherSuites(); }
        @Override public Socket createSocket() throws IOException { return delegate.createSocket(); }
        @Override public Socket createSocket(Socket s, String host, int port, boolean autoClose) throws IOException {
            SSLSocket socket = (SSLSocket) delegate.createSocket(s, peerHost(host), port, autoClose);
            SSLParameters params = socket.getSSLParameters();
            params.setEndpointIdentificationAlgorithm("HTTPS");
            socket.setSSLParameters(params);
            return socket;
        }
        @Override public Socket createSocket(String host, int port) throws IOException {
            return wrap(new Socket(host, port), host, port);
        }
        @Override public Socket createSocket(String host, int port, InetAddress localAddress, int localPort) throws IOException {
            return wrap(new Socket(host, port, localAddress, localPort), host, port);
        }
        @Override public Socket createSocket(InetAddress host, int port) throws IOException {
            return wrap(new Socket(host, port), host.getHostName(), port);
        }
        @Override public Socket createSocket(InetAddress address, int port, InetAddress localAddress, int localPort) throws IOException {
            return wrap(new Socket(address, port, localAddress, localPort), address.getHostName(), port);
        }
    }

    private static String getenv(String key, String def) {
        String v = System.getenv(key);
        return (v == null || v.isEmpty()) ? def : v;
    }
}

Go

依赖：go get github.com/rabbitmq/amqp091-go。运行：go run consumer.go。

// consumer.go — PetaX RabbitMQ 订阅 Demo (Go, rabbitmq/amqp091-go)
// 运行：go mod init demo && go get github.com/rabbitmq/amqp091-go && go run consumer.go
// 注意：字段形状是基于 HTTP 回调 DTO 的假设，未经真实 PetaX 消息验证，
// 尤其 notify_issue_result 的 id 字段可能是 issueId 而非 roundId，接入前务必用真实消息核对。
package main

import (
    "crypto/tls"
    "crypto/x509"
    "encoding/json"
    "log"
    "net"
    "net/url"
    "os"
    "strconv"
    "strings"
    "sync"

    amqp "github.com/rabbitmq/amqp091-go"
)

func getenv(k, def string) string {
    if v := os.Getenv(k); v != "" {
        return v
    }
    return def
}

// 仅用于 Demo：记录已经"模拟失败过一次"的 key（ticketId 或 roundId），
// 第二次投递（redelivered=true）就放行成功。真实商户不需要这个逻辑，
// 这里只是让 demo 能演示 nack+requeue 而不会无限热循环重投同一条坏消息。
// 3 个队列各自开一个 goroutine 消费，共享这个 map，必须加锁。
var (
    mu         sync.Mutex
    failedOnce = map[string]bool{}
)

func handleMessage(d amqp.Delivery, role string) {
    var payload map[string]interface{}
    if err := json.Unmarshal(d.Body, &payload); err != nil {
        // 1. 解析失败 = 永久错误：ack 丢弃，避免毒消息卡队列。
        log.Printf("[bad-json] discarding, redelivered=%v", d.Redelivered)
        d.Ack(false)
        return
    }

    var key string
    if role == "settled_ticket" {
        // settled_win/settled_loss 假定按 SettledTicketRequest 字段解析。
        key, _ = payload["ticketId"].(string)
        log.Printf("[settled_ticket] ticketId=%s status=%v ccy=%v redelivered=%v", key, payload["status"], payload["ccy"], d.Redelivered)
    } else {
        // notify_issue_result 假定按 SettledRoundRequest 字段解析；roundId
        // 字段名是本仓库运行手册标注的最大不确定性来源，真实接入前务必用真实消息核对。
        key, _ = payload["roundId"].(string)
        log.Printf("[settled_round] roundId=%s result=%v redelivered=%v", key, payload["result"], d.Redelivered)
    }

    if key == "" {
        log.Printf("[warn] missing key field, discarding (ack)")
        d.Ack(false)
        return
    }

    // 2. 第一次见到这个 key：模拟一次瞬时处理失败，nack + requeue。
    //    只失败一次，避免同一条消息被无限重投形成热循环。
    mu.Lock()
    seen := failedOnce[key]
    if !seen {
        failedOnce[key] = true
    }
    mu.Unlock()

    if !seen {
        log.Printf("[simulate-failure] %s -> nack+requeue", key)
        d.Nack(false, true)
        return
    }

    // 3. 第二次投递（redelivered=true）：放行成功，ack。
    log.Printf("[success] %s -> ack", key)
    d.Ack(false)
}

func main() {
    host := getenv("RABBITMQ_HOST", "localhost")
    port := getenv("RABBITMQ_PORT", "5671")
    vhost := getenv("RABBITMQ_VHOST", "/")
    username := getenv("RABBITMQ_USERNAME", "guest")
    password := getenv("RABBITMQ_PASSWORD", "guest")
    caFile := getenv("RABBITMQ_CA_CERT", "ca_certificate.pem")
    certFile := getenv("RABBITMQ_CLIENT_CERT", "client_certificate.pem")
    keyFile := getenv("RABBITMQ_CLIENT_KEY", "client_key.pem")
    serverName := getenv("RABBITMQ_SERVER_NAME", "")
    prefetchRaw := getenv("RABBITMQ_PREFETCH", "10")
    prefetch, err := strconv.Atoi(prefetchRaw)
    if err != nil || prefetch <= 0 {
        log.Fatalf("invalid RABBITMQ_PREFETCH=%q", prefetchRaw)
    }

    // 三个队列名默认值是官方文档已公开的示例格式；真实队列名由 PetaX 分配，
    // 务必以 PetaX 提供的为准，不要假设跟这里的示例值相同。
    queueRoles := map[string]string{
        getenv("RABBITMQ_QUEUE_SETTLED_WIN", "staging.10102.merchant102_server.settled_win.q"):          "settled_ticket",
        getenv("RABBITMQ_QUEUE_SETTLED_LOSS", "staging.10102.merchant102_server.settled_loss.q"):        "settled_ticket",
        getenv("RABBITMQ_QUEUE_ISSUE_RESULT", "staging.10102.merchant102_server.notify_issue_result.q"): "settled_round",
    }

    caCert, err := os.ReadFile(caFile)
    if err != nil {
        log.Fatalf("read ca cert: %v", err)
    }
    caPool := x509.NewCertPool()
    caPool.AppendCertsFromPEM(caCert)
    cert, err := tls.LoadX509KeyPair(certFile, keyFile)
    if err != nil {
        log.Fatalf("load client cert: %v", err)
    }
    tlsConfig := &tls.Config{
        RootCAs:      caPool,
        Certificates: []tls.Certificate{cert},
        ServerName:   serverName,
    }

    vhostPath := vhost
    if !strings.HasPrefix(vhostPath, "/") {
        vhostPath = "/" + vhostPath
    }
    amqpURL := url.URL{
        Scheme: "amqps",
        User:   url.UserPassword(username, password),
        Host:   net.JoinHostPort(host, port),
        Path:   vhostPath,
    }
    conn, err := amqp.DialTLS(amqpURL.String(), tlsConfig)
    if err != nil {
        log.Fatalf("dial: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("channel: %v", err)
    }
    if err := ch.Qos(prefetch, 0, false); err != nil {
        log.Fatalf("qos: %v", err)
    }

    done := make(chan struct{})
    for queue, role := range queueRoles {
        // QueueDeclarePassive 只检查队列存在，不主动创建——真实队列由 PetaX 预先建好。
        if _, err := ch.QueueDeclarePassive(queue, true, false, false, false, nil); err != nil {
            log.Fatalf("passive declare %q: %v", queue, err)
        }
        deliveries, err := ch.Consume(queue, "", false, false, false, false, nil)
        if err != nil {
            log.Fatalf("consume %q: %v", queue, err)
        }
        go func(role string) {
            for d := range deliveries {
                handleMessage(d, role)
            }
        }(role)
    }

    names := make([]string, 0, len(queueRoles))
    for q := range queueRoles {
        names = append(names, q)
    }
    log.Printf("listening on %v", names)
    <-done
}

Python

依赖：pip install pika。运行：python3 consumer.py。

# consumer.py — PetaX RabbitMQ 订阅 Demo (Python, pika)
# 运行：pip install pika && python3 consumer.py
# 注意：字段形状是基于 HTTP 回调 DTO 的假设，未经真实 PetaX 消息验证，
# 尤其 notify_issue_result 的 id 字段可能是 issueId 而非 roundId，接入前务必用真实消息核对。
import json
import os
import ssl

import pika

HOST = os.environ.get("RABBITMQ_HOST", "localhost")
PORT = int(os.environ.get("RABBITMQ_PORT", "5671"))
VHOST = os.environ.get("RABBITMQ_VHOST", "/")
USERNAME = os.environ.get("RABBITMQ_USERNAME", "guest")
PASSWORD = os.environ.get("RABBITMQ_PASSWORD", "guest")
CA_CERT = os.environ.get("RABBITMQ_CA_CERT", "ca_certificate.pem")
CLIENT_CERT = os.environ.get("RABBITMQ_CLIENT_CERT", "client_certificate.pem")
CLIENT_KEY = os.environ.get("RABBITMQ_CLIENT_KEY", "client_key.pem")
SERVER_NAME = os.environ.get("RABBITMQ_SERVER_NAME", "")

# 三个队列名默认值是官方文档已公开的示例格式；真实队列名由 PetaX 分配，
# 务必以 PetaX 提供的为准，不要假设跟这里的示例值相同。
QUEUE_ROLES = {
    os.environ.get("RABBITMQ_QUEUE_SETTLED_WIN", "staging.10102.merchant102_server.settled_win.q"): "settled_ticket",
    os.environ.get("RABBITMQ_QUEUE_SETTLED_LOSS", "staging.10102.merchant102_server.settled_loss.q"): "settled_ticket",
    os.environ.get("RABBITMQ_QUEUE_ISSUE_RESULT", "staging.10102.merchant102_server.notify_issue_result.q"): "settled_round",
}

# 仅用于 Demo：记录已经"模拟失败过一次"的 key（ticketId 或 roundId），
# 第二次投递（redelivered=true）就放行成功。真实商户不需要这个逻辑，
# 这里只是让 demo 能演示 nack+requeue 而不会无限热循环重投同一条坏消息。
failed_once = set()


def handle_message(channel, method, properties, body, role):
    redelivered = method.redelivered
    try:
        payload = json.loads(body)
    except json.JSONDecodeError:
        # 1. 解析失败 = 永久错误：ack 丢弃，避免毒消息卡队列。
        print(f"[bad-json] discarding, redelivered={redelivered}", flush=True)
        channel.basic_ack(method.delivery_tag)
        return

    if role == "settled_ticket":
        # settled_win/settled_loss 假定按 SettledTicketRequest 字段解析。
        key = payload.get("ticketId")
        print(f"[settled_ticket] ticketId={key} status={payload.get('status')} ccy={payload.get('ccy')} redelivered={redelivered}", flush=True)
    else:
        # notify_issue_result 假定按 SettledRoundRequest 字段解析；roundId 字段名
        # 是本仓库运行手册标注的最大不确定性来源，真实接入前务必用真实消息核对。
        key = payload.get("roundId")
        print(f"[settled_round] roundId={key} result={payload.get('result')} redelivered={redelivered}", flush=True)

    if key is None:
        print("[warn] missing key field, discarding (ack)", flush=True)
        channel.basic_ack(method.delivery_tag)
        return

    # 2. 第一次见到这个 key：模拟一次瞬时处理失败，nack + requeue。
    #    只失败一次，避免同一条消息被无限重投形成热循环。
    if key not in failed_once:
        failed_once.add(key)
        print(f"[simulate-failure] {key} -> nack+requeue", flush=True)
        channel.basic_nack(method.delivery_tag, requeue=True)
        return

    # 3. 第二次投递（redelivered=true）：放行成功，ack。
    print(f"[success] {key} -> ack", flush=True)
    channel.basic_ack(method.delivery_tag)


def main():
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    context.load_verify_locations(CA_CERT)
    context.load_cert_chain(CLIENT_CERT, CLIENT_KEY)
    ssl_options = pika.SSLOptions(context, server_hostname=SERVER_NAME or HOST)

    credentials = pika.PlainCredentials(USERNAME, PASSWORD)
    parameters = pika.ConnectionParameters(
        host=HOST, port=PORT, virtual_host=VHOST,
        credentials=credentials, ssl_options=ssl_options,
    )
    connection = pika.BlockingConnection(parameters)
    channel = connection.channel()
    channel.basic_qos(prefetch_count=int(os.environ.get("RABBITMQ_PREFETCH", "10")))

    for queue, role in QUEUE_ROLES.items():
        # passive=True 只检查队列存在，不主动创建——真实队列由 PetaX 预先建好。
        channel.queue_declare(queue=queue, passive=True)
        channel.basic_consume(
            queue=queue,
            on_message_callback=lambda ch, method, properties, body, role=role: handle_message(ch, method, properties, body, role),
        )

    print(f"listening on {', '.join(QUEUE_ROLES.keys())}", flush=True)
    channel.start_consuming()


if __name__ == "__main__":
    main()