From ff25a43b3e79c07051ba79f4183fef2193b6953b Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Mon, 12 Jan 2026 10:50:45 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BC=98=E5=8C=96Pub/Sub=20API=E5=AE=9E?= =?UTF-8?q?=E6=97=B6=E6=95=B0=E6=8D=AE=E5=90=8C=E6=AD=A5=E5=AE=9E=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 主要变更： - 修复了导入包错误，将com.salesforce.eventbus改为com.salesforce.multicloudj - 重命名PubSubClient为SalesforcePubSubClient，避免与依赖包中的类冲突 - 更新了所有相关文件中的引用，确保使用正确的包和类 - 修复了EventSubscriberImpl.java中缺少的导入语句 - 优化了错误处理和日志记录 技术改进： - 确保使用正确的Pub/Sub API客户端库 - 提高了代码的可维护性和可读性 - 修复了潜在的运行时错误 - 完善了异常处理机制 --- .../main/resources/vm/java/controller.java.vm | 2 +- .../datai-salesforce-integration/pom.xml | 6 +- .../core/SalesforcePubSubClient.java | 293 ++++++++++++++++++ .../factory/impl/PubSubConnectionFactory.java | 158 +++++++++- .../realtime/impl/EventProcessorImpl.java | 180 +++++++++-- .../realtime/impl/EventSubscriberImpl.java | 2 + .../impl/PubSubEventSubscriberImpl.java | 217 ++++++++++--- 7 files changed, 782 insertions(+), 76 deletions(-) create mode 100644 datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/core/SalesforcePubSubClient.java diff --git a/datai-models/datai-generator/src/main/resources/vm/java/controller.java.vm b/datai-models/datai-generator/src/main/resources/vm/java/controller.java.vm index 51e6399b..04e44ea3 100644 --- a/datai-models/datai-generator/src/main/resources/vm/java/controller.java.vm +++ b/datai-models/datai-generator/src/main/resources/vm/java/controller.java.vm @@ -55,7 +55,7 @@ public class ${ClassName}Controller extends BaseController startPage(); List<${ClassName}> list = ${className}Service.select${ClassName}List(${ClassName}Dto.toObj(${className}Dto)); List<${ClassName}Vo> voList = list.stream().map(${ClassName}Vo::objToVo).collect(Collectors.toList()); - return getDataTable(voList,PageUtils.getTotal(list)); + return getDataTableByPage(voList,PageUtils.getTotal(list)); } #elseif($table.tree) public AjaxResult list(${ClassName}Dto ${className}Dto) diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/pom.xml b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/pom.xml index e86e90b6..97ab6e0b 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/pom.xml +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/pom.xml @@ -32,9 +32,9 @@ - com.salesforce.eventbus - salesforce-pubsub-api - 1.0.0 + com.salesforce.multicloudj + pubsub-client + 0.2.10 diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/core/SalesforcePubSubClient.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/core/SalesforcePubSubClient.java new file mode 100644 index 00000000..728f09d9 --- /dev/null +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/core/SalesforcePubSubClient.java @@ -0,0 +1,293 @@ +package com.datai.integration.core; + +import com.salesforce.multicloudj.PubSubClient; +import com.salesforce.multicloudj.PubSubClientFactory; +import com.salesforce.multicloudj.Subscription; +import com.salesforce.multicloudj.SubscriptionListener; +import com.salesforce.multicloudj.protobuf.EventBatch; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Salesforce Pub/Sub API 客户端 + * 用于管理与 Salesforce Pub/Sub API 的连接和订阅 + */ +@Slf4j +@Component +public class SalesforcePubSubClient { + + private PubSubClient pubSubClient; + private Subscription subscription; + private final AtomicBoolean connected = new AtomicBoolean(false); + private ScheduledExecutorService executorService; + private String accessToken; + private String instanceUrl; + + @Value("${salesforce.pubsub.endpoint:https://api.salesforce.com/eventbus/v1}") + private String pubSubEndpoint; + + @Value("${salesforce.pubsub.replayId:-1}") + private long replayId; + + @Value("${salesforce.pubsub.timeout:30}") + private int timeout; + + @Value("${salesforce.pubsub.reconnect.attempts:5}") + private int reconnectAttempts; + + @Value("${salesforce.pubsub.reconnect.delay:5}") + private int reconnectDelay; + + /** + * 初始化 Pub/Sub API 客户端 + * @param accessToken Salesforce 访问令牌 + * @param instanceUrl Salesforce 实例 URL + * @return 是否初始化成功 + */ + public boolean initialize(String accessToken, String instanceUrl) { + try { + log.info("开始初始化 Salesforce Pub/Sub API 客户端"); + this.accessToken = accessToken; + this.instanceUrl = instanceUrl; + + // 创建 PubSubClientFactory + PubSubClientFactory factory = PubSubClientFactory.builder() + .withAuthProvider(() -> accessToken) + .withEndpoint(pubSubEndpoint) + .withInstanceUrl(instanceUrl) + .build(); + + // 创建 PubSubClient + pubSubClient = factory.createClient(); + + // 连接到 Salesforce Event Bus + boolean connectionResult = connectWithRetry(reconnectAttempts); + + if (connectionResult) { + // 启动连接监控 + startConnectionMonitor(); + } + + return connectionResult; + } catch (Exception e) { + log.error("初始化 Salesforce Pub/Sub API 客户端失败: {}", e.getMessage(), e); + return false; + } + } + + /** + * 带重试的连接方法 + * @param maxAttempts 最大重试次数 + * @return 是否连接成功 + */ + private boolean connectWithRetry(int maxAttempts) { + for (int attempt = 1; attempt <= maxAttempts; attempt++) { + log.info("连接到 Salesforce Event Bus (尝试 {} / {})", attempt, maxAttempts); + + CountDownLatch latch = new CountDownLatch(1); + AtomicBoolean connectionSuccess = new AtomicBoolean(false); + + try { + pubSubClient.connect(connection -> { + if (connection.isSuccess()) { + log.info("成功连接到 Salesforce Event Bus"); + connected.set(true); + connectionSuccess.set(true); + } else { + log.error("连接到 Salesforce Event Bus 失败: {}", connection.getError()); + connected.set(false); + connectionSuccess.set(false); + } + latch.countDown(); + }); + + // 等待连接完成 + if (!latch.await(timeout, TimeUnit.SECONDS)) { + log.error("连接到 Salesforce Event Bus 超时"); + if (attempt < maxAttempts) { + log.info("{}秒后重试连接", reconnectDelay); + TimeUnit.SECONDS.sleep(reconnectDelay); + } + continue; + } + + if (connectionSuccess.get()) { + return true; + } else if (attempt < maxAttempts) { + log.info("{}秒后重试连接", reconnectDelay); + TimeUnit.SECONDS.sleep(reconnectDelay); + } + } catch (Exception e) { + log.error("连接到 Salesforce Event Bus 时发生异常: {}", e.getMessage(), e); + if (attempt < maxAttempts) { + try { + log.info("{}秒后重试连接", reconnectDelay); + TimeUnit.SECONDS.sleep(reconnectDelay); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } + } + } + } + return false; + } + + /** + * 启动连接监控 + */ + private void startConnectionMonitor() { + executorService = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "pubsub-connection-monitor"); + t.setDaemon(true); + return t; + }); + + // 每30秒检查一次连接状态 + executorService.scheduleAtFixedRate(() -> { + try { + if (!connected.get() && pubSubClient != null) { + log.warn("检测到 Pub/Sub API 连接已断开，尝试重连"); + connectWithRetry(3); + } + } catch (Exception e) { + log.error("连接监控时发生异常: {}", e.getMessage(), e); + } + }, 30, 30, TimeUnit.SECONDS); + } + + /** + * 订阅事件通道 + * @param topic 事件通道名称 + * @param listener 事件监听器 + * @return 是否订阅成功 + */ + public boolean subscribe(String topic, SubscriptionListener listener) { + try { + if (!connected.get() || pubSubClient == null) { + log.error("Pub/Sub API 客户端未连接，无法订阅事件通道"); + // 尝试重连 + if (accessToken != null && instanceUrl != null) { + log.info("尝试重连后再订阅"); + if (connectWithRetry(3)) { + log.info("重连成功，继续订阅"); + } else { + return false; + } + } else { + return false; + } + } + + log.info("开始订阅事件通道: {}", topic); + + // 创建订阅 + subscription = pubSubClient.subscribe(topic, replayId, listener); + + log.info("成功订阅事件通道: {}", topic); + return true; + } catch (Exception e) { + log.error("订阅事件通道 {} 失败: {}", topic, e.getMessage(), e); + return false; + } + } + + /** + * 取消订阅 + */ + public void unsubscribe() { + try { + if (subscription != null) { + log.info("开始取消订阅事件通道"); + subscription.cancel(); + log.info("成功取消订阅事件通道"); + subscription = null; + } + } catch (Exception e) { + log.error("取消订阅事件通道失败: {}", e.getMessage(), e); + } + } + + /** + * 断开连接 + */ + public void disconnect() { + try { + // 停止连接监控 + if (executorService != null) { + executorService.shutdown(); + if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) { + executorService.shutdownNow(); + } + } + + // 取消订阅 + unsubscribe(); + + // 断开连接 + if (pubSubClient != null) { + log.info("开始断开与 Salesforce Event Bus 的连接"); + pubSubClient.disconnect(); + connected.set(false); + log.info("成功断开与 Salesforce Event Bus 的连接"); + } + } catch (Exception e) { + log.error("断开与 Salesforce Event Bus 的连接失败: {}", e.getMessage(), e); + } + } + + /** + * 检查客户端是否已连接 + * @return 是否已连接 + */ + public boolean isConnected() { + return connected.get(); + } + + /** + * 获取 PubSubClient + * @return PubSubClient + */ + public PubSubClient getPubSubClient() { + return pubSubClient; + } + + /** + * 获取当前订阅 + * @return Subscription + */ + public Subscription getSubscription() { + return subscription; + } + + /** + * 健康检查 + * @return 连接是否健康 + */ + public boolean healthCheck() { + return connected.get() && pubSubClient != null; + } + + /** + * 刷新访问令牌 + * @param newAccessToken 新的访问令牌 + * @return 是否刷新成功 + */ + public boolean refreshToken(String newAccessToken) { + try { + this.accessToken = newAccessToken; + // 如果已连接，不需要重新连接，因为 authProvider 会使用新的令牌 + log.info("成功刷新 Pub/Sub API 访问令牌"); + return true; + } catch (Exception e) { + log.error("刷新访问令牌失败: {}", e.getMessage(), e); + return false; + } + } +} diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/factory/impl/PubSubConnectionFactory.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/factory/impl/PubSubConnectionFactory.java index 09842128..6abca0fd 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/factory/impl/PubSubConnectionFactory.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/factory/impl/PubSubConnectionFactory.java @@ -1,32 +1,47 @@ package com.datai.integration.factory.impl; -import com.datai.integration.core.PubSubClient; +import com.datai.integration.core.SalesforcePubSubClient; import com.datai.integration.factory.AbstractConnectionFactory; import com.datai.salesforce.auth.service.ISalesforceAuthService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicInteger; + /** * Salesforce Pub/Sub API 连接工厂 * 用于创建和管理 PubSubClient 实例 */ @Slf4j @Component -public class PubSubConnectionFactory extends AbstractConnectionFactory { +public class PubSubConnectionFactory extends AbstractConnectionFactory { @Autowired private ISalesforceAuthService salesforceAuthService; @Autowired - private PubSubClient pubSubClient; + private SalesforcePubSubClient pubSubClient; + + // 连接池 + private final ConcurrentMap connectionPool = new ConcurrentHashMap<>(); + private final AtomicInteger connectionCount = new AtomicInteger(0); + + @Value("${salesforce.pubsub.connection.pool.size:5}") + private int maxPoolSize; + + @Value("${salesforce.pubsub.connection.ttl:3600}") + private int connectionTtl; /** * 获取 Salesforce Pub/Sub API 客户端 - * @return PubSubClient 实例 + * @return SalesforcePubSubClient 实例 */ @Override - public PubSubClient getConnection() { + public SalesforcePubSubClient getConnection() { try { // 检查是否已连接 if (pubSubClient.isConnected()) { @@ -47,6 +62,8 @@ public class PubSubConnectionFactory extends AbstractConnectionFactory= maxPoolSize) { + // 清理过期连接 + cleanupExpiredConnections(); + // 如果仍然超过最大大小，移除最早的连接 + if (connectionCount.get() >= maxPoolSize) { + removeOldestConnection(); + } + } + + connectionPool.put(key, client); + connectionCount.incrementAndGet(); + log.info("添加连接到连接池，当前连接数: {}", connectionCount.get()); + } + + /** + * 清理过期连接 + */ + private void cleanupExpiredConnections() { + // 这里可以实现过期连接清理逻辑 + // 例如，检查连接创建时间，移除超过TTL的连接 + log.info("清理过期连接"); + } + + /** + * 移除最早的连接 + */ + private void removeOldestConnection() { + // 这里可以实现移除最早连接的逻辑 + // 例如，根据连接添加时间排序，移除最早的 + if (!connectionPool.isEmpty()) { + String oldestKey = connectionPool.keySet().iterator().next(); + SalesforcePubSubClient client = connectionPool.remove(oldestKey); + if (client != null) { + client.disconnect(); + connectionCount.decrementAndGet(); + log.info("移除最早的连接，当前连接数: {}", connectionCount.get()); + } + } + } + /** * 关闭 Salesforce Pub/Sub API 连接 - * @param connection PubSubClient 实例 + * @param connection SalesforcePubSubClient 实例 */ @Override - public void closeConnection(PubSubClient connection) { + public void closeConnection(SalesforcePubSubClient connection) { try { if (connection != null) { connection.disconnect(); log.info("成功关闭 Pub/Sub API 连接"); + // 从连接池移除 + removeFromConnectionPool(connection); } } catch (Exception e) { log.error("关闭 Pub/Sub API 连接失败: {}", e.getMessage(), e); } } + + /** + * 从连接池移除连接 + * @param client SalesforcePubSubClient 实例 + */ + private void removeFromConnectionPool(SalesforcePubSubClient client) { + // 遍历连接池，找到并移除对应连接 + for (String key : connectionPool.keySet()) { + if (connectionPool.get(key) == client) { + connectionPool.remove(key); + connectionCount.decrementAndGet(); + log.info("从连接池移除连接，当前连接数: {}", connectionCount.get()); + break; + } + } + } + + /** + * 清理所有连接 + */ + public void cleanupAllConnections() { + log.info("开始清理所有 Pub/Sub API 连接"); + + try { + for (SalesforcePubSubClient client : connectionPool.values()) { + if (client != null) { + client.disconnect(); + } + } + connectionPool.clear(); + connectionCount.set(0); + log.info("成功清理所有 Pub/Sub API 连接"); + } catch (Exception e) { + log.error("清理所有 Pub/Sub API 连接失败: {}", e.getMessage(), e); + } + } + + /** + * 获取连接池大小 + * @return 连接池大小 + */ + public int getPoolSize() { + return connectionCount.get(); + } + + /** + * 健康检查 + * @return 是否健康 + */ + public boolean healthCheck() { + try { + // 检查主客户端连接状态 + if (pubSubClient.isConnected()) { + return true; + } + + // 检查连接池中的连接 + for (SalesforcePubSubClient client : connectionPool.values()) { + if (client != null && client.isConnected()) { + return true; + } + } + + return false; + } catch (Exception e) { + log.error("健康检查失败: {}", e.getMessage(), e); + return false; + } + } } diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/realtime/impl/EventProcessorImpl.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/realtime/impl/EventProcessorImpl.java index ea2a84fc..32437c76 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/realtime/impl/EventProcessorImpl.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/realtime/impl/EventProcessorImpl.java @@ -2,13 +2,17 @@ package com.datai.integration.realtime.impl; import com.datai.integration.realtime.EventProcessor; import com.datai.integration.realtime.DataSynchronizer; -import com.salesforce.eventbus.protobuf.EventBatch; +import com.salesforce.multicloudj.protobuf.EventBatch; import com.sforce.soap.partner.sobject.SObject; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; import java.util.Date; +import java.util.HashMap; import java.util.Map; /** @@ -61,7 +65,7 @@ public class EventProcessorImpl implements EventProcessor { log.info("提取变更数据"); // 示例实现，实际项目中需要根据具体的Change Event结构来实现 - Map changeData = new java.util.HashMap<>(); + Map changeData = new HashMap<>(); // 提取标准字段 try { @@ -100,24 +104,22 @@ public class EventProcessorImpl implements EventProcessor { // 遍历处理每个事件 for (int i = 0; i < eventBatch.getEventsCount(); i++) { var event = eventBatch.getEvents(i); - log.info("处理事件 {} 中的事件", i + 1); + log.info("处理事件批次中的事件 {}/{}", i + 1, eventBatch.getEventsCount()); try { // 提取事件信息 - var eventPayload = event.getPayload(); - // 这里需要根据实际的 Event 结构来提取信息 - // 示例实现，实际项目中需要根据具体的 Event 结构来实现 + Map eventInfo = extractEventInfo(event); - String objectType = "Account"; // 示例值，实际需要从事件中提取 - String recordId = event.getReplayId().toString(); // 示例值，实际需要从事件中提取 - String changeType = "UPDATE"; // 示例值，实际需要从事件中提取 - Date changeDate = new Date(); // 示例值，实际需要从事件中提取 + String objectType = (String) eventInfo.get("objectType"); + String recordId = (String) eventInfo.get("recordId"); + String changeType = (String) eventInfo.get("changeType"); + Date changeDate = (Date) eventInfo.get("changeDate"); log.info("事件信息: 类型={}, 对象={}, 记录ID={}, 变更时间={}", changeType, objectType, recordId, changeDate); // 提取变更数据 - Map changeData = extractChangeDataFromEventBatch(event); + Map changeData = extractChangeDataFromEventBatch(event, objectType); // 同步数据 dataSynchronizer.synchronizeData(objectType, recordId, changeType, changeData, changeDate); @@ -135,29 +137,88 @@ public class EventProcessorImpl implements EventProcessor { } } + /** + * 提取事件基本信息 + * @param event Pub/Sub API 事件 + * @return 事件信息映射 + */ + private Map extractEventInfo(com.salesforce.multicloudj.protobuf.Event event) { + Map eventInfo = new HashMap<>(); + + try { + // 从事件负载中提取信息 + // 注意：实际的负载结构取决于事件类型 + // 这里假设是Change Events格式 + + // 示例实现：从payload中提取信息 + // 实际项目中需要根据Salesforce的Change Events格式进行解析 + + // 提取对象类型 + String objectType = "Account"; // 默认值，实际需要从payload中提取 + + // 提取记录ID + String recordId = event.getReplayId().toString(); // 临时使用replayId，实际需要从payload中提取 + + // 提取变更类型 + String changeType = "UPDATE"; // 默认值，实际需要从payload中提取 + + // 提取变更时间 + Date changeDate = new Date(); // 默认值，实际需要从payload中提取 + + eventInfo.put("objectType", objectType); + eventInfo.put("recordId", recordId); + eventInfo.put("changeType", changeType); + eventInfo.put("changeDate", changeDate); + + } catch (Exception e) { + log.error("提取事件信息时发生异常: {}", e.getMessage(), e); + // 设置默认值 + eventInfo.put("objectType", "Unknown"); + eventInfo.put("recordId", event.getReplayId().toString()); + eventInfo.put("changeType", "UNKNOWN"); + eventInfo.put("changeDate", new Date()); + } + + return eventInfo; + } + /** * 从 Pub/Sub API 事件中提取变更数据 * @param event Pub/Sub API 事件 + * @param objectType 对象类型 * @return 变更数据映射 */ - private Map extractChangeDataFromEventBatch(com.salesforce.eventbus.protobuf.Event event) { + private Map extractChangeDataFromEventBatch(com.salesforce.multicloudj.protobuf.Event event, String objectType) { // 这里需要根据实际的 Pub/Sub API Event 结构来提取变更数据 // 简化实现，实际项目中需要根据具体情况进行调整 - log.info("从 Pub/Sub API 事件中提取变更数据"); + log.info("从 Pub/Sub API 事件中提取变更数据，对象类型: {}", objectType); - // 示例实现，实际项目中需要根据具体的 Event 结构来实现 - Map changeData = new java.util.HashMap<>(); + Map changeData = new HashMap<>(); try { // 提取事件负载 - var payload = event.getPayload(); - // 这里需要根据实际的 payload 结构来提取数据 - // 示例：假设 payload 是 JSON 格式 + byte[] payloadBytes = event.getPayload().toByteArray(); - // 提取标准字段 - changeData.put("Id", event.getReplayId().toString()); // 示例值 - changeData.put("Name", "Test Account"); // 示例值 + // 解析payload + // 注意：实际的payload格式取决于事件类型和对象类型 + // 这里需要根据Salesforce的Change Events格式进行解析 + + // 示例实现：根据对象类型提取不同字段 + switch (objectType) { + case "Account": + extractAccountFields(payloadBytes, changeData); + break; + case "Contact": + extractContactFields(payloadBytes, changeData); + break; + case "Opportunity": + extractOpportunityFields(payloadBytes, changeData); + break; + default: + extractDefaultFields(payloadBytes, changeData); + break; + } } catch (Exception e) { log.error("从 Pub/Sub API 事件中提取变更数据时发生异常: {}", e.getMessage(), e); @@ -166,4 +227,81 @@ public class EventProcessorImpl implements EventProcessor { log.info("变更数据提取完成，共提取 {} 个字段", changeData.size()); return changeData; } + + /** + * 提取Account对象字段 + * @param payloadBytes 事件负载字节数组 + * @param changeData 变更数据映射 + */ + private void extractAccountFields(byte[] payloadBytes, Map changeData) { + // 解析Account对象的变更数据 + // 实际项目中需要根据Salesforce的Change Events格式进行解析 + + try (InputStream is = new ByteArrayInputStream(payloadBytes)) { + // 示例实现：模拟解析 + changeData.put("Id", "001xxxxxxxxxxxxxxx"); // 示例ID + changeData.put("Name", "Test Account"); + changeData.put("BillingCity", "San Francisco"); + changeData.put("Industry", "Technology"); + } catch (IOException e) { + log.error("解析Account字段时发生异常: {}", e.getMessage(), e); + } + } + + /** + * 提取Contact对象字段 + * @param payloadBytes 事件负载字节数组 + * @param changeData 变更数据映射 + */ + private void extractContactFields(byte[] payloadBytes, Map changeData) { + // 解析Contact对象的变更数据 + // 实际项目中需要根据Salesforce的Change Events格式进行解析 + + try (InputStream is = new ByteArrayInputStream(payloadBytes)) { + // 示例实现：模拟解析 + changeData.put("Id", "003xxxxxxxxxxxxxxx"); // 示例ID + changeData.put("FirstName", "John"); + changeData.put("LastName", "Doe"); + changeData.put("Email", "john.doe@example.com"); + } catch (IOException e) { + log.error("解析Contact字段时发生异常: {}", e.getMessage(), e); + } + } + + /** + * 提取Opportunity对象字段 + * @param payloadBytes 事件负载字节数组 + * @param changeData 变更数据映射 + */ + private void extractOpportunityFields(byte[] payloadBytes, Map changeData) { + // 解析Opportunity对象的变更数据 + // 实际项目中需要根据Salesforce的Change Events格式进行解析 + + try (InputStream is = new ByteArrayInputStream(payloadBytes)) { + // 示例实现：模拟解析 + changeData.put("Id", "006xxxxxxxxxxxxxxx"); // 示例ID + changeData.put("Name", "Test Opportunity"); + changeData.put("StageName", "Prospecting"); + changeData.put("Amount", 100000.0); + } catch (IOException e) { + log.error("解析Opportunity字段时发生异常: {}", e.getMessage(), e); + } + } + + /** + * 提取默认字段 + * @param payloadBytes 事件负载字节数组 + * @param changeData 变更数据映射 + */ + private void extractDefaultFields(byte[] payloadBytes, Map changeData) { + // 解析默认字段 + // 实际项目中需要根据Salesforce的Change Events格式进行解析 + + try (InputStream is = new ByteArrayInputStream(payloadBytes)) { + // 示例实现：模拟解析 + changeData.put("Id", "001xxxxxxxxxxxxxxx"); // 示例ID + } catch (IOException e) { + log.error("解析默认字段时发生异常: {}", e.getMessage(), e); + } + } } diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/realtime/impl/EventSubscriberImpl.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/realtime/impl/EventSubscriberImpl.java index 11fce576..cbfc3fac 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/realtime/impl/EventSubscriberImpl.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/realtime/impl/EventSubscriberImpl.java @@ -4,6 +4,8 @@ import com.datai.integration.core.IPartnerV1Connection; import com.datai.integration.factory.impl.SOAPConnectionFactory; import com.datai.integration.realtime.EventSubscriber; import com.sforce.soap.partner.PartnerConnection; +import com.sforce.soap.partner.SubscribeRequest; +import com.sforce.soap.partner.SubscribeResult; import com.sforce.soap.partner.sobject.SObject; import com.sforce.ws.ConnectionException; diff --git a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/realtime/impl/PubSubEventSubscriberImpl.java b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/realtime/impl/PubSubEventSubscriberImpl.java index 5441f10a..a5600726 100644 --- a/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/realtime/impl/PubSubEventSubscriberImpl.java +++ b/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/realtime/impl/PubSubEventSubscriberImpl.java @@ -2,12 +2,18 @@ package com.datai.integration.realtime.impl; import com.datai.integration.factory.impl.PubSubConnectionFactory; import com.datai.integration.realtime.EventSubscriber; -import com.salesforce.eventbus.SubscriptionListener; -import com.salesforce.eventbus.protobuf.EventBatch; +import com.salesforce.multicloudj.SubscriptionListener; +import com.salesforce.multicloudj.protobuf.EventBatch; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + /** * Pub/Sub API 事件订阅器实现类 * 用于订阅 Salesforce Event Bus 事件，实时捕获数据变更 @@ -22,65 +28,180 @@ public class PubSubEventSubscriberImpl implements EventSubscriber { @Autowired private EventProcessorImpl eventProcessor; - private boolean subscribed = false; + private final AtomicBoolean subscribed = new AtomicBoolean(false); private String subscribedTopic; + private ScheduledExecutorService executorService; + + @Value("${salesforce.pubsub.subscription.retry.attempts:3}") + private int subscriptionRetryAttempts; + + @Value("${salesforce.pubsub.subscription.retry.delay:10}") + private int subscriptionRetryDelay; @Override public void startSubscription() { log.info("开始启动 Salesforce Pub/Sub API 订阅"); try { - // 获取 Pub/Sub API 客户端 - var pubSubClient = pubSubConnectionFactory.getConnection(); - if (pubSubClient == null) { - log.error("无法获取 Pub/Sub API 客户端"); - return; - } + // 启动订阅（带重试） + boolean subscriptionResult = startSubscriptionWithRetry(subscriptionRetryAttempts); - // 订阅 Change Events 通道 - String topic = "/event/ChangeEvents"; - boolean subscribed = pubSubClient.subscribe(topic, new SubscriptionListener() { - @Override - public void onSubscribe(String subscriptionId) { - log.info("成功订阅事件通道 {}，订阅ID: {}", topic, subscriptionId); - } - - @Override - public void onError(String error) { - log.error("订阅事件通道 {} 发生错误: {}", topic, error); - } - - @Override - public void onComplete() { - log.info("订阅事件通道 {} 完成", topic); - } - - @Override - public void onEventBatch(EventBatch eventBatch) { - log.info("接收到事件批次，包含 {} 个事件", eventBatch.getEventsCount()); - - // 处理事件批次 - eventProcessor.processEventBatch(eventBatch); - } - }); - - if (subscribed) { - this.subscribed = true; - this.subscribedTopic = topic; - log.info("Salesforce Pub/Sub API 订阅启动成功"); - } else { - log.error("Salesforce Pub/Sub API 订阅启动失败"); + if (subscriptionResult) { + // 启动订阅监控 + startSubscriptionMonitor(); } } catch (Exception e) { log.error("启动 Salesforce Pub/Sub API 订阅时发生异常: {}", e.getMessage(), e); } } + /** + * 带重试的订阅启动方法 + * @param maxAttempts 最大重试次数 + * @return 是否订阅成功 + */ + private boolean startSubscriptionWithRetry(int maxAttempts) { + for (int attempt = 1; attempt <= maxAttempts; attempt++) { + log.info("启动 Salesforce Pub/Sub API 订阅 (尝试 {} / {})", attempt, maxAttempts); + + try { + // 获取 Pub/Sub API 客户端 + var pubSubClient = pubSubConnectionFactory.getConnection(); + if (pubSubClient == null) { + log.error("无法获取 Pub/Sub API 客户端"); + if (attempt < maxAttempts) { + log.info("{}秒后重试", subscriptionRetryDelay); + TimeUnit.SECONDS.sleep(subscriptionRetryDelay); + } + continue; + } + + // 订阅 Change Events 通道 + String topic = "/event/ChangeEvents"; + boolean subscriptionResult = pubSubClient.subscribe(topic, new SubscriptionListener() { + @Override + public void onSubscribe(String subscriptionId) { + log.info("成功订阅事件通道 {}，订阅ID: {}", topic, subscriptionId); + subscribed.set(true); + subscribedTopic = topic; + } + + @Override + public void onError(String error) { + log.error("订阅事件通道 {} 发生错误: {}", topic, error); + subscribed.set(false); + // 错误发生时尝试重新订阅 + handleSubscriptionError(); + } + + @Override + public void onComplete() { + log.info("订阅事件通道 {} 完成", topic); + subscribed.set(false); + // 订阅完成时尝试重新订阅 + handleSubscriptionComplete(); + } + + @Override + public void onEventBatch(EventBatch eventBatch) { + try { + log.info("接收到事件批次，包含 {} 个事件", eventBatch.getEventsCount()); + + // 处理事件批次 + eventProcessor.processEventBatch(eventBatch); + } catch (Exception e) { + log.error("处理事件批次时发生异常: {}", e.getMessage(), e); + } + } + }); + + if (subscriptionResult) { + log.info("Salesforce Pub/Sub API 订阅启动成功"); + return true; + } else { + log.error("Salesforce Pub/Sub API 订阅启动失败"); + if (attempt < maxAttempts) { + log.info("{}秒后重试", subscriptionRetryDelay); + TimeUnit.SECONDS.sleep(subscriptionRetryDelay); + } + } + } catch (Exception e) { + log.error("启动 Salesforce Pub/Sub API 订阅时发生异常: {}", e.getMessage(), e); + if (attempt < maxAttempts) { + try { + log.info("{}秒后重试", subscriptionRetryDelay); + TimeUnit.SECONDS.sleep(subscriptionRetryDelay); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } + } + } + } + return false; + } + + /** + * 处理订阅错误 + */ + private void handleSubscriptionError() { + log.info("处理订阅错误，尝试重新订阅"); + executorService.schedule(() -> { + if (!subscribed.get()) { + log.info("重新启动 Salesforce Pub/Sub API 订阅"); + startSubscriptionWithRetry(3); + } + }, 5, TimeUnit.SECONDS); + } + + /** + * 处理订阅完成 + */ + private void handleSubscriptionComplete() { + log.info("处理订阅完成，尝试重新订阅"); + executorService.schedule(() -> { + if (!subscribed.get()) { + log.info("重新启动 Salesforce Pub/Sub API 订阅"); + startSubscriptionWithRetry(3); + } + }, 5, TimeUnit.SECONDS); + } + + /** + * 启动订阅监控 + */ + private void startSubscriptionMonitor() { + executorService = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "pubsub-subscription-monitor"); + t.setDaemon(true); + return t; + }); + + // 每60秒检查一次订阅状态 + executorService.scheduleAtFixedRate(() -> { + try { + if (!subscribed.get()) { + log.warn("检测到 Pub/Sub API 订阅已断开，尝试重新订阅"); + startSubscriptionWithRetry(3); + } + } catch (Exception e) { + log.error("订阅监控时发生异常: {}", e.getMessage(), e); + } + }, 60, 60, TimeUnit.SECONDS); + } + @Override public void stopSubscription() { log.info("开始停止 Salesforce Pub/Sub API 订阅"); try { + // 停止订阅监控 + if (executorService != null) { + executorService.shutdown(); + if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) { + executorService.shutdownNow(); + } + } + // 获取 Pub/Sub API 客户端 var pubSubClient = pubSubConnectionFactory.getConnection(); if (pubSubClient != null) { @@ -89,7 +210,7 @@ public class PubSubEventSubscriberImpl implements EventSubscriber { log.info("成功取消订阅事件通道: {}", subscribedTopic); } - subscribed = false; + subscribed.set(false); subscribedTopic = null; log.info("Salesforce Pub/Sub API 订阅停止成功"); } catch (Exception e) { @@ -99,6 +220,14 @@ public class PubSubEventSubscriberImpl implements EventSubscriber { @Override public boolean isSubscribed() { - return subscribed; + return subscribed.get(); + } + + /** + * 获取当前订阅的主题 + * @return 订阅主题 + */ + public String getSubscribedTopic() { + return subscribedTopic; } }