Handle closed user streams properly (#499)

Author: barhun

src/websocket.js

@@ -609,6 +609,10 @@ export const userEventHandler = (cb, transform = true, variator) => msg => {
   )
 }
 
+const userErrorHandler = (cb, transform = true) => error => {
+  cb({[transform ? 'eventType' : 'type']: 'error', error})
+}
+
 const STREAM_METHODS = ['get', 'keep', 'close']
 
 const capitalize = (str, check) => (check ? `${str[0].toUpperCase()}${str.slice(1)}` : str)
@@ -630,22 +634,28 @@ const user = (opts, variator) => (cb, transform) => {
   let currentListenKey = null
   let int = null
   let w = null
+  let keepClosed = false
+  let errorHandler = userErrorHandler(cb, transform)
 
   const keepAlive = isReconnecting => {
     if (currentListenKey) {
-      keepStreamAlive(keepDataStream, currentListenKey).catch(() => {
+      keepStreamAlive(keepDataStream, currentListenKey).catch(err => {
         closeStream({}, true)
 
         if (isReconnecting) {
           setTimeout(() => makeStream(true), 30e3)
         } else {
           makeStream(true)
         }
+
+        opts.emitStreamErrors && errorHandler(err)
       })
     }
   }
 
-  const closeStream = (options, catchErrors) => {
+  const closeStream = (options, catchErrors = false, setKeepClosed = false) => {
+    keepClosed = setKeepClosed
+
     if (currentListenKey) {
       clearInterval(int)
 
@@ -661,24 +671,30 @@ const user = (opts, variator) => (cb, transform) => {
   }
 
   const makeStream = isReconnecting => {
-    return getDataStream()
+    return !keepClosed && getDataStream()
       .then(({ listenKey }) => {
+        if (keepClosed) {
+          return closeDataStream({ listenKey }).catch(f => f)
+        }
+
         w = openWebSocket(
           `${variator === 'futures' ? endpoints.futures : endpoints.base}/${listenKey}`,
         )
         w.onmessage = msg => userEventHandler(cb, transform, variator)(msg)
+        opts.emitSocketErrors && (w.onerror = ({error}) => errorHandler(error))
 
         currentListenKey = listenKey
 
         int = setInterval(() => keepAlive(false), 50e3)
 
         keepAlive(true)
 
-        return options => closeStream(options)
+        return options => closeStream(options, false, true)
       })
       .catch(err => {
         if (isReconnecting) {
-          setTimeout(() => makeStream(true), 30e3)
+          !keepClosed && setTimeout(() => makeStream(true), 30e3)
+          opts.emitStreamErrors && errorHandler(err)
         } else {
           throw err
         }