franpcode/Baileys
1
0
Fork 0
mirror of https://github.com/FranP-code/Baileys.git synced 2025-10-13 00:32:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

More reliable connect with automatic retries + default connect options

Browse Source
This commit is contained in:
Adhiraj
2020-08-31 14:39:21 +05:30
parent acc8e864fa
commit 0af9f8fbe4
9 changed files with 232 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/Tests/Tests.Connect.ts
View File

@@ -12,7 +12,7 @@ describe('QR Generation', () => {
conn.removeAllListeners ('qr')
conn.on ('qr', qr => calledQR += 1)
await conn.connect({ timeoutMs: 15000 })
await conn.connect()
.then (() => assert.fail('should not have succeeded'))
.catch (error => {
assert.equal (error.message, 'timed out')
@@ -49,11 +49,14 @@ describe('Test Connect', () => {
conn.loadAuthInfo ('./auth_info.json')
let timeout = 0.1
while (true) {
setTimeout (() => conn.close(), timeout*1000)
let tmout = setTimeout (() => conn.close(), timeout*1000)
try {
await conn.connect ()
clearTimeout (tmout)
conn.close ()
break
} catch (error) {
@@ -64,9 +67,10 @@ describe('Test Connect', () => {
})
it('should reconnect', async () => {
const conn = new WAConnection()
conn.connectOptions.timeoutMs = 20*1000
await conn
.loadAuthInfo (auth)
.connect ({timeoutMs: 20*1000})
.connect ()
.then (conn => {
assert.ok(conn.user)
assert.ok(conn.user.jid)
@@ -135,7 +139,7 @@ describe ('Reconnects', () => {
closes += 1
// let it fail reconnect a few times
if (closes > 3) {
if (closes >= 1) {
conn.removeAllListeners ('close')
conn.removeAllListeners ('connecting')
resolve ()

41
src/WAConnection/0.Base.ts
View File

@@ -19,6 +19,7 @@ import {
WAChat,
WAQuery,
ReconnectMode,
WAConnectOptions,
} from './Constants'
import { EventEmitter } from 'events'
import KeyedDB from '@adiwajshing/keyed-db'
@@ -38,7 +39,12 @@ export class WAConnection extends EventEmitter {
state: WAConnectionState = 'close'
/** New QR generation interval, set to null if you don't want to regenerate */
regenerateQRIntervalMs = 30*1000
connectOptions: WAConnectOptions = {
timeoutMs: 60*1000,
waitForChats: true,
maxRetries: 5
}
/** When to auto-reconnect */
autoReconnect = ReconnectMode.onConnectionLost
/** Whether the phone is connected */
phoneConnected: boolean = false
@@ -46,6 +52,7 @@ export class WAConnection extends EventEmitter {
maxCachedMessages = 25
chats: KeyedDB<WAChat> = new KeyedDB (Utils.waChatUniqueKey, value => value.jid)
contacts: { [k: string]: WAContact } = {}
/** Data structure of tokens & IDs used to establish one's identiy to WhatsApp Web */
protected authInfo: AuthenticationCredentials = null
@@ -62,24 +69,30 @@ export class WAConnection extends EventEmitter {
protected referenceDate = new Date () // used for generating tags
protected lastSeen: Date = null // last keep alive received
protected qrTimeout: NodeJS.Timeout
protected phoneCheck: NodeJS.Timeout
protected lastDisconnectReason: DisconnectReason
protected cancelledReconnect = false
protected cancelReconnect: () => void
constructor () {
super ()
this.registerCallback (['Cmd', 'type:disconnect'], json => this.unexpectedDisconnect(json[1].kind || 'unknown'))
this.registerCallback (['Cmd', 'type:disconnect'], json => (
this.unexpectedDisconnect(json[1].kind || 'unknown')
))
}
/**
* Connect to WhatsAppWeb
* @param options the connect options
*/
async connect() {
return this
}
async unexpectedDisconnect (error: DisconnectReason) {
const willReconnect =
(this.autoReconnect === ReconnectMode.onAllErrors ||
(this.autoReconnect === ReconnectMode.onConnectionLost && (error !== DisconnectReason.replaced))) &&
(this.autoReconnect === ReconnectMode.onConnectionLost && error !== DisconnectReason.replaced)) &&
error !== DisconnectReason.invalidSession // do not reconnect if credentials have been invalidated
this.closeInternal(error, willReconnect)
willReconnect && !this.cancelReconnect && this.reconnectLoop ()
willReconnect && this.connect ()
}
/**
* base 64 encode the authentication credentials and return them
@@ -213,6 +226,7 @@ export class WAConnection extends EventEmitter {
const response = await this.waitForMessage(tag, json, timeoutMs)
if (expect200 && response.status && Math.floor(+response.status / 100) !== 2) {
// read here: http://getstatuscode.com/599
if (response.status === 599) {
this.unexpectedDisconnect (DisconnectReason.badSession)
const response = await this.query ({json, binaryTags, tag, timeoutMs, expect200, waitForOpen})
@@ -286,13 +300,11 @@ export class WAConnection extends EventEmitter {
/** Close the connection to WhatsApp Web */
close () {
this.closeInternal (DisconnectReason.intentional)
this.cancelReconnect && this.cancelReconnect ()
}
protected closeInternal (reason?: DisconnectReason, isReconnecting: boolean=false) {
this.log (`closed connection, reason ${reason}${isReconnecting ? ', reconnecting in a few seconds...' : ''}`, MessageLogLevel.info)
this.qrTimeout && clearTimeout (this.qrTimeout)
this.phoneCheck && clearTimeout (this.phoneCheck)
this.keepAliveReq && clearInterval(this.keepAliveReq)
this.state = 'close'
@@ -308,6 +320,11 @@ export class WAConnection extends EventEmitter {
this.pendingRequests = []
}
this.removePendingCallbacks ()
// reconnecting if the timeout is active for the reconnect loop
this.emit ('close', { reason, isReconnecting })
}
protected removePendingCallbacks () {
Object.keys(this.callbacks).forEach(key => {
if (!key.includes('function:')) {
this.log (`cancelling message wait: ${key}`, MessageLogLevel.info)
@@ -315,12 +332,6 @@ export class WAConnection extends EventEmitter {
delete this.callbacks[key]
}
})
// reconnecting if the timeout is active for the reconnect loop
this.emit ('close', { reason, isReconnecting: this.cancelReconnect || isReconnecting})
}
protected async reconnectLoop () {
}
generateMessageTag () {
return `${Utils.unixTimestampSeconds(this.referenceDate)}.--${this.msgCount}`

1
src/WAConnection/1.Validation.ts
View File

@@ -58,7 +58,6 @@ export class WAConnection extends Base {
this.log('validated connection successfully', MessageLogLevel.info)
this.sendPostConnectQueries ()
this.lastSeen = new Date() // set last seen to right now
})
// load profile picture
.then (() => this.query({ json: ['query', 'ProfilePicThumb', this.user.jid], waitForOpen: false, expect200: false }))

291
src/WAConnection/3.Connect.ts
View File

@@ -1,83 +1,143 @@
import * as Utils from './Utils'
import { WAMessage, WAChat, WAContact, MessageLogLevel, WANode, KEEP_ALIVE_INTERVAL_MS, BaileysError, WAConnectOptions, DisconnectReason } from './Constants'
import { WAMessage, WAChat, MessageLogLevel, WANode, KEEP_ALIVE_INTERVAL_MS, BaileysError, WAConnectOptions, DisconnectReason, UNAUTHORIZED_CODES, WAContact, TimedOutError } from './Constants'
import {WAConnection as Base} from './1.Validation'
import Decoder from '../Binary/Decoder'
export class WAConnection extends Base {
/**
* Connect to WhatsAppWeb
* @param options the connect options
*/
async connect(options: WAConnectOptions = {}) {
/** Connect to WhatsApp Web */
async connect() {
// if we're already connected, throw an error
if (this.state !== 'close') throw new Error('cannot connect when state=' + this.state)
const options = this.connectOptions
this.state = 'connecting'
this.emit ('connecting')
const { ws, cancel } = Utils.openWebSocketConnection (5000, typeof options?.retryOnNetworkErrors === 'undefined' ? true : options?.retryOnNetworkErrors)
const promise = Utils.promiseTimeout(options?.timeoutMs, (resolve, reject) => {
ws
.then (conn => this.conn = conn)
.then (() => this.conn.on('message', data => this.onMessageRecieved(data as any)))
.then (() => this.log(`connected to WhatsApp Web server, authenticating via ${options.reconnectID ? 'reconnect' : 'takeover'}`, MessageLogLevel.info))
.then (() => this.authenticate(options?.reconnectID))
.then (() => {
this.conn.removeAllListeners ('error')
this.conn.removeAllListeners ('close')
this.conn.on ('close', () => this.unexpectedDisconnect (DisconnectReason.close))
})
.then (resolve)
.catch (reject)
})
.catch (err => {
cancel ()
throw err
}) as Promise<void>
let tries = 0
while (this.state === 'connecting') {
tries += 1
try {
// if the first try failed, delay & connect again
await this.connectInternal (options, tries > 1 && 2000)
this.phoneConnected = true
this.state = 'open'
} catch (error) {
const loggedOut = error instanceof BaileysError && UNAUTHORIZED_CODES.includes(error.status)
const willReconnect = !loggedOut && (tries <= (options?.maxRetries || 5)) && this.state === 'connecting'
this.log (`connect attempt ${tries} failed: ${error}${ willReconnect ? ', retrying...' : ''}`, MessageLogLevel.info)
if ((this.state as string) !== 'close' && !willReconnect) {
this.closeInternal (loggedOut ? DisconnectReason.invalidSession : error.message)
}
if (!willReconnect) throw error
}
}
this.emit ('open')
this.releasePendingRequests ()
this.startKeepAliveRequest()
this.log ('opened connection to WhatsApp Web', MessageLogLevel.info)
this.conn.on ('close', () => this.unexpectedDisconnect (DisconnectReason.close))
return this
}
/** Meat of the connect logic */
protected async connectInternal (options: WAConnectOptions, delayMs?: number) {
// actual connect
const connect = () => {
const tasks: Promise<void>[] = []
const timeoutMs = options?.timeoutMs || 60*1000
const { ws, cancel } = Utils.openWebSocketConnection (5000, false)
let cancelTask: () => void
if (typeof options?.waitForChats === 'undefined' ? true : options?.waitForChats) {
const {waitForChats, cancelChats} = this.receiveChatsAndContacts(timeoutMs, true)
tasks.push (waitForChats)
cancellations.push (cancelChats)
cancelTask = () => { cancelChats(); cancel() }
} else cancelTask = cancel
// determine whether reconnect should be used or not
const shouldUseReconnect = this.lastDisconnectReason !== DisconnectReason.replaced &&
this.lastDisconnectReason !== DisconnectReason.unknown &&
this.lastDisconnectReason !== DisconnectReason.intentional && this.user
const reconnectID = shouldUseReconnect ? this.user.jid.replace ('@s.whatsapp.net', '@c.us') : null
const promise = Utils.promiseTimeout(timeoutMs, (resolve, reject) => {
ws
.then (conn => this.conn = conn)
.then (() => this.conn.on('message', data => this.onMessageRecieved(data as any)))
.then (() => (
this.log(`connected to WhatsApp Web server, authenticating via ${reconnectID ? 'reconnect' : 'takeover'}`, MessageLogLevel.info)
))
.then (() => this.authenticate(reconnectID))
.then (() => (
this.conn
.removeAllListeners ('error')
.removeAllListeners('close')
))
.then (resolve)
.catch (reject)
})
.catch (err => {
this.removePendingCallbacks ()
throw err
}) as Promise<void>
tasks.push (promise)
return {
promise: Promise.all (tasks),
cancel: cancelTask
}
}
let promise = Promise.resolve ()
let cancellations: (() => void)[] = []
const cancel = () => cancellations.forEach (cancel => cancel())
this.on ('close', cancel)
try {
const tasks = [promise]
const waitForChats = typeof options?.waitForChats === 'undefined' ? true : options?.waitForChats
if (waitForChats) tasks.push (this.receiveChatsAndContacts(options?.timeoutMs, true))
await Promise.all (tasks)
this.phoneConnected = true
this.state = 'open'
this.emit ('open')
this.startKeepAliveRequest()
this.registerPhoneConnectionPoll ()
this.releasePendingRequests ()
this.log ('opened connection to WhatsApp Web', MessageLogLevel.info)
return this
} catch (error) {
const loggedOut = error instanceof BaileysError && error.status === 401
if (loggedOut && this.cancelReconnect) this.cancelReconnect ()
if ((this.state as string) !== 'close') {
this.closeInternal (loggedOut ? 'invalid_session' : error.message)
}
throw error
} finally {
this.off ('close', cancel)
if (delayMs) {
const {delay, cancel} = Utils.delayCancellable (delayMs)
promise = delay
cancellations.push (cancel)
}
return promise
.then (() => {
const {promise, cancel} = connect ()
cancellations.push (cancel)
return promise
})
.finally (() => {
cancel()
this.off('close', cancel)
})
}
/**
* Sets up callbacks to receive chats, contacts & messages.
* Must be called immediately after connect
* @returns [chats, contacts]
*/
protected async receiveChatsAndContacts(timeoutMs: number = null, stopAfterMostRecentMessage: boolean=false) {
protected receiveChatsAndContacts(timeoutMs: number = null, stopAfterMostRecentMessage: boolean=false) {
this.contacts = {}
this.chats.clear ()
let receivedContacts = false
let receivedMessages = false
let resolveTask: () => void
@@ -89,7 +149,12 @@ export class WAConnection extends Base {
this.deregisterCallback(['action', 'add:unread'])
}
this.deregisterCallback(['response', 'type:chat'])
}
this.deregisterCallback(['response', 'type:contacts'])
}
const checkForResolution = () => {
if (receivedContacts && receivedMessages) resolveTask ()
}
// wait for messages to load
const chatUpdate = json => {
receivedMessages = true
@@ -104,7 +169,7 @@ export class WAConnection extends Base {
})
}
// if received contacts before messages
if (isLast) resolveTask ()
if (isLast && receivedContacts) checkForResolution ()
}
// wait for actual messages to load, "last" is the most recent message, "before" contains prior messages
@@ -115,9 +180,10 @@ export class WAConnection extends Base {
}
// get chats
this.registerCallback(['response', 'type:chat'], json => {
if (json[1].duplicate) return
if (json[1].duplicate || !json[2]) return
json[2]?.forEach(([item, chat]: [any, WAChat]) => {
json[2]
.forEach(([item, chat]: [any, WAChat]) => {
if (!chat) {
this.log (`unexpectedly got null chat: ${item}, ${chat}`, MessageLogLevel.info)
return
@@ -133,24 +199,46 @@ export class WAConnection extends Base {
this.chats.insert (chat) // chats data (log json to see what it looks like)
})
this.log (`received ${this.chats.all().length} chats`, MessageLogLevel.info)
// if there are no chats
if (this.chats.all().length === 0) {
this.log (`received ${json[2].length} chats`, MessageLogLevel.info)
if (json[2].length === 0) {
receivedMessages = true
resolveTask ()
checkForResolution ()
}
})
// get contacts
this.registerCallback(['response', 'type:contacts'], json => {
if (json[1].duplicate || !json[2]) return
receivedContacts = true
json[2].forEach(([type, contact]: ['user', WAContact]) => {
if (!contact) return this.log (`unexpectedly got null contact: ${type}, ${contact}`, MessageLogLevel.info)
contact.jid = Utils.whatsappID (contact.jid)
this.contacts[contact.jid] = contact
})
this.log (`received ${json[2].length} contacts`, MessageLogLevel.info)
checkForResolution ()
})
// wait for the chats & contacts to load
await Utils.promiseTimeout (timeoutMs, (resolve, reject) => {
resolveTask = resolve
const rejectTask = (reason) => {
reject (new Error(reason))
this.off ('close', rejectTask)
}
this.on ('close', rejectTask)
})
const {delay, cancel} = Utils.delayCancellable (timeoutMs)
const waitForChats = Promise.race ([
new Promise (resolve => resolveTask = resolve),
delay.then (() => { throw TimedOutError() })
])
.then (() => (
this.chats
.all ()
.forEach (chat => {
const respectiveContact = this.contacts[chat.jid]
chat.name = respectiveContact?.name || respectiveContact?.notify || chat.name
})
))
.finally (deregisterCallbacks)
return { waitForChats, cancelChats: cancel }
}
private releasePendingRequests () {
this.pendingRequests.forEach (({resolve}) => resolve()) // send off all pending request
@@ -221,62 +309,27 @@ export class WAConnection extends Base {
}
/** Send a keep alive request every X seconds, server updates & responds with last seen */
private startKeepAliveRequest() {
this.keepAliveReq && clearInterval (this.keepAliveReq)
this.keepAliveReq = setInterval(() => {
const diff = (new Date().getTime() - this.lastSeen.getTime())
if (!this.lastSeen) this.lastSeen = new Date ()
const diff = new Date().getTime() - this.lastSeen.getTime()
/*
check if it's been a suspicious amount of time since the server responded with our last seen
it could be that the network is down
*/
if (diff > KEEP_ALIVE_INTERVAL_MS+5000) this.unexpectedDisconnect (DisconnectReason.lost)
else this.send ('?,,') // if its all good, send a keep alive request
}, KEEP_ALIVE_INTERVAL_MS)
}
protected async reconnectLoop () {
this.cancelledReconnect = false
try {
while (true) {
const {delay, cancel} = Utils.delayCancellable (2500)
this.cancelReconnect = () => {
this.cancelledReconnect = true
this.cancelReconnect = null
cancel ()
}
await delay
else if (this.conn) this.send ('?,,') // if its all good, send a keep alive request
try {
// if an external connect causes the connection to be open
if (this.state === 'open') break
const shouldUseReconnect = this.lastDisconnectReason !== DisconnectReason.replaced && this.lastDisconnectReason !== DisconnectReason.unknown && this.user
const reconnectID = shouldUseReconnect ? this.user.jid.replace ('@s.whatsapp.net', '@c.us') : null
await this.connect ({ timeoutMs: 30000, retryOnNetworkErrors: true, reconnectID })
this.cancelReconnect = null
break
} catch (error) {
// don't continue reconnecting if error is 401
if (error instanceof BaileysError && error.status === 401) {
break
}
this.log (`error in reconnecting: ${error}, reconnecting...`, MessageLogLevel.info)
}
}
} catch {
}
}
protected registerPhoneConnectionPoll () {
this.phoneCheck = setInterval (() => {
this.checkPhoneConnection (5000) // 5000 ms for timeout
// poll phone connection as well,
// 5000 ms for timeout
this.checkPhoneConnection (5000)
.then (connected => {
if (this.phoneConnected != connected) {
this.emit ('connection-phone-change', {connected})
}
this.phoneConnected !== connected && this.emit ('connection-phone-change', {connected})
this.phoneConnected = connected
})
.catch (error => this.log(`error in getting phone connection: ${error}`, MessageLogLevel.info))
}, 15000)
}, KEEP_ALIVE_INTERVAL_MS)
}
/**
* Check if your phone is connected

19
src/WAConnection/4.Events.ts
View File

@@ -31,9 +31,11 @@ export class WAConnection extends Base {
const user = node[1] as WAContact
user.jid = whatsappID(user.jid)
this.contacts[user.jid] = user
const chat = this.chats.get (user.jid)
if (chat) {
chat.name = user.name || user.notify
chat.name = user.name || user.notify || chat.name
this.emit ('chat-update', { jid: chat.jid, name: chat.name })
}
}
@@ -103,19 +105,6 @@ export class WAConnection extends Base {
this.emit ('chat-update', { jid: chat.jid, count: chat.count })
})
// get contacts
this.registerCallback(['response', 'type:contacts'], json => {
if (json[1].duplicate || !json[2]) return
const contacts: {[k: string]: WAContact} = {}
json[2].forEach(([type, contact]: ['user', WAContact]) => {
if (!contact) return this.log (`unexpectedly got null contact: ${type}, ${contact}`, MessageLogLevel.info)
contact.jid = whatsappID (contact.jid)
contacts[contact.jid] = contact
})
this.emit ('contacts-received', contacts)
})
/*// genetic chat action
this.registerCallback (['Chat', 'cmd:action'], json => {
const data = json[1].data as WANode
@@ -306,8 +295,6 @@ export class WAConnection extends Base {
on (event: 'user-presence-update', listener: (update: PresenceUpdate) => void): this
/** when a user's status is updated */
on (event: 'user-status-update', listener: (update: {jid: string, status?: string}) => void): this
/** when a user receives contacts */
on (event: 'contacts-received', listener: (contacts: {[k: string]: WAContact}) => void): this
/** when a new chat is added */
on (event: 'chat-new', listener: (chat: WAChat) => void): this
/** when a chat is updated (archived, deleted, pinned) */

14
src/WAConnection/Constants.ts
View File

@@ -39,6 +39,9 @@ export class BaileysError extends Error {
this.context = context
}
}
export const TimedOutError = () => new BaileysError ('timed out', { status: 408 })
export const CancelledError = () => new BaileysError ('cancelled', { status: 500 })
export interface WAQuery {
json: any[] | WANode
binaryTags?: WATag
@@ -56,17 +59,17 @@ export enum ReconnectMode {
onAllErrors = 2
}
export type WAConnectOptions = {
/** timeout after which the connect will fail, set to null for an infinite timeout */
/** timeout after which the connect attempt will fail, set to null for default timeout value */
timeoutMs?: number
/** maximum attempts to connect */
maxRetries?: number
/** should the chats be waited for */
waitForChats?: boolean
/** retry on network errors while connecting */
retryOnNetworkErrors?: boolean
/** use the 'reconnect' tag to reconnect instead of the 'takeover' tag */
reconnectID?: string
}
export type WAConnectionState = 'open' | 'connecting' | 'close'
export const UNAUTHORIZED_CODES = [401, 419]
/** Types of Disconnect Reasons */
export enum DisconnectReason {
/** The connection was closed intentionally */
@@ -344,7 +347,6 @@ export type BaileysEvent =
'connection-phone-change' |
'user-presence-update' |
'user-status-update' |
'contacts-received' |
'chat-new' |
'chat-update' |
'message-new' |

8
src/WAConnection/Utils.ts
View File

@@ -8,7 +8,7 @@ import {platform, release} from 'os'
import WS from 'ws'
import Decoder from '../Binary/Decoder'
import { MessageType, HKDFInfoKeys, MessageOptions, WAChat, WAMessageContent, BaileysError, WAMessageProto } from './Constants'
import { MessageType, HKDFInfoKeys, MessageOptions, WAChat, WAMessageContent, BaileysError, WAMessageProto, TimedOutError, CancelledError } from './Constants'
const platformMap = {
'aix': 'AIX',
@@ -80,7 +80,7 @@ export const delayCancellable = (ms: number) => {
})
const cancel = () => {
clearTimeout (timeout)
reject (new Error('cancelled'))
reject (CancelledError())
}
return { delay, cancel }
}
@@ -99,7 +99,7 @@ export async function promiseTimeout<T>(ms: number, promise: (resolve: (v?: T)=>
try {
const content = await Promise.race([
p,
delay.then(() => pReject(new BaileysError('timed out', p)))
delay.then(() => pReject(TimedOutError()))
])
cancel ()
return content as T
@@ -139,7 +139,7 @@ export const openWebSocketConnection = (timeoutMs: number, retryOnNetworkError:
await delay (1000)
}
}
throw new Error ('cancelled')
throw CancelledError()
}
const cancel = () => cancelled = true
return { ws: connect(), cancel }