Emissions

A scope often produces values from deep inside it: an item from a list, a row from a table, a field of a form. Emissions carry those out to where you collect them. They thread through the graph as an ordinary reactive value. Emit adds to a named stream, Gather reads one back, If merges a body's emissions into its parent, and Each walks a gathered list.

An emissions value is a plain object with one field: an emissionsByName map. Each name in the map holds a stream: an array of entries in emission order, and an override, its own pending flag. The value carries no global flag of its own. The global pending lives on the emissions node's .pending, so it propagates for free like any other node's pending. That split is the whole trick to loading: one stream can be still loading while the rest are ready, so a driver can act on what has arrived and treat pending as "more may come".

{
  emissionsByName: {
    dom: { array: [ {id, value}, ... ], override: undefined },   // a ready stream
    img: { array: [],                   override: true      },   // this stream is loading
  },
}

An emissions value: a stream per name, with the global pending on the node

An override is three-state. undefined means "defer to the node's global pending". true means "this stream is loading" (a value was emitted while still pending, and withheld). false means "this stream was gathered or reset, explicitly not-pending". An explicit override, true or false, wins over the node's global; only undefined defers to it. So the effective pending of a name is its own override if it has one, else the emissions node's own pending flag:

Reactive._emitStreamPending = (emissionsRef, name) => {
  const stream = emissionsRef.value.emissionsByName[name];
  if (stream !== undefined && stream.override !== undefined) {
    return stream.override;
  }
  return emissionsRef.pending === true;
};

A name's effective pending: its override, or the node's own pending flag

The Emission Thread on a Scope

Each scope carries its latest emissions as a reactive reference. As a result, _scopeCreate now seeds one to the shared empty value. Every Emit, If and Gather advances it:

Reactive._emitEmpty = { emissionsByName: {} };   // one shared empty value, so empty emissions compare equal

The shared empty emissions value

_scopeCreate gains one line: the scope now carries a currentEmissionsRef, seeded to the empty value before the body runs. A scope is a plain object, and the ambient scope is that object, so nothing maps ids to it:

Reactive._scopeCreate = (parentNode, callback) => {
  const scope = { nodes: [], parentNode, currentEmissionsRef: undefined, contexts: {}, destroyed: false };
  const previousScope = Reactive._currentScope;
  Reactive._currentScope = scope;
  try {
    scope.currentEmissionsRef = Reactive.Constant(Reactive._emitEmpty);
    callback();
  } finally {
    Reactive._currentScope = previousScope;
  }
  return scope;
};

_scopeCreate seeds each scope's emissions thread

The seed emissions build one node. That node takes index 0 in every scope, and a program's own nodes start at index 1. This is the one shift the emission thread introduces.

Emit

An Emit is its own node. It reads the previous emissions and the emitted value, and its value is the previous emissions with this name's stream advanced. Every entry carries a stable id, minted once from a running counter, so a list can be reconciled by identity later. It keeps its inputs, so it recomputes if the emitted value changes or resolves.

Reactive._nextEntryId = 1;
Reactive._entry = (id, value) => {
  return { id, value };
};
Reactive._mintEntryId = () => {
  const id = Reactive._nextEntryId;
  Reactive._nextEntryId ++;
  return id;
};

Every entry gets a stable id of its own

Now that entries are in play, _value earns its array branch. On the values page it resolved a reference and passed a plain value through. Here it also normalises an entry array: it walks the elements, normalising each, and rebuilds the array only if something changed, returning the original array untouched when nothing did. An element already an entry keeps its id with its value re-walked; a plain element is given a freshly minted one. The object branch walks nested references the same way:

Reactive._value = (value) => {
  if (Reactive._isReactive(value)) {
    return value.value;
  }
  if (Array.isArray(value)) {
    let changed = false;
    const outArray = [];
    value.forEach((item) => {
      if (Reactive._isEntry(item)) {
        const normalisedItemValue = Reactive._value(item.value);
        if (normalisedItemValue === item.value) {
          outArray.push(item);
        } else {
          changed = true;
          outArray.push(Reactive._entry(item.id, normalisedItemValue));
        }
      } else {
        changed = true;
        outArray.push(Reactive._entry(Reactive._mintEntryId(), Reactive._value(item)));
      }
    });
    return changed ? outArray : value;
  }
  if (value && typeof value === 'object') {
    let changed = false;
    const outObject = {};
    Object.keys(value).forEach((key) => {
      const normalisedFieldValue = Reactive._value(value[key]);
      outObject[key] = normalisedFieldValue;
      if (normalisedFieldValue !== value[key]) {
        changed = true;
      }
    });
    return changed ? outObject : value;
  }
  return value;
};

