threads.js

threads.js copied to clipboard

Is there any way to use this package with react-native?

The path to the worker file isn't being resolved correctly when using the provided example from the README:

TypeError: undefined is not a constructor (evaluating 'new _threads.Worker("./Worker.js")')

Interesting point. Haven't tried anything alike, nor am I aware of anyone else using threads.js with react-native.

I think you might be able to use threads.js with react-native-threads like this (untested!) as they seem to implement the web worker API:

import { Thread } from "react-native-threads"
import { spawn } from "threads"

const worker = await spawn(new Thread("path/to/thread.js"))

Thanks for the suggestion @andywer! I tried it out and it seems like react-native-threads isn't compatible with threads.js:

// master.js
const worker = await spawn(new Thread('worker.js'));
const result = await worker.myFunction('This should be received by the worker');
console.log('Worker response:', result); 

// worker.js
import { expose } from 'threads/worker';
expose({
  myFunction(message) {
    return `Received message ${message}`;
  }
});

This returns the following error:

expose() called in the master thread.

Integrating support for react-native into threads.js could be a better approach but it's understandable if that's outside of the scope of this project.

We could look into supporting react-native if it's just a few simple changes.

Can you try to disable the line that does the expose() called in the master thread. check? That's just a safeguard that falsy complains in that case.

Thanks for the idea @andywer! I gave that a shot but unfortunately it leads us to another issue, this line:

const worker = await spawn(new Thread('worker.js'));

generates the following error:

TypeError: worker.addEventListener is not a function. (In 'worker.addEventListener("message", messageHandler)', 'worker.addEventListener' is undefined)

Can you try to polyfill the *EventListener methods on the worker, similar to this? https://github.com/andywer/threads.js/blob/master/src/master/implementation.node.ts#L199-L210

As you can see we had the same situation with the tiny-worker in the beginning 😉

This library kicks ass and lack of React Native support is really its only shortcoming. I hope we can get this working.

@andywer can you clarify your polyfill idea? I might get a chance to give this a shot in a couple weeks.

Thanks, @Slapbox!

The polyfill… The web worker spec only requires the Worker to support an onmessage and an onerror property, both callback functions. Those are plenty inconvenient to use once you want to assign more than one handler each, so we polyfill the addEventListener and removeEventListener methods on the Worker.

That happens whenever a worker is instantiated. All code running later just blindly assumes that addEventListener & removeEventListener are present and fails if they aren't.

Current implementation for tiny-worker: https://github.com/andywer/threads.js/blob/d476cf18c7c5d5bb7fa014aea0a548e4e7af96a3/src/master/implementation.node.ts#L199-L210

A potential alternative could be to not polyfill, but instead make the code in threads.js that needs to set/unset those handlers use some dedicated EventTargetUtils.addEventListener(worker, listener) and EventTargetUtils.removeEventListener(worker, listener) functions that are not patched onto the workers.

Thanks once again @andywer for all this information!

I polyfilled the missing EventListener methods in Thread() using the code snippets that you've provided but unfortunately it doesn't work as intended:

Error: Timeout: Did not receive an init message from worker after 10000ms. Make sure the worker calls expose().

It appears as if react-native-threads doesn't actually implement the Web Worker API but rather provide its own implementation which has various limitations. For instance, you can only post strings:

com.facebook.react.bridge.ReadableNativeMap cannot be cast to java.lang.String

Native support for this in threads.js would be the ideal solution for the long term. The approach which you mentioned sounds plausible 👍

Thanks for sharing the update! :)

If you want, you can take it one step further and wrap the worker.postMessage() method, use a package like flatted, serialize all data passed to worker.postMessage() before passing it to the native method and unserialize in the worker's implementation.ts before calling the exposed function.

It quite sounds like the serialization issue is the last blocker. We've then covered all functionality of the worker that we use, I think.

PS: I would be happy to integrate these fixes as a dedicated react-native backend and make it part of the core package if the serialization fix does the trick.

Sorry for any misunderstanding @andywer, I'll clarify. I wasn't able to establish a working communication link between threads.js and react-native-threads. The serialization issue was offtopic since I only wanted to point out that it could become a possible issue.

Nice to see this here! It would be much needed for us! Right now we have a uggggly hack for threads in react native that I am ashamed of and isn't maintainable.

Hey @crubier. Sounds dark, but if it's working it might still be worth sharing 🙂

It's so ugly and intricated in our codebase that I can't even share it because I'd have to share large parts of our private monorepo for it to make sense.

I can try to show a simplified version of the most relevant parts though:

// compute-thread-react-native / thread.ts
import { self } from "react-native-threads";
import { has, getOr } from "lodash/fp";

window.WebSocket = WebSocket; // Don't ask questions

function isPromise(obj) {
  return (
    !!obj &&
    (typeof obj === "object" || typeof obj === "function") &&
    typeof obj.then === "function" &&
    typeof obj.subscribe !== "function"
  );
}

const postJsonMessage = (jsonMessage: {}) =>
  self.postMessage(JSON.stringify(jsonMessage));

