优化同步逻辑
This commit is contained in:
@@ -1,14 +1,15 @@
|
||||
import WebSocket from 'ws'
|
||||
import { encryptMsg, decryptMsg } from './utils'
|
||||
import * as modules from './modules'
|
||||
import { modules, callObj } from './modules'
|
||||
// import { action as commonAction } from '@root/store/modules/common'
|
||||
// import { getStore } from '@root/store'
|
||||
import registerSyncListHandler from './syncList'
|
||||
// import registerSyncListHandler from './syncList'
|
||||
import log from '../log'
|
||||
import { SYNC_CLOSE_CODE, SYNC_CODE } from '@common/constants'
|
||||
import { dateFormat } from '@common/utils/common'
|
||||
import { aesEncrypt, getAddress } from '../utils'
|
||||
import { sendClientStatus } from '@main/modules/winMain'
|
||||
import { createMsg2call } from 'message2call'
|
||||
|
||||
let status: LX.Sync.ClientStatus = {
|
||||
status: false,
|
||||
@@ -31,8 +32,14 @@ export const sendSyncMessage = (message: string) => {
|
||||
}
|
||||
|
||||
const handleConnection = (socket: LX.Sync.Client.Socket) => {
|
||||
for (const moduleInit of Object.values(modules)) {
|
||||
moduleInit(socket)
|
||||
for (const { registerEvent } of Object.values(modules)) {
|
||||
registerEvent(socket)
|
||||
}
|
||||
}
|
||||
|
||||
const handleDisconnection = () => {
|
||||
for (const { unregisterEvent } of Object.values(modules)) {
|
||||
unregisterEvent()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -123,7 +130,7 @@ const heartbeatTools = {
|
||||
let client: LX.Sync.Client.Socket | null
|
||||
// let listSyncPromise: Promise<void>
|
||||
export const connect = (urlInfo: LX.Sync.Client.UrlInfo, keyInfo: LX.Sync.ClientKeyInfo) => {
|
||||
client = new WebSocket(`${urlInfo.wsProtocol}//${urlInfo.hostPath}?i=${encodeURIComponent(keyInfo.clientId)}&t=${encodeURIComponent(aesEncrypt(SYNC_CODE.msgConnect, keyInfo.key))}`, {
|
||||
client = new WebSocket(`${urlInfo.wsProtocol}//${urlInfo.hostPath}/socket?i=${encodeURIComponent(keyInfo.clientId)}&t=${encodeURIComponent(aesEncrypt(SYNC_CODE.msgConnect, keyInfo.key))}`, {
|
||||
}) as LX.Sync.Client.Socket
|
||||
client.data = {
|
||||
keyInfo,
|
||||
@@ -131,39 +138,67 @@ export const connect = (urlInfo: LX.Sync.Client.UrlInfo, keyInfo: LX.Sync.Client
|
||||
}
|
||||
heartbeatTools.connect(client)
|
||||
|
||||
// listSyncPromise = registerSyncListHandler(socket)
|
||||
let events: Partial<{ [K in keyof LX.Sync.ActionSyncSendType]: Array<(data: LX.Sync.ActionSyncSendType[K]) => (void | Promise<void>)> }> = {}
|
||||
let closeEvents: Array<(err: Error) => (void | Promise<void>)> = []
|
||||
|
||||
const message2read = createMsg2call({
|
||||
funcsObj: {
|
||||
...callObj,
|
||||
list_sync_finished() {
|
||||
log.info('sync list success')
|
||||
client!.isReady = true
|
||||
handleConnection(client as LX.Sync.Client.Socket)
|
||||
sendSyncStatus({
|
||||
status: true,
|
||||
message: '',
|
||||
})
|
||||
heartbeatTools.failedNum = 0
|
||||
},
|
||||
},
|
||||
timeout: 120 * 1000,
|
||||
sendMessage(data) {
|
||||
void encryptMsg(keyInfo, JSON.stringify(data)).then((data) => {
|
||||
client?.send(data)
|
||||
}).catch((err) => {
|
||||
log.error('encrypt msg error: ', err)
|
||||
client?.close(SYNC_CLOSE_CODE.failed)
|
||||
})
|
||||
},
|
||||
onCallBeforeParams(rawArgs) {
|
||||
return [client, ...rawArgs]
|
||||
},
|
||||
onError(error, path, groupName) {
|
||||
const name = groupName ?? ''
|
||||
log.error(`sync call ${name} ${path.join('.')} error:`, error)
|
||||
if (groupName == null) return
|
||||
client?.close(SYNC_CLOSE_CODE.failed)
|
||||
sendSyncStatus({
|
||||
status: false,
|
||||
message: error.message,
|
||||
})
|
||||
},
|
||||
})
|
||||
|
||||
client.remoteSyncList = message2read.createSyncRemote('list')
|
||||
|
||||
client.addEventListener('message', ({ data }) => {
|
||||
if (data == 'ping') return
|
||||
if (typeof data === 'string') {
|
||||
let syncData: LX.Sync.ActionSync
|
||||
try {
|
||||
syncData = JSON.parse(decryptMsg(keyInfo, data))
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
const handlers = events[syncData.action]
|
||||
if (handlers) {
|
||||
// @ts-expect-error
|
||||
for (const handler of handlers) void handler(syncData.data)
|
||||
}
|
||||
void decryptMsg(keyInfo, data).then((data) => {
|
||||
let syncData: LX.Sync.ServerActions
|
||||
try {
|
||||
syncData = JSON.parse(data)
|
||||
} catch (err) {
|
||||
log.error('parse msg error: ', err)
|
||||
client?.close(SYNC_CLOSE_CODE.failed)
|
||||
return
|
||||
}
|
||||
message2read.onMessage(syncData)
|
||||
}).catch((error) => {
|
||||
log.error('decrypt msg error: ', error)
|
||||
client?.close(SYNC_CLOSE_CODE.failed)
|
||||
})
|
||||
}
|
||||
})
|
||||
client.onRemoteEvent = function(eventName, handler) {
|
||||
let eventArr = events[eventName]
|
||||
if (!eventArr) events[eventName] = eventArr = []
|
||||
// let eventArr = events.get(eventName)
|
||||
// if (!eventArr) events.set(eventName, eventArr = [])
|
||||
eventArr.push(handler)
|
||||
|
||||
return () => {
|
||||
eventArr!.splice(eventArr!.indexOf(handler), 1)
|
||||
}
|
||||
}
|
||||
client.sendData = function(eventName, data, callback) {
|
||||
client?.send(encryptMsg(keyInfo, JSON.stringify({ action: eventName, data })), callback)
|
||||
}
|
||||
client.onClose = function(handler: typeof closeEvents[number]) {
|
||||
closeEvents.push(handler)
|
||||
return () => {
|
||||
@@ -171,6 +206,7 @@ export const connect = (urlInfo: LX.Sync.Client.UrlInfo, keyInfo: LX.Sync.Client
|
||||
}
|
||||
}
|
||||
|
||||
const initMessage = 'Wait syncing...'
|
||||
client.addEventListener('open', () => {
|
||||
log.info('connect')
|
||||
// const store = getStore()
|
||||
@@ -178,39 +214,14 @@ export const connect = (urlInfo: LX.Sync.Client.UrlInfo, keyInfo: LX.Sync.Client
|
||||
client!.isReady = false
|
||||
sendSyncStatus({
|
||||
status: false,
|
||||
message: 'Wait syncing...',
|
||||
})
|
||||
void registerSyncListHandler(client as LX.Sync.Client.Socket).then(() => {
|
||||
log.info('sync list success')
|
||||
handleConnection(client as LX.Sync.Client.Socket)
|
||||
log.info('register list sync service success')
|
||||
client!.isReady = true
|
||||
heartbeatTools.failedNum = 0
|
||||
sendSyncStatus({
|
||||
status: true,
|
||||
message: '',
|
||||
})
|
||||
}).catch(err => {
|
||||
if (err.message == 'closed') {
|
||||
sendSyncStatus({
|
||||
status: false,
|
||||
message: '',
|
||||
})
|
||||
} else {
|
||||
console.log(err)
|
||||
log.r_error(err.stack)
|
||||
sendSyncStatus({
|
||||
status: false,
|
||||
message: err.message,
|
||||
})
|
||||
}
|
||||
message: initMessage,
|
||||
})
|
||||
})
|
||||
client.addEventListener('close', ({ code }) => {
|
||||
const err = new Error('closed')
|
||||
for (const handler of closeEvents) void handler(err)
|
||||
handleDisconnection()
|
||||
closeEvents = []
|
||||
events = {}
|
||||
switch (code) {
|
||||
case SYNC_CLOSE_CODE.normal:
|
||||
// case SYNC_CLOSE_CODE.failed:
|
||||
@@ -218,6 +229,15 @@ export const connect = (urlInfo: LX.Sync.Client.UrlInfo, keyInfo: LX.Sync.Client
|
||||
status: false,
|
||||
message: '',
|
||||
})
|
||||
break
|
||||
case SYNC_CLOSE_CODE.failed:
|
||||
if (!status.message || status.message == initMessage) {
|
||||
sendSyncStatus({
|
||||
status: false,
|
||||
message: 'failed',
|
||||
})
|
||||
}
|
||||
break
|
||||
}
|
||||
})
|
||||
client.addEventListener('error', ({ message }) => {
|
||||
|
||||
Reference in New Issue
Block a user