_value gains entry-array normalisation; unchanged arrays return as-is

The work is _emitCompute. A ready value appends a new entry to the name's array. A value that is still pending is withheld: its entry is not added, so the array never holds a stand-in, and the stream is flagged loading with override: true instead. A ready value leaves the override at its previous state, defaulting to undefined so the name defers to the global. When the value later resolves, the emit node recomputes, the entry appears in its id order, and the flag clears. Every other name is carried across untouched.

Reactive._emitCompute = (emitPayload, inputNodeRefs) => {
  const previousEmissions = Reactive._value(inputNodeRefs[0]);
  const emitValueRef = inputNodeRefs[1];
  const emittedValue = Reactive._value(emitValueRef);
  const emittedPending = emitValueRef.pending;
  const emitLeaf = Reactive._entry(emitPayload.leafId, emittedValue);

  const previousStream = previousEmissions.emissionsByName[emitPayload.name];
  const previousArray = previousStream === undefined ? [] : previousStream.array;
  const previousOverride = previousStream === undefined ? undefined : previousStream.override;
  const array = emittedPending ? previousArray : previousArray.concat([emitLeaf]);   // withhold if pending
  const override = emittedPending ? true : previousOverride;   // withhold pending; ready inherits (undefined default)

  const emissionsByName = Object.assign({}, previousEmissions.emissionsByName);   // carries the other streams across
  emissionsByName[emitPayload.name] = { array, override };
  return { emissionsByName };
};

Reactive.Emit = (emissionName, emitValue) => {
  const currentScope = Reactive._currentScope;
  const previousEmissionsRef = currentScope.currentEmissionsRef;
  const emitValueRef = Reactive._resolveInput(emitValue);
  const inputNodeRefs = [previousEmissionsRef, emitValueRef];
  const leafId = Reactive._mintEntryId();
  const payload = { emit: { name: emissionName, leafId } };
  const emissions = Reactive._emitCompute(payload.emit, inputNodeRefs);
  // the emit's node value carries the per-name streams; its global pending carries from the previous
  // emissions node (an emit contributes to a named stream, not the global).
  currentScope.currentEmissionsRef = Reactive._nodeCreate(payload, emissions, inputNodeRefs, previousEmissionsRef.pending, false);
};

Append a ready value, withhold a pending one, and thread the result on

Gather

Gather is a node of its own. Its value is one name's array, already flat and in emission order, and it carries that name's effective pending too, so a gather over a still-loading stream reads as loading. It reads the current emissions, so it reflows whenever a stream advances or resolves.

Reactive._emitGatherArray = (gatherPayload, emissionsRef) => {
  const stream = emissionsRef.value.emissionsByName[gatherPayload.name];
  return stream === undefined ? Reactive._emptyArray : stream.array;
};

Reactive._emitGatherPending = (gatherPayload, emissionsRef) => {
  return Reactive._emitStreamPending(emissionsRef, gatherPayload.name);
};

A gather reads a name's array, and that name's effective pending

Having read the name, it restarts it in the thread as an empty stream with an explicit not-pending override (override: false). That override wins over a later global, so a consumed name is not spuriously re-flagged pending by a pending If or Each: an enclosing gather of the same name re-collects nothing and does not read the fresh stream as loading. It is restarted, not deleted: a missing stream would defer to the node's global and read pending, which a just-emptied stream is not. The node's own global carries for free through _anyPending in this calculation.

Reactive._emitRemoveArray = (existingEmissionsRef, gatherName) => {
  return Reactive.Calculate([existingEmissionsRef, gatherName], (existingEmissions, existingName) => {
    const emissionsByName = Object.assign({}, existingEmissions.emissionsByName);
    emissionsByName[existingName] = { array: Reactive._emptyArray, override: false };
    return { emissionsByName };
  });
};

