Configuration.mux()

Description

A mux filter does the following:

• At the input to a mux filter, the Message is merged to a sub-pipeline that is shared by multiple mux filters
• At the input to the shared sub-pipeline, Messages from multiple mux filters are queued up into one single Message stream
• Whatever is outputted from the shared sub-pipeline is discared
• At the output from the mux filter, out come the same Mesasge as its input

Multiple mux filters can have one shared sub-pipeline, in which many input Messages are queued up.

Stream sharing

You can control where a mux filter merges its input Message to by specifying a "merging target". It can be a value of any type or a function that returns it. Filters with the same merging target share and merge to the same sub-pipeline.

Merging only happens among filters coming from the same place in the same pipeline layout. Two filters from different pipeline layouts or different places in the same pipeline layout will never merge to the same sub-pipeline, even when they have the same merging target.

Sub-pipeline lifecycle

Since the sub-pipeline is shared by multiple mux filter instances, it won't be closed until the last mux ends. After all mux filter instances sharing it are done, the sub-pipeline will wait for 60 seconds to ensure no more new mux filters come and merge to it. You can change this waiting time by the maxIdle option in the options parameter. It can be either a number in seconds or a string with a unit suffix such as 's', 'm', 'h', 'd'.

When the merging target is an object, it will be a weak ref, just like a key in a WeakMap. When the object is dead, so is the sub-pipeline being weakly referenced by the object, regardless of the idleTime option.

Syntax

pipy()
  .pipeline()
  .mux().to(
    subPipelineLayout
  )

pipy()
  .pipeline()
  .mux(
    () => whereToMerge()
  ).to(
    subPipelineLayout
  )

pipy()
  .pipeline()
  .mux({
    maxIdle,
    maxQueue,
  }).to(
    subPipelineLayout
  )

pipy()
  .pipeline()
  .mux(
    () => whereToMerge(),
    {
      maxIdle,
      maxQueue,
    }
  ).to(
    subPipelineLayout
  )

Parameters

mux(sessionSelector, options?)

mux(options?)

See Also

• Configuration
• demux()
• demuxQueue()
• demuxHTTP()
• muxQueue()
• muxHTTP()
• serveHTTP()