"""模块：MSYNC Socket.IO 客户端 替代 HTTP REST，通过 Socket.IO 与 SMANAGE 主服务器实时通信。 所有方法名、事件字段全部使用 PascalCase。 优势： - 连接复用，无需每次请求建立 TCP/TLS - 内置心跳保活，替代应用层心跳线程 - 服务端可主动推送事件（服务上下线通知） - 断线自动重连 """ import hashlib import logging import threading import time import socketio from gvsdsdk.msync.auth import GenerateChallenge, BuildAuthPayload, SignPayload logger = logging.getLogger(__name__) class MSyncSocketClient: """MSYNC Socket.IO 客户端——子服务器接入 SMANAGE 的实时通道""" def __init__(self, master_url, service_name, service_domain, service_secret, scope=None, heartbeat_interval=30, service_type=0, metadata=None): self.MasterURL = master_url.rstrip('/') self.ServiceName = service_name self.ServiceDomain = service_domain self.ServiceSecret = service_secret self.Scope = scope or ['ogm:read', 'ogm:write', 'registry:read'] self.HeartbeatInterval = heartbeat_interval self.ServiceType = service_type self.Metadata = metadata or {} self._sio = socketio.Client( reconnection=True, reconnection_attempts=0, reconnection_delay=2, reconnection_delay_max=30, logger=True, engineio_logger=True, ) self._token_data = None self._registered = False self._running = False self._heartbeat_thread = None self._initial_connected = False self._reconnect_lock = threading.Lock() self._SetupEvents() def _SetupEvents(self): @self._sio.on('connect', namespace='/msync') def on_connect(): logger.info(f'[SIO] ← connect ns=/msync') if self._initial_connected: logger.info(f'[SIO] 重连 detected, 重新注册/认证') t = threading.Thread(target=self._DoReconnect, daemon=True) t.start() @self._sio.on('disconnect', namespace='/msync') def on_disconnect(): logger.warning(f'[SIO] ← disconnect ns=/msync') self._registered = False self._token_data = None logger.info(f'[SIO] 状态已清理: registered=False, token_data=None') @self._sio.on('msync:service:online', namespace='/msync') def on_service_online(data): logger.info(f'[SIO] ← msync:service:online data={data}') @self._sio.on('msync:service:offline', namespace='/msync') def on_service_offline(data): logger.info(f'[SIO] ← msync:service:offline data={data}') @self._sio.on('msync:force:deregister', namespace='/msync') def on_force_deregister(data): logger.warning(f'[SIO] ← msync:force:deregister data={data}') self._registered = False self._token_data = None # 延迟后自动重新注册 t = threading.Thread(target=self._DoReconnect, daemon=True) t.start() @self._sio.on('*') def catch_all(event, data): logger.info(f'[SIO] ← unhandled event="{event}" data={data}') # ---------- 连接管理 ---------- def Connect(self): self._running = True url = self.MasterURL try: logger.info(f'[SIO] → connect url={url} ns=/msync') self._sio.connect(url, namespaces=['/msync'], wait=True, wait_timeout=15) except Exception as e: logger.warning(f'Socket.IO 连接失败: {e}，启动兜底重连线程') self._StartBootstrap() return self._OnReady() def _StartBootstrap(self): """初次连接失败的兜底：循环重连，成功后走完整注册/认证流程 python-socketio 的 reconnection 仅在"已连接后断开"时生效， 初次 connect 失败不会自动重连，需自行兜底。 """ def bootstrap(): while self._running and not self._sio.connected: try: logger.info(f'[SIO] 兜底重连尝试: {self.MasterURL}') self._sio.connect( self.MasterURL, namespaces=['/msync'], wait=True, wait_timeout=10, ) break except Exception as e: logger.warning(f'[SIO] 兜底重连失败: {e}') time.sleep(3) if self._sio.connected: self._OnReady() threading.Thread(target=bootstrap, daemon=True, name='msync-bootstrap').start() def _OnReady(self): """连接建立后的统一就绪流程：注册/认证/心跳/ORM 中转""" logger.info(f'[SIO] Socket 已连接, 开始注册/认证...') for attempt in range(1, 4): reg_ok = self._DoRegister() auth_ok = self._DoAuth() if reg_ok and auth_ok: break logger.warning(f'[SIO] 首次注册/认证失败 第 {attempt}/3 次 (reg={reg_ok}, auth={auth_ok})') if attempt < 3: time.sleep(2) self._DoHeartbeat() self._initial_connected = True self._StartHeartbeat() # 注册 ORM 中转处理器 try: from gvsdsdk.msync.orm_handler import ORMRelayHandler self._orm_handler = ORMRelayHandler() self._orm_handler.Register(self) except Exception as e: logger.warning(f'[ORM] ORM 中转处理器注册失败: {e}') logger.info(f'MSYNC Socket.IO 客户端已启动: {self.ServiceName}') def Disconnect(self): self._running = False try: self._DoDeregister() except Exception: pass try: self._sio.disconnect() except Exception: pass logger.info(f'MSYNC Socket.IO 客户端已断开: {self.ServiceName}') def _DoReconnect(self, max_retries=3): """重连后重新注册/认证，失败时自动重试（内部持有互斥锁）""" if not self._reconnect_lock.acquire(blocking=False): logger.warning('[SIO] 重连已在进行中，跳过本次重连') return try: for attempt in range(1, max_retries + 1): logger.info(f'[SIO] 重连注册/认证 第 {attempt}/{max_retries} 次') reg_ok = self._DoRegister() auth_ok = self._DoAuth() if reg_ok and auth_ok: logger.info(f'[SIO] 重连注册/认证成功') return logger.warning(f'[SIO] 重连注册/认证失败 (reg={reg_ok}, auth={auth_ok})，{2}s 后重试') if attempt < max_retries: time.sleep(2) logger.error(f'[SIO] 重连注册/认证 {max_retries} 次均失败，等待 Socket.IO 下次重连触发') finally: self._reconnect_lock.release() def _StartHeartbeat(self): if self._heartbeat_thread and self._heartbeat_thread.is_alive(): return self._heartbeat_thread = threading.Thread( target=self._HeartbeatLoop, daemon=True, name=f'msync-heartbeat-{self.ServiceName}', ) self._heartbeat_thread.start() logger.info(f'心跳线程已启动, 间隔={self.HeartbeatInterval}s') def _HeartbeatLoop(self): consecutive_failures = 0 while self._running: time.sleep(self.HeartbeatInterval) try: if not self._sio.connected: continue # 检查令牌是否即将过期，主动续期 if not self.IsAuthenticated: logger.info('[SIO] 令牌已过期或即将过期，主动续期') try: self.EnsureAuthenticated() except Exception as e: logger.warning(f'[SIO] 令牌续期失败: {e}，触发重连') consecutive_failures += 1 if consecutive_failures >= 3: logger.warning('[SIO] 连续失败次数过多，强制重连') self._ForceReconnect() consecutive_failures = 0 continue # 发送心跳 result = self._Call('msync:heartbeat', {}) if result and result.get('code') not in (200, 0): code = result.get('code') msg = result.get('msg', '') logger.warning(f'[SIO] 心跳异常: code={code} msg={msg}') # 404 = 服务不存在/已禁用，需要立即重新注册 if code == 404: logger.warning('[SIO] 服务已被注销，立即重新注册') try: reg_ok = self._DoRegister() auth_ok = self._DoAuth() if reg_ok and auth_ok: logger.info('[SIO] 重新注册/认证成功') consecutive_failures = 0 continue except Exception as e: logger.error(f'[SIO] 重新注册失败: {e}') # 注册也失败，强制重连 self._ForceReconnect() consecutive_failures = 0 continue consecutive_failures += 1 if consecutive_failures >= 3: logger.warning('[SIO] 心跳连续失败次数过多，强制重连') self._ForceReconnect() consecutive_failures = 0 elif result: consecutive_failures = 0 except Exception as e: logger.warning(f'[SIO] 心跳异常: {e}') consecutive_failures += 1 if consecutive_failures >= 3: self._ForceReconnect() consecutive_failures = 0 def _Call(self, event, data, timeout=15): if not self._sio.connected: logger.warning(f'[SIO] ✗ 未连接，无法发送 event={event}') return None try: logger.info(f'[SIO] → event="{event}" ns=/msync data_keys={list(data.keys()) if isinstance(data, dict) else type(data).__name__}') result = self._sio.call(event, data, namespace='/msync', timeout=timeout) logger.info(f'[SIO] ← ack event="{event}" result={result}') return result except socketio.exceptions.TimeoutError: logger.error(f'[SIO] ✗ timeout event="{event}" ({timeout}s)') return None except Exception as e: logger.error(f'[SIO] ✗ call failed event="{event}": {type(e).__name__}: {e}') return None # ---------- 注册与认证 ---------- def _DoRegister(self): payload = { 'ServiceName': self.ServiceName, 'ServiceDomain': self.ServiceDomain, 'ServiceSecretHash': hashlib.sha256(self.ServiceSecret.encode()).hexdigest(), 'ServiceSecret': self.ServiceSecret, 'ServiceScope': self.Scope, 'ServiceType': self.ServiceType, 'ServiceMetadata': self.Metadata, } result = self._Call('msync:register', payload) if result and result.get('code') == 200: self._registered = True logger.info(f'服务注册成功: {self.ServiceName}') return True if result and result.get('code') == 409: logger.warning(f'服务已存在(409冲突)，可能已用不同密钥注册: {self.ServiceName}，尝试认证') return True logger.warning(f'服务注册失败: result={result}') return False def _DoAuth(self): challenge = GenerateChallenge() payload = BuildAuthPayload(self.ServiceName, challenge) signature = SignPayload(payload, self.ServiceSecret) # 调试：打印签名详情 import json canonical = json.dumps(payload, sort_keys=True, separators=(',', ':')) logger.info(f'[AUTH] 客户端签名: service={self.ServiceName}, canonical={canonical}, sig={signature}') result = self._Call('msync:auth', { 'ServiceName': self.ServiceName, 'Payload': payload, 'Signature': signature, }) if result and result.get('code') == 200 and result.get('data'): token_data = result['data'] self._token_data = { 'AccessToken': token_data.get('AccessToken'), 'RefreshToken': token_data.get('RefreshToken'), 'Scope': token_data.get('Scope', []), 'ExpiresAt': token_data.get('ExpiresAt'), } logger.info(f'服务认证成功: {self.ServiceName}') return True logger.warning(f'服务认证失败: result={result}') return False def _DoDeregister(self): self._Call('msync:deregister', {'ServiceName': self.ServiceName}) self._registered = False def _DoHeartbeat(self): result = self._Call('msync:heartbeat', {}) if result and result.get('code') == 200: logger.info(f'初始心跳成功: {self.ServiceName}') else: logger.warning(f'初始心跳失败: {result}') # ---------- 状态 ---------- @property def AccessToken(self): return self._token_data.get('AccessToken') if self._token_data else None @property def RefreshToken(self): return self._token_data.get('RefreshToken') if self._token_data else None @property def IsAuthenticated(self): if not self._token_data: return False expires = self._token_data.get('ExpiresAt', 0) if expires and time.time() > expires - 60: return False return True @property def IsRegistered(self): return self._registered @property def IsConnected(self): return self._sio.connected def EnsureAuthenticated(self): """确保认证有效，过期则续期，续期失败则断开重连""" if not self.IsAuthenticated: if self._token_data and self._token_data.get('RefreshToken'): result = self._RefreshToken() if result: return result logger.warning('[SIO] 令牌刷新失败，尝试完整认证') auth_ok = self._DoAuth() if auth_ok: return self._token_data # 认证彻底失败，断开连接触发 Socket.IO 自动重连 logger.error('[SIO] 完整认证也失败，断开连接触发自动重连') self._ForceReconnect() return None return self._token_data def _RefreshToken(self): result = self._Call('msync:auth:refresh', { 'RefreshToken': self._token_data.get('RefreshToken', ''), }) if result and result.get('code') == 200 and result.get('data'): token_data = result['data'] self._token_data = { 'AccessToken': token_data.get('AccessToken'), 'RefreshToken': token_data.get('RefreshToken'), 'Scope': token_data.get('Scope', []), 'ExpiresAt': token_data.get('ExpiresAt'), } logger.info(f'令牌刷新成功: {self.ServiceName}') return self._token_data logger.info('令牌刷新失败，回退到完整认证') return None def _ForceReconnect(self): """强制断开连接，让 Socket.IO 自动重连""" try: self._registered = False self._token_data = None self._sio.disconnect() except Exception as e: logger.warning(f'[SIO] 强制断开失败: {e}') def _Headers(self): token = self.AccessToken if not token: self.EnsureAuthenticated() token = self.AccessToken return {'Authorization': f'Bearer {token}'} if token else {} # ---------- 服务发现 ---------- def DiscoverAll(self, active_only=True): result = self._Call('msync:discover', {'ActiveOnly': active_only}) if result and result.get('code') == 200: return result.get('data', []) return [] def Discover(self, service_name): result = self._Call('msync:discover:one', {'ServiceName': service_name}) if result and result.get('code') == 200: return result.get('data') return None # ---------- OGM ---------- def OGMRegister(self, object_type, local_id, service_name, global_id=None, metadata=None): body = { 'ObjectType': object_type, 'ObjectLocalID': str(local_id), 'ServiceName': service_name, } if global_id: body['ObjectGlobalID'] = global_id if metadata: body['Metadata'] = metadata result = self._Call('msync:ogm:register', body) if result and result.get('code') == 200: return result.get('data') return None def OGMResolve(self, global_id): result = self._Call('msync:ogm:resolve', {'ObjectGlobalID': global_id}) if result and result.get('code') == 200: return result.get('data') return None def OGMQuery(self, object_type=None, service_name=None, local_id=None): body = {} if object_type: body['ObjectType'] = object_type if service_name: body['ServiceName'] = service_name if local_id: body['ObjectLocalID'] = local_id result = self._Call('msync:ogm:query', body) if result and result.get('code') == 200: return result.get('data') return None def OGMDeregister(self, global_id): result = self._Call('msync:ogm:deregister', {'ObjectGlobalID': global_id}) return result is not None and result.get('code') == 200 # ---------- 事务 ---------- def TxCreate(self, name, steps, **kwargs): body = {'TransactionName': name, 'Steps': steps} for key in ('TransactionDescription', 'TransactionType', 'InitiatorServiceName', 'ReferenceOGMUUID', 'Payload', 'TimeoutSeconds'): kwarg_key = key[0].lower() + key[1:] if kwarg_key in kwargs: body[key] = kwargs.pop(kwarg_key) body.update(kwargs) result = self._Call('msync:tx:create', body, timeout=30) if result and result.get('code') == 200: return result.get('data') return None def TxExecute(self, transaction_code, access_token=None): result = self._Call('msync:tx:execute', { 'TransactionCode': transaction_code, 'AccessToken': access_token, }, timeout=60) if result and result.get('code') == 200: return result.get('data') return None def TxDetail(self, transaction_code): result = self._Call('msync:tx:detail', {'TransactionCode': transaction_code}) if result and result.get('code') == 200: return result.get('data') return None # ---------- 远程调用 ---------- def CallRemote(self, service_name, method, path, data=None, json=None, **kwargs): import requests service_info = self.Discover(service_name) if not service_info: logger.error(f'目标服务未发现: {service_name}') return None domain = service_info.get('ServiceDomain', '') url = f'{domain.rstrip("/")}{path}' kwargs.setdefault('headers', {}).update(self._Headers()) kwargs.setdefault('timeout', 30) method_func = getattr(requests, method.lower(), requests.get) return method_func(url, data=data, json=json, **kwargs) # ---------- 应用清单上报（远端 App 注入） ---------- def ReportAppManifests(self, manifests, service_domain=None): """上报应用清单到主服务器（远端 App 注入 GERP 菜单机制） Args: manifests: 清单 dict 列表，结构与主服务器 AppManifest 一致： {AppCode, AppName, AppIcon, AppDesc, AppType, FrontendRoute, FrontendType, LegacySrc, Menus: [{MenuName, MenuTitle, MenuLink, MenuIcon, MenuSort, ParentMenuName}], ConfigKeys: []} service_domain: 本服务对外域名（缺省用注册时 ServiceDomain， 页面 iframe 加载 base） Returns: (ok, result)：ok=True 时 result={Registered:[], Failed:[]}； ok=False 时 result 为原始响应 dict """ body = { 'ServiceName': self.ServiceName, 'ServiceDomain': service_domain or self.ServiceDomain, 'Manifests': manifests, } result = self._Call('msync:app:manifest:report', body, timeout=30) if result and result.get('code') == 200: return True, result.get('data', {}) return False, result