Reactive.Gather = (gatherName) => {
  const currentScope = Reactive._currentScope;
  const currentEmissionsRef = currentScope.currentEmissionsRef;
  const payload = { gather: { name: gatherName } };
  const inputNodeRefs = [currentEmissionsRef];
  const gatheredArray = Reactive._emitGatherArray(payload.gather, currentEmissionsRef);
  const gatheredPending = Reactive._emitGatherPending(payload.gather, currentEmissionsRef);
  const gatheredRef = Reactive._nodeCreate(payload, gatheredArray, inputNodeRefs, gatheredPending, false);
  currentScope.currentEmissionsRef = Reactive._emitRemoveArray(currentEmissionsRef, gatherName);
  return gatheredRef;
};

Gather a name's array and pending, then restart the name empty in the thread

If Threads Its Body

Now If earns its value. Its value is its body's emissions, threaded onto the parent through the if node, not a snapshot, so a change inside the body reflows the gathers below it. Merging two emissions is _emitMerge: per name it concatenates the arrays and combines the overrides. There is no global to OR here: the global rides the append Calculate's own node, whose pending is _anyPending of the existing and incoming refs, so the OR happens for free. Combining overrides is short: loading wins, then not-loading, else neutral.

Reactive._emitCombineOverride = (existingStream, incomingStream) => {
  const existingOverride = existingStream === undefined ? undefined : existingStream.override;
  const incomingOverride = incomingStream.override;
  if (existingOverride === true || incomingOverride === true) {
    return true;
  }
  if (existingOverride === false || incomingOverride === false) {
    return false;
  }
  return undefined;
};

Reactive._emitMerge = (existingEmissions, incomingEmissions) => {
  const incomingNames = Object.keys(incomingEmissions.emissionsByName);
  if (incomingNames.length === 0) {
    return existingEmissions;
  }
  const emissionsByName = Object.assign({}, existingEmissions.emissionsByName);
  incomingNames.forEach((incomingName) => {
    const existingStream = emissionsByName[incomingName];
    const incomingStream = incomingEmissions.emissionsByName[incomingName];
    const array = existingStream === undefined ? incomingStream.array : existingStream.array.concat(incomingStream.array);
    const override = Reactive._emitCombineOverride(existingStream, incomingStream);
    emissionsByName[incomingName] = { array, override };
  });
  return { emissionsByName };
};

Reactive._emitAppend = (existingEmissionsRef, incomingEmissionsRef) => {
  return Reactive.Calculate([existingEmissionsRef, incomingEmissionsRef], (existingEmissions, incomingEmissions) => {   // node.pending = _anyPending([existing, incoming]) = the global-OR, for free
    return Reactive._emitMerge(existingEmissions, incomingEmissions);
  });
};

Merge per name: concat arrays and combine overrides; the append node's pending is the global-OR

And If grows an emissions value and a global pending, kept apart. Its value is its body's emissions, or the empty value when the condition is plainly false or still loading. Its pending is true while the condition loads (the body is not built yet, so any name might still emit), otherwise the body scope's own global. Either way it threads the if node onto the parent, so a change to its value reruns the merge below it. A scope's emissions value and global read as _emitScopeValue and _emitScopePending, straight off the scope object:

Reactive._emitScopeValue = (scope) => {
  return scope.currentEmissionsRef.value;
};
Reactive._emitScopePending = (scope) => {
  return scope.currentEmissionsRef.pending === true;
};

Reactive._emitIfValue = (conditionPending, bodyScope) => {
  if (conditionPending || bodyScope === undefined) {
    return Reactive._emitEmpty;
  }
  return Reactive._emitScopeValue(bodyScope);
};
Reactive._emitIfPending = (conditionPending, bodyScope) => {
  if (conditionPending) {
    return true;
  }
  return bodyScope !== undefined && Reactive._emitScopePending(bodyScope);
};

