ion/Melo
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

完善同步

Browse Source
This commit is contained in:
lyswhut
2023-03-01 18:44:51 +08:00
parent e75ab3a7d7
commit 3399333360
14 changed files with 89 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/modules/sync/client/client.ts
+38 -14
View File
@@ -38,6 +38,7 @@ const handleConnection = (socket: LX.Sync.Client.Socket) => {
const heartbeatTools = {
failedNum: 0,
maxTryNum: 3,
pingTimeout: null as NodeJS.Timeout | null,
delayRetryTimeout: null as NodeJS.Timeout | null,
handleOpen() {
@@ -64,16 +65,21 @@ const heartbeatTools = {
// client = null
if (!client) return
if (this.failedNum > 3) throw new Error('connect error')
if (++this.failedNum > this.maxTryNum) {
this.failedNum = 0
throw new Error('connect error')
}
this.delayRetryTimeout = setTimeout(() => {
this.delayRetryTimeout = null
if (!client) return
console.log(dateFormat(new Date()), 'reconnnect...')
sendSyncStatus({
status: false,
message: `Try reconnnect... (${this.failedNum}/${this.maxTryNum})`,
})
connect(client.data.urlInfo, client.data.keyInfo)
}, 2000)
this.failedNum++
},
clearTimeout() {
if (this.delayRetryTimeout) {
@@ -176,23 +182,40 @@ export const connect = (urlInfo: LX.Sync.Client.UrlInfo, keyInfo: LX.Sync.Client
message: '',
})
}).catch(err => {
console.log(err)
log.r_error(err.stack)
sendSyncStatus({
status: false,
message: err.message,
})
if (err.message == 'closed') {
sendSyncStatus({
status: false,
message: '',
})
} else {
console.log(err)
log.r_error(err.stack)
sendSyncStatus({
status: false,
message: err.message,
})
}
})
})
client.addEventListener('close', () => {
sendSyncStatus({
status: false,
message: '',
})
client.addEventListener('close', ({ code }) => {
const err = new Error('closed')
for (const handler of closeEvents) void handler(err)
closeEvents = []
events = {}
switch (code) {
case SYNC_CLOSE_CODE.normal:
// case SYNC_CLOSE_CODE.failed:
sendSyncStatus({
status: false,
message: '',
})
}
})
client.addEventListener('error', ({ message }) => {
sendSyncStatus({
status: false,
message,
})
})
}
@@ -202,6 +225,7 @@ export const disconnect = async() => {
client.close(SYNC_CLOSE_CODE.normal)
client = null
heartbeatTools.clearTimeout()
heartbeatTools.failedNum = 0
}
export const getStatus = (): LX.Sync.ClientStatus => status
src/main/modules/sync/client/index.ts
+9 -6
View File
@@ -5,13 +5,18 @@ import { SYNC_CODE } from '@common/constants'
import log from '../log'
import { parseUrl } from './utils'
let connectId = 0
const handleConnect = async(host: string, authCode?: string) => {
// const hostInfo = await getSyncHost()
// console.log(hostInfo)
// if (!hostInfo || !hostInfo.host || !hostInfo.port) throw new Error(SYNC_CODE.unknownServiceAddress)
const id = connectId
const urlInfo = parseUrl(host)
await disconnectServer(false)
if (id != connectId) return
const keyInfo = await handleAuth(urlInfo, authCode)
if (id != connectId) return
socketConnect(urlInfo, keyInfo)
}
const handleDisconnect = async() => {
@@ -23,12 +28,9 @@ const connectServer = async(host: string, authCode?: string) => {
status: false,
message: SYNC_CODE.connecting,
})
return handleConnect(host, authCode).then(() => {
sendSyncStatus({
status: true,
message: '',
})
}).catch(async err => {
const id = connectId
return handleConnect(host, authCode).catch(async err => {
if (id != connectId) return
sendSyncStatus({
status: false,
message: err.message,
@@ -49,6 +51,7 @@ const connectServer = async(host: string, authCode?: string) => {
const disconnectServer = async(isResetStatus = true) => handleDisconnect().then(() => {
log.info('disconnect...')
if (isResetStatus) {
connectId++
sendSyncStatus({
status: false,
message: '',
src/main/modules/sync/data.ts
+9 -5
View File
@@ -9,15 +9,15 @@ import log from './log'
export const getSyncAuthKey = async(serverId: string) => {
const store = getStore(STORE_NAMES.SYNC)
const keys = store.get('syncAuthKey') as Record<string, LX.Sync.ClientKeyInfo> | null
const keys = store.get('syncAuthKey_v3') as Record<string, LX.Sync.ClientKeyInfo> | null
if (!keys) return null
return keys[serverId] ?? null
}
export const setSyncAuthKey = async(serverId: string, info: LX.Sync.ClientKeyInfo) => {
const store = getStore(STORE_NAMES.SYNC)
let keys: Record<string, LX.Sync.ClientKeyInfo> = (store.get('syncAuthKey') as Record<string, LX.Sync.ClientKeyInfo> | null) ?? {}
let keys: Record<string, LX.Sync.ClientKeyInfo> = (store.get('syncAuthKey_v3') as Record<string, LX.Sync.ClientKeyInfo> | null) ?? {}
keys[serverId] = info
store.set('syncAuthKey', keys)
store.set('syncAuthKey_v3', keys)
}
let syncHost: string
@@ -73,7 +73,7 @@ const devicesInfo: DevicesInfo = { serverId: '', clients: {}, snapshotInfo: { la
let deviceKeys: string[] = []
const saveDevicesInfoThrottle = throttle(() => {
const store = getStore(STORE_NAMES.SYNC)
store.set('keys', devicesInfo.clients)
store.set('clients', devicesInfo.clients)
})
const initDeviceInfo = () => {
@@ -82,7 +82,8 @@ const initDeviceInfo = () => {
if (serverId) devicesInfo.serverId = serverId
else {
devicesInfo.serverId = randomBytes(4 * 4).toString('base64')
saveDevicesInfoThrottle()
const store = getStore(STORE_NAMES.SYNC)
store.set('serverId', devicesInfo.serverId)
}
const devices = store.get('clients') as DevicesInfo['clients'] | undefined
if (devices) devicesInfo.clients = devices
@@ -155,6 +156,7 @@ export const saveSnapshotInfo = (info: SnapshotInfo) => {
}
export const getSnapshot = async(name: string) => {
console.log('getSnapshot', name)
const filePath = path.join(global.lxDataPath, `snapshot_${name}`)
let listData: LX.Sync.ListData
try {
@@ -166,6 +168,7 @@ export const getSnapshot = async(name: string) => {
return listData
}
export const saveSnapshot = async(name: string, data: string) => {
console.log('saveSnapshot', name)
const filePath = path.join(global.lxDataPath, `snapshot_${name}`)
return fs.promises.writeFile(filePath, data).catch((err) => {
log.error(err)
@@ -173,6 +176,7 @@ export const saveSnapshot = async(name: string, data: string) => {
})
}
export const removeSnapshot = async(name: string) => {
console.log('removeSnapshot', name)
const filePath = path.join(global.lxDataPath, `snapshot_${name}`)
return fs.promises.unlink(filePath).catch((err) => {
log.error(err)
src/main/modules/sync/server/modules/list.ts
+5 -5
View File
@@ -2,8 +2,8 @@
// import { sendSyncActionList } from '@main/modules/winMain'
import { SYNC_CLOSE_CODE } from '@common/constants'
import { updateDeviceSnapshotKey } from '../../data'
import { handleRemoteListAction } from '../../utils'
import { createSnapshot, encryptMsg } from '../utils'
import { handleRemoteListAction, registerListActionEvent } from '../../utils'
import { createSnapshot, encryptMsg, getCurrentListInfoKey } from '../utils'
let wss: LX.Sync.Server.SocketServer | null
let removeListener: (() => void) | null
@@ -29,16 +29,16 @@ const broadcast = async(key: string, data: any, excludeIds: string[] = []) => {
}
}
export const sendListAction = async(action: LX.Sync.ActionList) => {
const sendListAction = async(action: LX.Sync.ActionList) => {
console.log('sendListAction', action.action)
// io.sockets
await broadcast('list:sync:action', action)
await broadcast(await getCurrentListInfoKey(), action)
}
export const registerListHandler = (_wss: LX.Sync.Server.SocketServer, socket: LX.Sync.Server.Socket) => {
if (!wss) {
wss = _wss
// removeListener = registerListActionEvent()
removeListener = registerListActionEvent(sendListAction)
}
socket.onRemoteEvent('list:sync:action', (action) => {
src/main/modules/sync/server/server.ts
+16 -4
View File
@@ -19,6 +19,7 @@ let status: LX.Sync.ServerStatus = {
code: '',
devices: [],
}
let stopingServer = false
const codeTools: {
timeout: NodeJS.Timer | null
@@ -148,6 +149,7 @@ const handleStartServer = async(port = 9527, ip = '0.0.0.0') => await new Promis
// const events = new Map<keyof ActionsType, Array<(err: Error | null, data: LX.Sync.ActionSyncType[keyof LX.Sync.ActionSyncType]) => void>>()
// const events = new Map<keyof LX.Sync.ActionSyncType, Array<(err: Error | null, data: LX.Sync.ActionSyncType[keyof LX.Sync.ActionSyncType]) => void>>()
let events: Partial<{ [K in keyof LX.Sync.ActionSyncType]: Array<(data: LX.Sync.ActionSyncType[K]) => void> }> = {}
let closeEvents: Array<(err: Error) => (void | Promise<void>)> = []
socket.addEventListener('message', ({ data }) => {
if (typeof data === 'string') {
let syncData: LX.Sync.ActionSync
@@ -167,11 +169,9 @@ const handleStartServer = async(port = 9527, ip = '0.0.0.0') => await new Promis
})
socket.addEventListener('close', () => {
const err = new Error('closed')
for (const handler of Object.values(events).flat()) {
// @ts-expect-error
handler(err, null)
}
for (const handler of closeEvents) void handler(err)
events = {}
closeEvents = []
if (!status.devices.length) handleUnconnection()
log.info('deconnection', socket.keyInfo.deviceName)
})
@@ -186,6 +186,12 @@ const handleStartServer = async(port = 9527, ip = '0.0.0.0') => await new Promis
eventArr!.splice(eventArr!.indexOf(handler), 1)
}
}
socket.onClose = function(handler: typeof closeEvents[number]) {
closeEvents.push(handler)
return () => {
closeEvents.splice(closeEvents.indexOf(handler), 1)
}
}
socket.sendData = function(eventName, data, callback) {
socket.send(encryptMsg(socket.keyInfo, JSON.stringify({ action: eventName, data })), callback)
}
@@ -248,6 +254,7 @@ const handleStartServer = async(port = 9527, ip = '0.0.0.0') => await new Promis
const handleStopServer = async() => new Promise<void>((resolve, reject) => {
if (!wss) return
for (const client of wss.clients) client.close(SYNC_CLOSE_CODE.normal)
wss.close()
wss = null
httpServer.close((err) => {
@@ -271,6 +278,9 @@ export const stopServer = async() => {
return
}
console.log('stoping sync server...')
status.message = 'stoping...'
sendServerStatus(status)
stopingServer = true
await handleStopServer().then(() => {
console.log('sync server stoped')
status.status = false
@@ -281,12 +291,14 @@ export const stopServer = async() => {
console.log(err)
status.message = err.message
}).finally(() => {
stopingServer = false
sendServerStatus(status)
})
}
export const startServer = async(port: number) => {
console.log('status.status', status.status)
if (stopingServer) return
if (status.status) await handleStopServer()
log.info('starting sync server')
src/main/modules/sync/server/utils.ts
+2 -7
View File
@@ -52,11 +52,6 @@ export const createSnapshot = async() => {
export const getCurrentListInfoKey = async() => {
if (!snapshotInfo) snapshotInfo = getSnapshotInfo()
if (snapshotInfo.latest) {
return snapshotInfo.latest
}
snapshotInfo.latest = toMD5(JSON.stringify(await getLocalListData()))
saveSnapshotInfo(snapshotInfo)
return snapshotInfo.latest
// if (!snapshotInfo) snapshotInfo = getSnapshotInfo()
return createSnapshot()
}