Websocket json error

@kwonoj I know what’s happening I can walk through it because otherwise you would need a Microsoft botframework scenario setup.

So what’s happening is that the websocket will emit MessageEvents that have data, and when they don’t have data the error gets emitted.

Now, right there to me that seems very odd. Like data doesn’t get emitted so an error is thrown. Now the error is recognized but it is not handled so the stream dies and any subsequent messages don’t make it through the stream.

So what i’ve done, or have tried to do is filter out the errors. MULTIPLE, lol multiple ways, but what’s odd is that filtering doesn’t work when the error is emitted.

Here is what is what I mean.
This is a correct message.

message received:  
MessageEvent {isTrusted: true, data: "{
↵  "activities": [
↵    {
↵      "type": "messag…hT-5|0000007"
↵    }
↵  ],
↵  "watermark": "8"
↵}", origin: "wss://directline.botframework.com", lastEventId: "", source: null, …}
bubbles: false
cancelBubble: false
cancelable: false
composed: false
currentTarget: WebSocket {__zone_symbol__openfalse: Array(1), __zone_symbol__errorfalse: Array(1), __zone_symbol__ON_PROPERTYopen: ƒ, __zone_symbol__ON_PROPERTYerror: ƒ, __zone_symbol__ON_PROPERTYclose: ƒ, …}
data: "{
↵  "activities": [
↵    {
↵      "type": "message",
↵      "id": "IB4EjWeat0ttVlDCylFhT-5|0000008",
↵      "timestamp": "2020-06-22T15:19:51.3151486Z",
↵      "channelId": "directline",
↵      "from": {
↵        "id": "xxxx-xxxx-bot",
↵        "name": "xxxx-xxxx-bot"
↵      },
↵      "conversation": {
↵        "id": "IB4EjWeat0ttVlDCylFhT-5"
↵      },
↵      "text": "Before I begin, do you give me permission to know your location?",
↵      "speak": "Before I begin, do you give me permission to know your location?",
↵      "inputHint": "expectingInput",
↵      "suggestedActions": {
↵        "actions": [
↵          {
↵            "type": "imBack",
↵            "title": "Yes",
↵            "value": "Yes"
↵          },
↵          {
↵            "type": "imBack",
↵            "title": "No",
↵            "value": "No"
↵          }
↵        ]
↵      },
↵      "replyToId": "IB4EjWeat0ttVlDCylFhT-5|0000007"
↵    }
↵  ],
↵  "watermark": "8"
↵}"
defaultPrevented: false

Now, because I have deserialized the event output i can mitigate the error BUT I still can’t filter it. So when it propagates through this is what it looks like now.
This message will be pushed through every minute or so when there is no activity. There is an empty data string. I have no control over this service.

message received:  
MessageEvent {isTrusted: true, data: "", origin: "wss://directline.botframework.com", lastEventId: "", source: null, …}
bubbles: false
cancelBubble: false
cancelable: false
composed: false
currentTarget: WebSocket {__zone_symbol__openfalse: Array(1), __zone_symbol__errorfalse: Array(1), __zone_symbol__ON_PROPERTYopen: ƒ, __zone_symbol__ON_PROPERTYerror: ƒ, __zone_symbol__ON_PROPERTYclose: ƒ, …}
**data: ""**

Now as I subscribe to it and I try to do any filtering at all it will break the same way as before as if I did NOT serialize the output. The commented out code is me trying to filter.

subject
      .subscribe(
        // msg => console.log('message received: ' + JSON.stringify(msg['activities'][0]['text'])), // Called whenever there is a message from the server.
        msg => {
          //   const message = JSON.parse(msg.data);
          //  if (message && Object.keys(message).length >= 0) {
          //     console.log(' message.activities[0].text === ', message)
          //     console.log(' message.activities[1].text === ', message.activities)
          //     if (message.activities && message.activities.length >= 0 && message.activities[0].type === 'message') {
          //       console.log('THI IS E ', message.activities[0]);
          //     }
          //   }
          if (msg) {
            console.log('YOUR MOM landed here', typeof msg);
          }
          console.log('message received: ', msg)
          this.test = msg;
        }, // Called whenever there is a message from the server.
        err => console.log('ERROR from socket ', err), // Called if at any point WebSocket API signals some kind of error.
        () => console.log('complete') // Called when connection is closed (for whatever reason).
      );

Next I tried to use the multiplex feature which isn’t explained very well but I assume it is for filtering and multiple subscription streams… please correct me if I have misunderstanding about that.

This does not work either on an empty message data stream. It will break. I turned off the deserializer here so it would work properly.

