Учим webworkers хорошим манерам

Рано или поздно у каждого кто работал с webworkers возникает ситуация когда код превращается в кашу вроде этого:

const worker = new Worker('test.worker.js');
worker.onmessage = (data) => {
  if(data.eventName === 'someFuncResult')
    someFunc();
  else if(data.eventName === 'someFunc2Result')
    someFunc2();
};

worker.postMessage({eventName: 'someFunc'});

self.addEventListener('message', (data) => {
  if(data.eventName === 'someFunc') {
    doSomeFunc();
    self.postMessage('someFuncResult');
  }
  
 if(data.eventName === 'someFunc2') {
    doSomeFunc();
    self.postMessage('someFunc2Result');
  }
})

Если даже закрыть глаза на нечитаемый код, вы можете обнаружить, что не можете одновременно несколько раз запустить одну функцию, так, чтобы вызовы не конфликтовали.
После долгих мучений и страшного кода, было решено реализовать удобную обертку над воркерами.

Цели:
— Читаемость кода
— Конкурентные запросы
— Ассинхронные функции
— Прозрачная обработка ошибок
— Возможность отправки промежуточных результатов выполнения процедуры

Установка

Для установки можно воспользоваться npm

npm install webworker-promise

Далее вы можете импортировать из неё

const WebWorker = require('webworker-promise'); //main-process
const WebWorkerRegister = require('webworker-promise/lib/register');

Либо скачать umd-версию
В этом случае, при отсутствии commonjs и requirejs будут доступны глобальные объекты
— WebWorkerPromise (webworker-promise в cjs, requirejs)
— WebWorkerPromiseRegister — функция будет доступен в воркере при импортировании через importScripts('dist/register.js');

Основной функционал

Вся работа с библиотекой происходит с помощью промисов, соответственно вы можете использовать async/await при выполнении функций

Для инициализации «сервера» в worker-файле необходимо подгрузить функцию webworker-promise/lib/register и выполнить её, после чего все сообщения отправляемые в воркер будут обрабатываться webworker-сервером.

//ping.worker.js
const register = require('webworker-promise/lib/register');

register()
  .operation('ping', (message) => {
    //message - hello,
    return 'pong'; 
  });

И на клиенте

//main.js
const WebWorker = require('webworker-promise');
const worker = new WebWorker(new Worker('ping.worker.js'));

worker.exec('ping', 'hello')
  .then(result => {/* result - pong*/});

Конкурентные запросы

Каждому вызову процедуры назначается уникальный ид к которому привязывается и его результат, что позволяет вызвать несколько раз одну и ту же функцию в один момент времени и получить соответствующий результат вне зависимости от порядка выполнения.

register()
  .operation('hello', ({name, delay}) => {
    return new Promise((res) => {
      setTimeout(() => res(`Hello ${name}!`), delay);
    });
  });

worker.exec('hello', {name: 'Bob', delay: 300})
  .then(message => {/* message - Hello Bob!*/});

worker.exec('hello', {name: 'Alice', delay: 200})
  .then(message => {/* message - Hello Alice!*/});

Асинхронные процедуры

Асинхронные процедуры определяются также как и обычные с одним лишь отличием — обработчик должен возвращать promise-объект в качестве результата

register()
  .operation('get-random-text', (limit) => {
    // fetch как известно возвращает промисы
    return fetch(`https://baconipsum.com/api/?callback=?type=meat-and-filler&paras=${limit}`)
      .then(result => result.json())
  });

const worker = new WebWorker(new Worker('async.worker.js'));

worker.exec('get-random-text', 2)
  .then(texts => {/*texts - массив из двух случайных строк*/});

Обработка ошибок

Если во время выполнения произошла ошибка, она будет выкинута в промис и может быть отловлена как и принято в catch. Т.к. воркеры не умеют обмениваться оригинальным объектом ошибки при отлавливании ошибки вытаскивается stack и message, а на клиенте просто кидается объект ошибки throw {stack: 'error trace', message: 'error message'}. В следующих версиях обработка ошибок будет улучшена

worker.exec('get-random-text', 2)
  .catch(e => {/* e.message, e.stack */});

Отправка промежуточных результатов

Часто возникает ситуация, что нужно отправлять промежуточные результаты выполнения процедуры. Например, если вы отправляете файл с помощью воркера и нужно переодически отправлять прогресс загрузки. Именно для этого были введены события. Как уже говорил выше, каждому вызову процедуры присваивается свой уникальный идентификатор, так вот, к этому идентификатору также привязываются события.

register()
  .operation('get-events', (str, emit) => {
    return new Promise((res) => {
      // отправляем сначала 20, 90 прогресс и только потом резолвим промис
      setTimeout(() => emit('progress', 20), 100);
      setTimeout(() => emit('progress', 90), 200);
      setTimeout(() => res('hello'), 300);
    });
  });

worker.exec('get-events', '', [], (eventName, progress) => {
  if(eventName === 'progress')
    console.log(progress); // progress, 20, 90
})
.then(result => {/*result - hello*/});

Trasferable objects

Для обмена данными между основным процессом и воркером без копирования памяти в js были введены transferable objects. Если коротко, то при передаче объекта arrayBuffer вместо того, чтобы копировать данные из памяти основного потока в поток воркера, вы можете их просто делегировать, более подробно можно прочитать здесь

Чтобы передать transferable объекты необходимо их указать третьим аргументом в виде массива, а вот чтобы вернуть transferable необходимо в результате функции вернуть объект специального класса TransferableResponse.

register()
  .operation('send-buffer', (obj) => {
    //можно произвести некоторые операции с obj.myBuffer;
    return new register.TransferableResponse({myBuffer}, [myBuffer]);
  });

worker.exec('send-buffer', {myBuff: myBuffer}, [myBuffer])
  .then((obj) => {/* obj.myBuffer */});

«Однопроцедурные» воркеры

Если ваш воркер выполняет лишь одну функцию, то нет необходимости объявлять отдельную процедуру, вы можете просто передать в register обработчик

register( (name) => `Hello ${name}!`);

worker.postMessage('hello');

Весь код открыт и покрыт тестами.
Исходники на github

Автор: kwolfy

Источник