Configuration.muxHTTP()

Description

A muxHTTP filter does the following:

• At the input to a muxHTTP filter, an HTTP request is merged to a sub-pipeline that is shared by multiple muxHTTP filters
• At the input to the shared sub-pipeline, HTTP requests from multiple muxHTTP filters are encoded and multiplexed into one single Data stream
• At the output from the shared sub-pipeline, Data stream is de-multiplexed and decoded before individual responses are distributed back to their corresponding muxHTTP filters
• At the output from the muxHTTP filter, out come an HTTP response as a Message

Multiple muxHTTP filters can have one shared sub-pipeline, in which many HTTP requests and responses are multiplexed. Compare this to demuxHTTP().

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.

HTTP versions

You can select between HTTP/1.1 and HTTP/2 by using the option version in the options parameter. It can be 1 for HTTP/1.1, or 2 for HTTP/2. You can also specify a callback function that gets called at session start and returns the desired protocol version. Default value is 1.

Multiplexing in HTTP/1.1

In HTTP/1.1, multiplexing is done by HTTP pipelining. A message queue is used internally by each group of muxHTTP filters sharing the same sub-pipeline, so that each filter can pick the right response for its request from the sub-pipeline's output.

Multiplexing in HTTP/2

In HTTP/2, multiplexing is inherently supported by the protocol itself. With HTTP/2, every muxHTTP filter runs a dedicated virtual stream in an HTTP/2 transport stream. There is no shared queue to wait as in HTTP/1.1, so responses to the requests can arrive out of order without having to wait for each other.

Chunked transfer

When encoding an HTTP/1.x request, the Content-Length header needs to come before the body, so muxHTTP has to buffer the entire body until it sees a MessageEnd event, only by then can the filter output a value for Content-Length header, followed by the previously buffered body.

The buffering is limited to 4KB by default. When the buffered data is over 4KB, the encoder will opt for chunked encoding, where a Content-Length header is no longer needed. You can change this limit by the option bufferSize in the options parameter.

Syntax

pipy()
  .pipeline()
  .muxHTTP().to(
    subPipelineLayout
  )

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

pipy()
  .pipeline()
  .muxHTTP({
    maxIdle,
    maxQueue,
    bufferSize,
    version,
  }).to(
    subPipelineLayout
  )

pipy()
  .pipeline()
  .muxHTTP(
    () => whereToMerge(),
    {
      maxIdle,
      maxQueue,
      bufferSize,
      version,
    }
  ).to(
    subPipelineLayout
  )

Parameters

muxHTTP(sessionSelector, options?)

muxHTTP(options?)

Example

pipy()

  .listen(8000)                       // Accept TCP connection on port 8000
  .demuxHTTP().to(                    // De-multiplex and decode HTTP requests
    $=>$.muxHTTP().to(                // Encode and multiplex the request into a shared TCP stream
      $=>$.connect('localhost:8080')  // Send to and receive from localhost:8080
    )
  )

See Also

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