Here is that code

 const observableA = subject.multiplex(
        () => ({subscribe: '*'}), // When server gets this message, it will start sending messages for 'A'...
        () => ({unsubscribe: 'A'}), // ...and when gets this one, it will stop.
        message => message['activities'][0].type === 'message' // If the function returns `true` message is passed down the stream. Skipped if the function returns false.
      );
      const subA = observableA.subscribe(messageForA => console.log('messageForA', messageForA));

I am at a loss for this. I’ve tried everything I can think of but it seems very odd that it can’t handle an empty data return on the messageevent stream.

Also, it is very difficult to work with the data inside of the subscription. It seems ultra rigid and final it a sense that if I do something to the data returned in the message stream it will break very easily and the data mutated is very specific to what happened upstream rather than a resultant downstream object.

Уже не единожды на просторах интернетов обсуждались плюсы вебсокетов над xmlhttprequest(ajax) запросами. Из них основные — это скорость доставки(т.к. соединение всегда открыто) и экономия ресурсов (т.к. соединение открыто единожды). Особенно хорошо вебсокеты показывают себя при больших нагрузках, где производительность начинает отличаться не в разы, а на порядок.

НО! Как по мне, есть один большой минус — это потеря состояние вызова. О чем это я? Ajax уже везде обернуто в Promise (будь то браузерный fetch, axios или superagent) и мы имеем удобный интерфейс, пользуясь ajax. Вот пример:

fetch('www.tralala.com/api', {})
.then(console.log)
.catch(console.error)

Так на каждый вызов fetch мы описываем удачную логику в then, а неудачную в catch. И сколько бы вызовов не произошло, мы точно знаем на какой запрос получили ответ. Удобненько? Да! С вебсокетами все сложнее:

const ws = new WebSocket("wss://tralala.com/ws")
ws.onmessage = event => {
  console.log(event.data) // Тут у нас строка
}
// Отправить можем только строку
ws.send(JSON.stringify({ data: "привет" }))

Если вы уже игрались в вебсокеты, то знаете, что при получении пакета вебсокет возвращает нативное событие в котором в поле data будет лежать строка(и только строка), в которой может быть как и успешный ответ, так и ошибка. И вообще на кой нам строка…

А если мы отправим 20 запросов в очень короткий промежуток, а сервер каждый запрос обрабатывает с разной скоростью (и вернет ответы по мере готовности), то получится, что пакеты-ответы мы получим обратно в случайном порядке (так кстати обычно и происходит), как определить, на какой запрос мы получили ответ? Получается, что у нас есть крутая, быстрая сабля, махать которой мы не умеем. Жизнь начинает казаться серой, давайте исправлять!

На просторах Github я не нашел достойной реализации, скорее потому что универсальной её сделать сложно, но я уверен, что мы можем описать кейс под конкретно наш случай и уложиться в пару десятков строк. Готовьте спицы и резину, пишем велосипед!

Итак, давайте определимся с концепцией:

• Во-первых, для любого общения нужен протокол (именно поэтому люди здороваются, прежде чем начать общение, устанавливают язык/протокол/формат общения). Давайте его придумаем! Условимся, что клиент всегда будет называть метод который хочет вызвать, а сервер выдавать результат вызова этого метода, либо ошибку (где будут поля message и code ошибки), если такового не существует.
• Мы хотим какую-то простую функцию, которую будем вызывать, например так: WS(“название_метода”, “параметры”).then().catch()
• Чтобы на конкретный ответ мы знали, какой запрос ему соответствовал, нам нужен маркер (uniq_id), это будет любой уникальный айди. Я составлю его из случайно сгенерированной строки и текущего таймпстемпа. И когда сервер пришлет помеченный нашим id ответ, который мы можем ассоциировать с тем, что лежит на фронтенде, но нам нужен какой-то message-broker для правильного ассоциирования айди.
• Чтобы хранить и управлять id, можно было обойтись массивом, либо списком и парой функциями поверх, но мы возьмем вот эту полезную штуку. Классический паттерн Event Emitter, который дает возможность подписываться/отписываться на события и вызывать callback по событию. Названием события будет уникальный id.
• В качестве сервера возьмем node.js ws и express.js . Потому что быстро и просто. Но вы можете попробовать другой бекенд.

Начнем с сервера:

Во-первых, создадим json заглушки, которые просто будут возвращать какие-то данные, без разницы. Это нужно будет для тестов. Вы их найдете в ./app/stub/

Далее основа сервера, index.js

const express = require('express')
const app = express()
const WebSocket = require('ws')
const { privileges, users } = require('./client/stub/index')
const {
  utils: {
    throwError,
    handleInvalidJSON,
    handleInvalidData,
    handleUnknownMethod,
    getRandomInt
  }
} = require('./app')

const appPort = 3000
const socketPort = 3001

// Сверим наш css и js
app.use(express.static(`${__dirname}/public`))