Reactive.If = (condition, callback) => {
  const currentScope = Reactive._currentScope;
  const conditionRef = Reactive._resolveInput(condition);
  const conditionIsPresent = Reactive._value(conditionRef);

  const inputNodeRefs = [conditionRef];
  const pending = Reactive._anyPending(inputNodeRefs);
  const payload = { if_: { optionalScope: undefined, callback, contentsUpdated: false } };
  const ifNode = Reactive._nodeCreate(payload, Reactive._emitEmpty, inputNodeRefs, pending, false);

  const bodyScope = conditionIsPresent ? Reactive._scopeCreate(ifNode, callback) : undefined;
  payload.if_.optionalScope = bodyScope;
  ifNode.value = Reactive._emitIfValue(pending, bodyScope);
  ifNode.pending = Reactive._emitIfPending(pending, bodyScope);

  currentScope.currentEmissionsRef = Reactive._emitAppend(currentScope.currentEmissionsRef, ifNode);
};

An If contributes its body's emissions as its value, and a global pending as its node flag

Each

Each walks a gathered array and calls back with each value in order. Unlike an If, which owns one body scope, an Each owns one scope per element, and threads all of their emissions onto the parent through the each node. So a Gather after the Each collects them, and a chain of one Each into another stays flat, one element per value. The array it walks is an entry list, the same {id, value} shape a stream holds. Reading it is _arrayToList: the entry list itself, and the undefined of a never-emitted name reads as the empty list.

Reactive._arrayToList = (array) => {
  return array === undefined ? [] : array;
};

The entry list itself; a never-emitted name's undefined reads as []

An entry is not tagged; it is a shape, the same {id, value} every stream holds. A value reads as an entry when it is an object with a numeric id and a value, and an entry array is one whose first element is an entry. The array an Each walks is exactly such an entry array:

// whether a value is one entry: an object with a numeric id and a value. Shape, not a tag (see the
// section note). A plain field object that happens to carry a numeric id and a value reads as an
// entry, which is the deliberate cost of dropping the marker.
Reactive._isEntry = (value) => {
  return value !== undefined && typeof value === 'object' && typeof value.id === 'number' && 'value' in value;
};

// whether a value is an entry array: an array whose first element is an entry (an empty array
// qualifies, there being nothing to say it is not one).
Reactive._isEntryArray = (value) => {
  if (!Array.isArray(value)) {
    return false;
  }
  return value.length === 0 || Reactive._isEntry(value[0]);
};

An entry is a shape, {id, value}; an entry array is one whose first element is an entry

Each element gets its own scope, owned by the each node so it is torn down with it. The callback runs inside that scope, so anything it emits lands in that entry's own emissions thread:

Reactive._eachEntryCreate = (eachNodeRef, callback, element) => {
  return Reactive._scopeCreate(eachNodeRef, () => {
    callback(element.value);
  });
};

An entry's own scope, owned by the each node

The each node's value is the merge of every entry scope's emissions, folded in order with the same _emitMerge an If uses. Its separate global pending is true if any one entry scope is globally pending:

Reactive._emitCombineEntries = (entries) => {
  const entryLeaves = Reactive._arrayToList(entries);
  let combinedEmissions = Reactive._emitEmpty;
  entryLeaves.forEach((entryLeaf) => {
    const entryEmissions = Reactive._emitScopeValue(entryLeaf.value);
    combinedEmissions = Reactive._emitMerge(combinedEmissions, entryEmissions);
  });
  return combinedEmissions;
};

Reactive._emitEntriesPending = (entries) => {
  const entryLeaves = Reactive._arrayToList(entries);
  return entryLeaves.some((entryLeaf) => {
    return Reactive._emitScopePending(entryLeaf.value);
  });
};

Merge every entry scope's emissions; the global is true if any entry is pending

As with If, the each node gets its value and its global pending apart. Its value is empty while the array is still pending (the entries are not built yet), else the combined entries. Its pending is true while the array loads, else the entries' own global:

Reactive._emitEachValue = (arrayPending, entries) => {
  if (arrayPending) {
    return Reactive._emitEmpty;
  }
  return Reactive._emitCombineEntries(entries);
};
Reactive._emitEachPending = (arrayPending, entries) => {
  if (arrayPending) {
    return true;
  }
  return Reactive._emitEntriesPending(entries);
};

Empty while the array is pending, else the combined entries; pending mirrors it

Now Each itself. It creates the each node first, so each entry scope can own it, then builds one entry scope per element and records its stable id. With the entries built, it fills the node's value and pending, then appends the each node onto the parent thread, exactly as an If does:

