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 two fields: a pending flag and 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 top-level pending is the global flag. 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".

{
  pending: true,                                            // the global "all might be pending" flag
  emissionsByName: {
    dom: { array: [ {id, value}, ... ], override: undefined },   // a ready stream
    img: { array: [],                   override: true      },   // this stream is loading
  },
}

An emissions value: a global pending flag, plus a stream per name

An override is three-state. undefined means "defer to the global". true means "this stream is loading" (a value was emitted while still pending, and withheld). false means "this stream is freshly gathered, exclude it from the global". So the effective pending of a name is its own override if it has one, else the global flag:

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

A name's effective pending: its override, or the global 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 = { pending: false, 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._mintEntryId = () => { return Reactive._nextEntryId++; };
Reactive._entry = (id, value) => { return { id, value }; };

Every entry gets a stable id of its own

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. 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, including the global-pending key.

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

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

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

Reactive.Emit = (emissionName, emitValue) => {
  const currentScope = Reactive._currentScope;
  const emitValueRef = Reactive._resolveInput(emitValue);
  const inputNodeRefs = [currentScope.currentEmissionsRef, emitValueRef];
  const payload = { emit: { name: emissionName, leafId: Reactive._mintEntryId() } };
  const emissions = Reactive._emitCompute(payload.emit, inputNodeRefs);
  currentScope.currentEmissionsRef = Reactive._nodeCreate(payload, emissions, inputNodeRefs, false, 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 emissions = Reactive._value(emissionsRef);
  const stream = emissions.emissionsByName[gatherPayload.name];
  return stream === undefined ? Reactive._emptyArray : stream.array;
};

Reactive._emitGatherPending = (gatherPayload, emissionsRef) => {
  const emissions = Reactive._value(emissionsRef);
  return Reactive._emitStreamPending(emissions, 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, explicitly-not-pending stream (override: false), so 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 global flag and read pending, which a just-emptied stream is not.

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 { pending: existingEmissions.pending, emissionsByName };
  });
};

Reactive.Gather = (gatherName) => {
  const currentScope = Reactive._currentScope;
  const current = currentScope.currentEmissionsRef;
  const payload = { gather: { name: gatherName } };
  const array = Reactive._emitGatherArray(payload.gather, current);
  const pending = Reactive._emitGatherPending(payload.gather, current);
  const gatheredRef = Reactive._nodeCreate(payload, array, [current], pending, false);
  currentScope.currentEmissionsRef = Reactive._emitRemoveArray(current, 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, and it ORs the global flags. Combining overrides is short: loading wins, then not-loading, else neutral.

Reactive._emitGlobalPending = () => { return { pending: true, emissionsByName: {} }; };

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

Reactive._emitMerge = (existingEmissions, incomingEmissions) => {
  const incomingNames = Object.keys(incomingEmissions.emissionsByName);
  const incomingGlobal = incomingEmissions.pending === true;
  if (incomingNames.length === 0 && !incomingGlobal) { return existingEmissions; }
  const emissionsByName = Object.assign({}, existingEmissions.emissionsByName);
  if (incomingGlobal) {
    // a pending If/Each could emit under any name, so re-expose a gathered-fresh stream to the global
    Object.keys(emissionsByName).forEach((name) => {
      if (emissionsByName[name].override === false) { emissionsByName[name] = { array: emissionsByName[name].array, override: undefined }; }
    });
  }
  incomingNames.forEach((name) => {
    const existingStream = emissionsByName[name];
    const incomingStream = incomingEmissions.emissionsByName[name];
    const array = existingStream === undefined ? incomingStream.array : existingStream.array.concat(incomingStream.array);
    emissionsByName[name] = { array, override: Reactive._emitCombineOverride(existingStream, incomingStream) };
  });
  return { pending: existingEmissions.pending === true || incomingGlobal, emissionsByName };
};

Reactive._emitAppend = (existingEmissionsRef, incomingEmissionsRef) => {
  return Reactive.Calculate([existingEmissionsRef, incomingEmissionsRef], (existingEmissions, incomingEmissions) => {
    return Reactive._emitMerge(existingEmissions, incomingEmissions);
  });
};

Merge per name: concat arrays, combine overrides, OR the global flag

And If grows an emissions value. When its condition is still pending the body is not built yet, so it contributes only the global-pending flag. Otherwise its value is its body's emissions, or the empty value when the condition is plainly false. 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 read as _emitScopeValue, straight off the scope object:

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

Reactive._emitIfValue = (conditionPending, bodyScope) => {
  if (conditionPending) { return Reactive._emitGlobalPending(); }
  return bodyScope === undefined ? Reactive._emitEmpty : Reactive._emitScopeValue(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);

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

An If contributes its body's emissions, or just the global flag while its condition loads

Each

Each walks a gathered array and calls back with each value in order. The callback emits, and those emissions thread into the current scope. As a result, a Gather after the Each collects them. A chain of one Each into another stays flat, one element per value:

Reactive.Each = (array, callback) => {
  const arrayValue = Reactive._value(array);
  const elements = Reactive._arrayToList(arrayValue);
  elements.forEach((element) => {
    callback(element.value);
  });
};

Walk a gathered list; the callback emits into the scope

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 →