// отдаем index.html по рутовому роуту
app.get('/', (req, res) => {
  res.sendFile(`${__dirname}/public/index.html`)
})

// Запускаем экспресс приложение по порту appPort
app.listen(appPort, initWebSockets)

function initWebSockets() {
  // Запускаем вебсокет сервер
  const WS = new WebSocket.Server({ port: socketPort })

  WS.on('connection', socket => {
    // После соединения вебсокетов, подписываем на событие message
    socket.on('message', data => {
      // Пытаемся распарсить JSON строку
      try {
        data = JSON.parse(data)
      } catch (e) {
        return handleInvalidJSON(socket)
      }

      // Распаковываем распарсенные данные из JSON
      const { method, id = null, params = {} } = data
      // Вам скорее всего пригодится, сюда можно передать параметры для описание логики вашего кастомного приложения, но я просто залогирую это
      console.log('request params: ', params)

      try {
        // Убедимся что фронтенд не забыл передать id(маркер)
        if (!id) throwError('id is required', 400)

        // Это не нужно для вашей реализаци, тут я просто искусственно создаю разное время ответа сервера
        const timeout = getRandomInt(500, 1250)

        // Простым образом создается задержка ответа
        setTimeout(() => {
          // Роутинг по методам
          switch(method) {
            case 'getUsers':
              return socket.send(JSON.stringify({ id, data: users }))
            case 'getPrivileges':
              return socket.send(JSON.stringify({ id, data: privileges }))
            default:
              handleUnknownMethod(method, id, socket)
          }
        }, timeout)
      } catch (err) {
        // Обработаем все возникшие ошибки и покажем это фронтенду
        handleInvalidData(err, id, socket)
      }
    })
  })
}

Методы-хелперы вынесены в отдельный файл app/utils/index.js

/**
 * Более удобный генератор ошибок
 * @param {String} message
 * @param {String|Number} code
 */
exports.throwError = (message = 'Internal error', code = 500) => {
  const err = new Error(message)
  err.code = code
  throw err
}

/**
 * Handler of the invalid JSON
 * @param {Object} socket
 */
exports.handleInvalidJSON = socket => {
  const errData = JSON.stringify({
    id: null,
    error: {
      message: 'Invalid JSON',
      code: 400
    }
  })
  socket.send(errData)
}

/**
 * Обработчик для невалидных данных
 * @param {Object} err
 * @param {String|Number} id
 * @param {object} socket
 */
exports.handleInvalidData = (err, id, socket) => {
  const { message, code } = err
  const errData = JSON.stringify({
    id,
    error: { message, code }
  })
  socket.send(errData)
}

/**
 * Обработчик для метода, которого не существует
 * @param {String} method
 * @param {String|Number} id
 * @param {Object} socket
 */
exports.handleUnknownMethod = (method, id, socket) => {
  const err = new Error(`Unknown method: ${method}`)
  err.code = 400
  this.handleInvalidData(err, id, socket)
}

/**
 * Возвращаем случайное число в промежутке
 * @param {Number} min
 * @param {Number} max
 * @returns {Number}
 */
exports.getRandomInt = (min, max) => {
  return Math.floor(Math.random() * (max - min) + min)
}

Теперь клиентская часть!

CSS и index.html не особо интересно, там 4 кнопки которые вызывают соответствующие функции и записывают результат в соседний блок. Все находится в папке public

// Как вы заметили я использую функцию require из node.js
// Для подгрузки event-emitter на фронтенде,
// чтобы легко этого добиться, вы можете взять
// watchify - это модуль из browserify или использовать
любой другой сборщик, например webpack
// https://github.com/browserify/browserify
// https://github.com/browserify/watchify

const emitter = require('event-emitter')()
const hasListeners = require('event-emitter/has-listeners')

// Функция возвращает уникальную строку состоящу из timestamp#random_str
const getUniqId = () => Date.now().toString() + '#' + Math.random().toString(36).substr(2, 9)

// Создаем вебсокет соединение с бекендом
const ws = new WebSocket("ws://localhost:3001")

// Выберем заранее все кнопочки и DOM элементы
const getUserBtn = document.querySelector('#get-users')
const getUserResult = document.querySelector('#get-users-response')
const getPrivilegesBtn = document.querySelector('#get-privileges')
const getPrivilegesResult = document.querySelector('#get-privileges-response')
const getUsersAndPrivilegesBtn = document.querySelector('#get-all')
const getUsersAndPrivilegesResult = document.querySelector('#get-all-response')
const errorBtn = document.querySelector('#error')
const errorResult = document.querySelector('#error-response')

// Установим максимальное время ожидание ответа сервера в секундах
const serverExceedTimeout = 2