Reactive.Each = (array, callback) => {
  const currentScope = Reactive._currentScope;
  const arrayRef = Reactive._resolveInput(array);
  const arrayValue = Reactive._value(arrayRef);
  const elements = Reactive._arrayToList(arrayValue);
  const inputNodeRefs = [arrayRef];
  const pending = Reactive._anyPending(inputNodeRefs);

  // create the each node first, so each entry scope can own it; fill its value once entries are built.
  const payload = { each: { callback, entries: [], array: arrayValue, optionalContentsUpdatedScopes: undefined } };
  const eachNode = Reactive._nodeCreate(payload, Reactive._emitEmpty, inputNodeRefs, false, false);
  const entries = [];
  elements.forEach((element) => {
    const entryScope = Reactive._eachEntryCreate(eachNode, callback, element);
    entries.push({ id: element.id, value: entryScope });
  });
  payload.each.entries = entries;
  eachNode.value = Reactive._emitEachValue(pending, entries);
  eachNode.pending = Reactive._emitEachPending(pending, entries);

  currentScope.currentEmissionsRef = Reactive._emitAppend(currentScope.currentEmissionsRef, eachNode);
};

Own a scope per element, thread their emissions, and track a global pending

Drivers

Gathered output is where the outside world attaches. A driver takes a stream a program produced under its name and turns it into effect: it draws to the page, runs a request, or ticks an animation. As a result, Reactive.Root grows a second argument, a named set of drivers.

You need nothing new to collect them. Whatever a program emits and never gathers stays in the root scope's currentEmissionsRef, in emission order. The root scope has no enclosing If to gather it. As a result, those leftover streams, read by name, are the drivers' output. Root only needs to keep the driver set:

Reactive.Root = (programCallback, drivers) => {
  const previousRoot = Reactive._currentRoot;
  const root = {
    drivers: drivers || {},   // the named drivers
    rootScope: undefined,
    nextReverseId: 1,
    transactionsHostRef: undefined,
  };
  Reactive._currentRoot = root;
  try {
    root.rootScope = Reactive._scopeCreate(undefined, programCallback);
    return root;
  } finally {
    Reactive._currentRoot = previousRoot;
  }
};

Root keeps the driver set; the leftover emissions are their streams

A program emits under a driver's name from anywhere inside it. The root scope's final emissions then hold each driver's stream, by name:

const root = Reactive.Root(() => {
  Reactive.Emit('dom', 'a');
  Reactive.Emit('dom', 'b');
  Reactive.Emit('log', 'x');
}, { dom: {}, log: {} });

const rootScope = root.rootScope;
const emissions = rootScope.currentEmissionsRef.value;
emissions.emissionsByName.dom.array;   // [ {id, value: 'a'}, {id, value: 'b'} ]   the 'dom' stream
emissions.emissionsByName.log.array;   // [ {id, value: 'x'} ]                      the 'log' stream

Each driver's stream is its leftover, un-gathered emissions

Running It

A Gather reads a value straight back under the name it was emitted:

let gathered;
const root = Reactive.Root(() => {
  const one = Reactive.Constant(1);
  const two = Reactive.State(2);
  const sum = Reactive.Calculate([one, two], (a, b) => a + b);
  Reactive.Emit('xyz', sum);
  gathered = Reactive.Gather('xyz');
});
// gathered flattens to [3]

Emit then Gather

Emissions keep document order across an If, whether its body is present or absent, and across an Each that re-emits or filters:

// Emit 1, If(true){ Emit 2; Emit 3 }, Emit 4, Gather   ->  [1, 2, 3, 4]
// Emit 1, If(false){ Emit 2 },        Emit 4, Gather   ->  [1, 4]

// Emit 10,20,30 under 'src', then over the gather:
Reactive.Each(src, (v) => Reactive.Emit('out', v));            // out -> [10, 20, 30]
Reactive.Each(src, (v) => Reactive.If(v > 12, () => Reactive.Emit('out', v)));   // [15, 20]

Document order holds through If and Each

Next Steps

Now a program has outputs. Each name is a stream the program builds, and a name it never gathers stays in the root scope, ready to hand out. A whole named stream carries its own pending flag. As a result, the flag marks an output that still loads. Next, Drivers take those leftover streams and turn them into real effect: paint the DOM, start a timer, fetch.

Drivers →