State handler `listen` triggered multiple times

[christian-bromann]

Given you update the examples/worker_threads.js to:

import { URL } from 'url';
import { Worker, isMainThread, workerData } from 'worker_threads';
import Channel from '../dist/esm/worker_threads.js';

const filename = new URL('', import.meta.url).pathname;

const ch = new Channel('test', {});

if (isMainThread) {
    console.log(Date.now(), 'START MAIN THREAD');
    const bus = await ch.registerPromise([
        new Worker(filename, { workerData: { id: 'worker #1' } }),
        // new Worker(filename, { workerData: { id: 'worker #2' } }),
        // new Worker(filename, { workerData: { id: 'worker #3' } })
    ]);

    bus.listen('onCustomEvent', (msg) =>
        console.log(Date.now(), 'Received from worker thread:', msg));

    // bus.listen('onExit', (e) => {
    //     console.log(Date.now(), 'Bye bye');
    //     ch.providers.map((p) => p.terminate());
    // });

    setTimeout(() => bus.broadcast({ onCustomWorkerEvent: 'worker #3' }), 100);
} else {
    console.log(Date.now(), 'START WORKER THREAD', workerData.id);
    const client = ch.attach();

    /**
     * listen to events within the same sandbox
     */
    client.listen('onCustomEvent', (msg) =>
        console.log(Date.now(), `Another worker message received in ${workerData.id}; ${msg}`));

    /**
     * broadcast to all
     */
    // await new Promise((r) => setTimeout(r, 200))
    client.broadcast({ onCustomEvent: `Hello from ${workerData.id} 👋` });

    // /**
    //  * listen to messages from message bus
    //  */
    // client.listen('onCustomWorkerEvent', (id) => {
    //     if (workerData.id === id) {
    //         console.log(Date.now(), 'EMIT');
    //         client.broadcast({ 'onExit': workerData.id });
    //     }
    // });
}

You will receive the following output:

1657291971292 START MAIN THREAD
1657291971299 Received from worker thread: undefined
1657291971330 START WORKER THREAD worker #1
1657291971333 Received from worker thread: Hello from worker #1 👋
1657291971333 Another worker message received in worker #1; Hello from worker #1 👋
1657291971400 Received from worker thread: Hello from worker #1 👋
1657291971400 Another worker message received in worker #1; Hello from worker #1 👋

Which is not one I would expect. As you can see:

Received from worker thread: Hello from worker #1 👋
Another worker message received in worker #1; Hello from worker #1 👋

is printed twice even though the event is just emitted once. Commenting out this line:

setTimeout(() => bus.broadcast({ onCustomWorkerEvent: 'worker #3' }), 100);

Gives the expected result. So it seems that events are not filtered properly.

[sourishkrout]

1. While I understand what appears confusing, it's not entirely unintuitive to me. broadcast is meant literally as in "let everyone know an emission happened". No matter whether the value changed. Since tangle was originally conceived as an RPC mechanism (in other words message passing), I did think it would make sense to let the client decide whether or not they are interested in it the emission (in the same vein as a react component using contexts for state mgmt would use useMemo to scope state changes down to what the receiver side, the component, is interested in.

2. broadcast and listen aren't great for event handlers. That's what emit and on/once are for. Not sure if this would solve your issue here but I'm pretty certain that "emit" for a specific prop wouldn't cause emissions on unrelated handlers. A simple solution could be to have a update method that will emit a single prop vs broadcasting the entirety. However, let's be sure we are looking at the big picture.

3. As suggested in the past, I'm totally up to consolidate around use-cases, it does sound like you've got more real world exposure lately leveraging tangle. I do, however, think we need to get on the same page about semantics. I could be wrong, but I'm also sensing that we have to orthogonal design fundamentals at play here, where initial I conceived tangle with an coarse RPC interface whereas you've build a higher-level, more practical, stream/event based interface on top of it. A secondary thing I have run into with the WebWorker implementation is that it actually uses shared memory and won't always run into classic separate sandbox isolation issue, e.g. de-/marshaling from to JSON as wire format. Not an issue here but just as a word of caution.

Let me dig through #39 before we decide on anything.