// Напишем обработчик на нативное сообщение вебсокета
ws.onmessage = e => {
  try {
    // Пытаемся распарсить JSON данные от сервера
    const { id, data, error } = JSON.parse(e.data)
    // Генерируем событие по присланному с серверу id
    emitter.emit(id, { data, error })
  } catch (e) {
    console.warn('onmessage error: ', e)
  }
}

// Тут основная идея нашей статьи
const WS = (method, params) => new Promise((resolve, reject) => {
  // получаем уникальный айди
  const id = getUniqId()
  // Посылаем на сервер id, метод и параметры
  // параметры никак не используются, просто я показал как это сделать.
  // На их основе вы будете описывать условия и логику.
  // По названию метода мы будем попадать в нужный case
  // и отдавать запрашиваемые даные, сейчас так это проще
  // но в будущем я бы сделал динамический вызов функций, но статья не об этом.
  // Держите в голове мысль о том что по названию метода мы роутим свою логику.
  ws.send(JSON.stringify({ id, method, params }))
  // С помощью event emiiter мы единожны подписываемся на событие
  // https://github.com/medikoo/event-emitter#usage
  // Проверяем что вернул сервер, если данные то вызываем resolve,
  // иначе вызываем reject с ошибкой
  emitter.once(id, ({ data, error}) => {
    data
      ? resolve(data)
      : reject(error || new Error('Unexpected server response.'))
  })

  // Ждем определенный в константе serverExceedTimeout промежуток времени
  // Если сервер так и не ответил, значит сеть оборвалась или
  // что-то пошло не по плану поэтому генерируем ошибку
  // и отписываем наш event emiiter от сообщения чтобы не разбухла память
  setTimeout(() => {
    if (hasListeners(emitter, id)) {
      reject(new Error(`Timeout exceed, ${serverExceedTimeout} sec`))
      emitter.off(id, resolve)
    }
  }, serverExceedTimeout * 1000)
})

// обработчик кнопки getUserBtn handler, вызывает метод
// на сервер "getUsers" и передаем тоде тестовые параметры
getUserBtn.onclick = () => {
  // Перед выполнением функции очистим контейнер результата, так наглядней
  getUserResult.innerHTML = ''

  WS('getUsers', { someData: [1, 2, 3] })
    .then(data => {
      console.log(data)
      getUserResult.innerHTML = JSON.stringify(data)
    })
    .catch(console.error)
}

// обработчик кнопки getPrivilegesBtn handler, вызывает метод
// на сервер "getPrivileges" и передаем тоде тестовые параметры
getPrivilegesBtn.onclick = () => {
  // Перед выполнением функции очистим контейнер результата, так наглядней
  getPrivilegesResult.innerHTML = ''

  WS('getPrivileges', { anotherData: [3, 2, 1] })
    .then(data => {
      console.log('getUsers received: ', data)
      getPrivilegesResult.innerHTML = JSON.stringify(data)
    })
    .catch(console.error)
}

// Давайте убедимя что промисы работаю правильно 
// и мы можем использовать Promise.all
// Вызовим Promise.all([]) с методами "getPrivileges" и "getUserBtn"
getUsersAndPrivilegesBtn.onclick = () => {
  // Перед выполнением функции очистим контейнер результата, так наглядней
  getUsersAndPrivilegesResult.innerHTML = ''

  Promise.all([
    WS('getUsers', { someData: [1, 2, 3] }),
    WS('getPrivileges', { someData: [5, 4, 1] }),
  ])
    .then(([users, privileges]) => {
      console.log('users: ', users)
      console.log('privileges: ', privileges)
      getUsersAndPrivilegesResult.innerHTML = JSON.stringify({users, privileges})
    })
    .catch(console.error)
}

// А теперь попробуем вызвать метод которого на сервере нету
// ожидается что сервер не найдет нужного кейса и вернет ошибку
// а мы её обработает в блоке catch
errorBtn.onclick = () => {
  // Перед выполнением функции очистим контейнер результата, так наглядней
  errorResult.innerHTML = ''

  WS('omgWrongMethod')
    .catch(err => {
      console.error(err)
      errorResult.innerHTML = JSON.stringify(err)
    })
}

Вот и вся реализация. Теперь вызов выглядит так, как мы и хотели:

WS('getUsers', { /* передаем какие-то данные на сервер, опционально */  })
  .then(data => {
    // обрабатываем успешный ответ от вебсокета
  })
  .catch(data => {
    // обрабатываем ошибку
  })

Эффект достигнут!

Весь исходный код можете увидеть тут.

Возможно кто-то заметил некое идейное сходство нашего примитивного “протокола” с JSON-RPC, в следующей статье я расскажу, как подружить вебсокеты с JSON-RPC и решить последний недостаток, в данной реализации нету “гарантии” доставки пакета.