socket: fix error
This commit is contained in:
+47
-43
@@ -87,6 +87,49 @@ export const makeSocket = (config: SocketConfig) => {
|
|||||||
url.searchParams.append('ED', authState.creds.routingInfo.toString('base64url'))
|
url.searchParams.append('ED', authState.creds.routingInfo.toString('base64url'))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** ephemeral key pair used to encrypt/decrypt communication. Unique for each connection */
|
||||||
|
const ephemeralKeyPair = Curve.generateKeyPair()
|
||||||
|
/** WA noise protocol wrapper */
|
||||||
|
const noise = makeNoiseHandler({
|
||||||
|
keyPair: ephemeralKeyPair,
|
||||||
|
NOISE_HEADER: NOISE_WA_HEADER,
|
||||||
|
logger,
|
||||||
|
routingInfo: authState?.creds?.routingInfo
|
||||||
|
})
|
||||||
|
|
||||||
|
const ws = new WebSocketClient(url, config)
|
||||||
|
|
||||||
|
ws.connect()
|
||||||
|
|
||||||
|
|
||||||
|
const sendPromise = promisify(ws.send)
|
||||||
|
/** send a raw buffer */
|
||||||
|
const sendRawMessage = async (data: Uint8Array | Buffer) => {
|
||||||
|
if (!ws.isOpen) {
|
||||||
|
throw new Boom('Connection Closed', { statusCode: DisconnectReason.connectionClosed })
|
||||||
|
}
|
||||||
|
|
||||||
|
const bytes = noise.encodeFrame(data)
|
||||||
|
await promiseTimeout<void>(connectTimeoutMs, async (resolve, reject) => {
|
||||||
|
try {
|
||||||
|
await sendPromise.call(ws, bytes)
|
||||||
|
resolve()
|
||||||
|
} catch (error) {
|
||||||
|
reject(error)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
/** send a binary node */
|
||||||
|
const sendNode = (frame: BinaryNode) => {
|
||||||
|
if (logger.level === 'trace') {
|
||||||
|
logger.trace({ xml: binaryNodeToString(frame), msg: 'xml send' })
|
||||||
|
}
|
||||||
|
|
||||||
|
const buff = encodeBinaryNode(frame)
|
||||||
|
return sendRawMessage(buff)
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Wait for a message with a certain tag to be received
|
* Wait for a message with a certain tag to be received
|
||||||
* @param msgId the message tag to await
|
* @param msgId the message tag to await
|
||||||
@@ -143,12 +186,13 @@ export const makeSocket = (config: SocketConfig) => {
|
|||||||
const msgId = node.attrs.id
|
const msgId = node.attrs.id
|
||||||
|
|
||||||
const result = await promiseTimeout<any>(timeoutMs, async (resolve, reject) => {
|
const result = await promiseTimeout<any>(timeoutMs, async (resolve, reject) => {
|
||||||
const result = await waitForMessage(msgId, timeoutMs).catch(reject)
|
const result = waitForMessage(msgId, timeoutMs).catch(reject)
|
||||||
sendNode(node)
|
sendNode(node)
|
||||||
.then(() => resolve(result))
|
.then(async () => resolve(await result))
|
||||||
.catch(reject)
|
.catch(reject)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
||||||
if (result && 'tag' in result) {
|
if (result && 'tag' in result) {
|
||||||
assertNodeErrorFree(result)
|
assertNodeErrorFree(result)
|
||||||
}
|
}
|
||||||
@@ -230,20 +274,7 @@ export const makeSocket = (config: SocketConfig) => {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
const ws = new WebSocketClient(url, config)
|
|
||||||
|
|
||||||
ws.connect()
|
|
||||||
|
|
||||||
const ev = makeEventBuffer(logger)
|
const ev = makeEventBuffer(logger)
|
||||||
/** ephemeral key pair used to encrypt/decrypt communication. Unique for each connection */
|
|
||||||
const ephemeralKeyPair = Curve.generateKeyPair()
|
|
||||||
/** WA noise protocol wrapper */
|
|
||||||
const noise = makeNoiseHandler({
|
|
||||||
keyPair: ephemeralKeyPair,
|
|
||||||
NOISE_HEADER: NOISE_WA_HEADER,
|
|
||||||
logger,
|
|
||||||
routingInfo: authState?.creds?.routingInfo
|
|
||||||
})
|
|
||||||
|
|
||||||
const { creds } = authState
|
const { creds } = authState
|
||||||
// add transaction capability
|
// add transaction capability
|
||||||
@@ -256,34 +287,6 @@ export const makeSocket = (config: SocketConfig) => {
|
|||||||
let qrTimer: NodeJS.Timeout
|
let qrTimer: NodeJS.Timeout
|
||||||
let closed = false
|
let closed = false
|
||||||
|
|
||||||
const sendPromise = promisify(ws.send)
|
|
||||||
/** send a raw buffer */
|
|
||||||
const sendRawMessage = async (data: Uint8Array | Buffer) => {
|
|
||||||
if (!ws.isOpen) {
|
|
||||||
throw new Boom('Connection Closed', { statusCode: DisconnectReason.connectionClosed })
|
|
||||||
}
|
|
||||||
|
|
||||||
const bytes = noise.encodeFrame(data)
|
|
||||||
await promiseTimeout<void>(connectTimeoutMs, async (resolve, reject) => {
|
|
||||||
try {
|
|
||||||
await sendPromise.call(ws, bytes)
|
|
||||||
resolve()
|
|
||||||
} catch (error) {
|
|
||||||
reject(error)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
/** send a binary node */
|
|
||||||
const sendNode = (frame: BinaryNode) => {
|
|
||||||
if (logger.level === 'trace') {
|
|
||||||
logger.trace({ xml: binaryNodeToString(frame), msg: 'xml send' })
|
|
||||||
}
|
|
||||||
|
|
||||||
const buff = encodeBinaryNode(frame)
|
|
||||||
return sendRawMessage(buff)
|
|
||||||
}
|
|
||||||
|
|
||||||
/** log & process any unexpected errors */
|
/** log & process any unexpected errors */
|
||||||
const onUnexpectedError = (err: Error | Boom, msg: string) => {
|
const onUnexpectedError = (err: Error | Boom, msg: string) => {
|
||||||
logger.error({ err }, `unexpected error in '${msg}'`)
|
logger.error({ err }, `unexpected error in '${msg}'`)
|
||||||
@@ -409,6 +412,7 @@ export const makeSocket = (config: SocketConfig) => {
|
|||||||
|
|
||||||
// Upload to server (outside transaction, can fail without affecting local keys)
|
// Upload to server (outside transaction, can fail without affecting local keys)
|
||||||
try {
|
try {
|
||||||
|
console.log("LOG", node)
|
||||||
await query(node)
|
await query(node)
|
||||||
logger.info({ count }, 'uploaded pre-keys successfully')
|
logger.info({ count }, 'uploaded pre-keys successfully')
|
||||||
lastUploadTime = Date.now()
|
lastUploadTime = Date.now()
|
||||||
|
|||||||
Reference in New Issue
Block a user