upgrade spot ws

Author: pcriadoperez

src/websocket.js

@@ -3,6 +3,7 @@ import JSONbig from 'json-bigint'
 
 import httpMethods from 'http-client'
 import _openWebSocket from 'open-websocket'
+import { createHmacSignature, createAsymmetricSignature } from 'signature'
 
 const endpoints = {
     base: 'wss://stream.binance.com:9443/ws',
@@ -880,6 +881,107 @@ const getStreamMethods = (opts, variator = '') => {
 
 export const keepStreamAlive = (method, listenKey) => method({ listenKey })
 
+const userWebSocketApi = opts => (cb, transform) => {
+    const isTestnet = opts.testnet || (opts.httpBase && opts.httpBase.includes('testnet'))
+    const wsApiUrl = isTestnet
+        ? opts.wsApiTestnet || 'wss://ws-api.testnet.binance.vision/ws-api/v3'
+        : opts.wsApi || 'wss://ws-api.binance.com:443/ws-api/v3'
+
+    let requestId = 1
+    const errorHandler = userErrorHandler(cb, transform)
+    const w = openWebSocket(wsApiUrl)
+
+    const sendSubscribe = () => {
+        const timestamp = opts.getTime ? opts.getTime() : Date.now()
+        const paramsStr = `apiKey=${opts.apiKey}&timestamp=${timestamp}`
+
+        const doSend = signature => {
+            w.send(
+                JSONbig.stringify({
+                    id: requestId++,
+                    method: 'userDataStream.subscribe.signature',
+                    params: {
+                        apiKey: opts.apiKey,
+                        timestamp,
+                        signature,
+                    },
+                }),
+            )
+        }
+
+        if (opts.apiSecret) {
+            createHmacSignature(paramsStr, opts.apiSecret).then(doSend)
+        } else if (opts.privateKey) {
+            doSend(createAsymmetricSignature(paramsStr, opts.privateKey))
+        }
+    }
+
+    return new Promise((resolve, reject) => {
+        let resolved = false
+
+        w.onopen = () => {
+            sendSubscribe()
+            if (opts.emitSocketOpens) {
+                userOpenHandler(cb, transform)()
+            }
+        }
+
+        w.onmessage = msg => {
+            const data = JSONbig.parse(msg.data)
+
+            // Control response (subscription/unsubscription)
+            if ('id' in data) {
+                if (data.error) {
+                    const err = new Error(data.error.msg || 'WebSocket API error')
+                    err.code = data.error.code
+                    if (!resolved) {
+                        resolved = true
+                        reject(err)
+                    } else if (opts.emitStreamErrors) {
+                        errorHandler(err)
+                    }
+                } else if (!resolved) {
+                    resolved = true
+                    resolve(options => {
+                        try {
+                            w.send(
+                                JSONbig.stringify({
+                                    id: requestId++,
+                                    method: 'userDataStream.unsubscribe',
+                                }),
+                            )
+                        } catch (e) {
+                            // Ignore send errors during close
+                        }
+                        w.close(1000, 'Close handle was called', {
+                            keepClosed: true,
+                            ...options,
+                        })
+                    })
+                }
+                return
+            }
+
+            // User data event - unwrap if in wrapped format
+            let eventData = data
+            if (data.event && typeof data.event === 'object') {
+                eventData = data.event
+            }
+
+            if (eventData.e) {
+                userEventHandler(cb, transform)({ data: JSONbig.stringify(eventData) })
+            }
+        }
+
+        w.onerror = event => {
+            const error = event.error || event.message || new Error('WebSocket error')
+            if (opts.emitSocketErrors) {
+                errorHandler(typeof error === 'string' ? new Error(error) : error)
+            }
+        }
+    })
+}
+
 const user = (opts, variator) => (cb, transform) => {
     const [getDataStream, keepDataStream, closeDataStream] = getStreamMethods(opts, variator)
 
@@ -1029,7 +1131,7 @@ export default opts => {
         miniTicker,
         allMiniTickers,
         customSubStream,
-        user: user(opts),
+        user: userWebSocketApi(opts),
 
         marginUser: user(opts, 'margin'),