const handleMessage = async (jsonMessage: Message) => {
  try {
    ///////////////////////////////////////////////////////////////////////////////
    // The actual business logic executed in the thread is here ! 
    const functionMap = {
      example: i => `All good ${i}`
      // Actually there are more here
     };
    ///////////////////////////////////////////////////////////////////////////////

    // Get which function of the functionMap to call based on the jsonMessage
    const func = getOr(
      handleDefault,
      getOr("default", "functionName", jsonMessage),
      functionMap
    );
   
    const resultPromise = func(...getOr([], "args", jsonMessage));

    let result;
    if (isPromise(resultPromise)) {
      result = await resultPromise;
    } else {
      result = resultPromise;
    }

    return {
      ...jsonMessage,
      result
    };
  } catch (error) {
    return {
      ...jsonMessage,
      error: getOr(error, "message", error)
    };
  }
};


type Message = {
  callReference: String;
  functionName: String;
  args: {}[];
  error: Error;
  result: {};
};

// listen for messages
self.onmessage = async (message: string) => {
  try {
    const jsonInputMessage = JSON.parse(message);
    if (!has("functionName", jsonInputMessage)) {
      return postJsonMessage({ error: "Message has no function name" });
    }
    postJsonMessage(await handleMessage(jsonInputMessage));
    return;
  } catch (error) {
    postJsonMessage({ error });
    return;
  }
};

self.postMessage(
  "Success (Do NOT edit this message, it is intended as a way to ensure sync between main thread and compute thread)"
);

And

// compute-thread-react-native  /  create-thread.js
import { Thread } from "react-native-threads";

export function createThread({
  threadFile = "index.thread.js"
}) {
  let thread;

  // The promise map stores all the current function calls currently being processed by the thread.
  // It is indexed by callReference
  const promiseMap: {
    [key: string]: { resolve: (a: {}) => void; reject: (a: {}) => void };
  } = {};

  const handleMessageFromThread = messageString => {
    try {
      const messageJson = JSON.parse(messageString);
      const { callReference, error, args, functionName, result } = messageJson;
      if (error === null || error === undefined) {
        promiseMap[callReference].resolve(result);
      } else {
        promiseMap[callReference].reject(new Error(error));
      }
      // Remove promise from map
      delete promiseMap[callReference];
    } catch (error) {
      console.log(`MAIN THREAD: Error ${error}`);
    }
  };

  const initializeThread = async ({ timeoutDuration = 120000 } = {}) => {
    thread = new Thread(threadFile);

    const timeoutPromise = new Promise((resolve, reject) => {
      setTimeout(
        () => {
          return reject(
            "Timeout when initializing thread and waiting for its message"
          );
        },
        timeoutDuration,
        "timeout"
      );
    });

    const threadInitializationPromise = new Promise((resolve, reject) => {
      thread.onmessage = message => {
        if (
          message ===
          "Success (Do NOT edit this message, it is intended as a way to ensure sync between main thread and compute thread)"
        ) {
          return resolve("Thread initialized successfully");
        } else {
          return reject("Thread gave the wrong welcome message");
        }
      };
    });

    await Promise.race([timeoutPromise, threadInitializationPromise]);

    thread.onmessage = handleMessageFromThread;
  };
  const terminateThread = async () => {
    thread.terminate();
    thread = null;
  };

  const callFunctionInThread = async (functionName, ...args) => {
    const callReference = Math.random().toString();
    // Create a promise for this function call, and add its resolve and rejects method to the promiseMap
    const promise = new Promise((resolve, reject) => {
      promiseMap[callReference] = { resolve, reject };
    });

    thread.postMessage(
      JSON.stringify({
        callReference,
        args,
        functionName
      })
    );

    const result = await promise;

    return result;
  };

  const callFunction = async (functionName, ...args) => {
    initializeThread();
    const result = await callFunctionInThread(functionName, ...args);
    terminateThread();
    return result;
  };

  const functions = {
    initializeThread,
    terminateThread,
    callFunctionInThread,
    callFunction
  };

  return functions;
}

Then in the package.json of the react native app:

{
  "scripts":{
    "build:bundle-threads": "npm run build:bundle-threads-ios && npm run build:bundle-threads-android",
    "build:bundle-threads-android": "mkdir -p ./android/app/src/main/assets/threads/external/compute-thread-react-native && NODE_OPTIONS=--max_old_space_size=8192 node node_modules/react-native/local-cli/cli.js bundle --dev false --assets-dest ./android/app/src/main/res/ --entry-file external/compute-thread-react-native-react-native/thread.esm.js --platform android --bundle-output ./android/app/src/main/assets/threads/external/compute-thread-react-native-react-native/thread.esm.jsbundle",
    "build:bundle-threads-ios": "mkdir -p ./ios/external/compute-thread-react-native-react-native &&  NODE_OPTIONS=--max_old_space_size=8192 node node_modules/react-native/local-cli/cli.js bundle --dev false --assets-dest ./ios --entry-file external/compute-thread-react-native-react-native/thread.esm.js --platform ios --bundle-output ./ios/external/compute-thread-react-native-react-native/thread.esm.jsbundle",
  }
}

And then somewhere in the app:

import { createComputeThread } from "../../../../external/compute-thread-react-native/index.esm.js";


        ///////////////////////////////////////////////////////////////////////////////
        // In a function somewhere, where we want to use the thread...
        const computeThread = createComputeThread({
          threadFile: "external/compute-thread-react-native/thread.esm.js"
        });
        const result = await computeThread.callFunction(
          "example",
          12
        );
        console.log(result)
        // Logs `All good 12` FINALLY
        ///////////////////////////////////////////////////////////////////////////////

😅