487 lines
19 KiB
Python
487 lines
19 KiB
Python
"""模块：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:
|
||
if not self._reconnect_lock.acquire(blocking=False):
|
||
logger.warning(f'[SIO] 重连已在进行中，跳过本次重连')
|
||
return
|
||
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._StartHeartbeat()
|
||
return
|
||
|
||
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):
|
||
"""重连后重新注册/认证，失败时自动重试
|
||
|
||
Args:
|
||
max_retries: 最大重试次数，默认 3 次
|
||
"""
|
||
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) |