most inline threading

most-thread.ts

import { of, Stream, Scheduler, Sink, Source } from 'most'
import { curry2, CurriedFunction2 } from '@most/prelude'

class ThreadSource<T, R> implements Source<R> {
  constructor (private fn: (a1: T) => void, private arg: T) {

  }

  run (sink: Sink<R>, scheduler: Scheduler) {
    const fnContextStr = `(${this.fn.toString()})(${JSON.stringify(this.arg)})`

    // Transform the passed function into an IIFE and then create a blob.
    const blob = new Blob([fnContextStr], { type: 'application/javascript' })
    // Create a URL from the aforementioned blob that handles the worker logic.
    const worker = new Worker(URL.createObjectURL(blob))

    worker.onmessage = (e) =>
      sink.event(scheduler.now(), e.data)

    worker.onerror = (e: ErrorEvent) =>
      sink.error(scheduler.now(), new Error(e.message))

    // Imperative initialization of worker
    worker.postMessage(null)

    return { dispose: () => worker.terminate() }
  }
}

export const thread = <T, R>(fn: (a1: T) => void, arg: T) => {
  if (typeof fn !== 'function') {
    // Ensure the passed parameter is actually a function.
    throw new Error('Parameter must be a function, received: ${JSON.stringify(a1)}')
  }

  return new Stream(new ThreadSource<T, R>(fn, arg))
}

export const threadCurry = curry2(thread)

usage:

import {threadCurry} from '../most-thread'


const workFn = threadCurry(x => {
  self.postMessage('hello: ' + x)

  setTimeout(() => self.postMessage('hello: ' + x), 1000)
})

workFn('world').observe(x => {
  console.log(x)
})