WebSocket Server

Before you run a pipeline with the WebSocket Server origin, complete the following prerequisites to configure the WebSocket clients.

The WebSocket Server origin performs parallel processing and enables the creation of a multithreaded pipeline.

The WebSocket Server origin uses multiple concurrent threads based on the Max Concurrent Requests property. Each thread connects to the origin system, creates a batch of data, and passes the batch to an available pipeline runner.

A pipeline runner is a sourceless pipeline instance - an instance of the pipeline that includes all of the processors and destinations in the pipeline and represents all pipeline processing after the origin. Each pipeline runner processes one batch at a time, just like a pipeline that runs on a single thread. When the flow of data slows, the pipeline runners wait idly until they are needed, generating an empty batch at regular intervals. You can configure the Runner Idle Time pipeline property to specify the interval or to opt out of empty batch generation.

Multithreaded pipelines preserve the order of records within each batch, just like a single-threaded pipeline. But since batches are processed by different pipeline instances, the order that batches are written to destinations is not ensured.

For example, say you set the Max Concurrent Requests property to 5. When you start the pipeline, the origin creates five threads, and Data Collector creates a matching number of pipeline runners. Upon receiving data, the origin passes a batch to each of the pipeline runners for processing. In the batch, WebSocket Server includes only the WebSocket requests with the specified application ID.

At any given moment, the five pipeline runners can each process a batch, so this multithreaded pipeline processes up to five batches at a time. When incoming data slows, the pipeline runners sit idle, available for use as soon as the data flow increases.

For more information about multithreaded pipelines, see Multithreaded Pipeline Overview.

Unlike standard origins, the WebSocket Server origin can receive records from microservice destinations in a microservice pipeline. When the pipeline is configured to use the Send Response to Origin pipeline error handling, the origin also receives all pipeline error records. The origin generates a response with all of the records that it receives from the pipeline and sends the response back to the originating WebSocket client.

Just as each request that the WebSocket Server origin receives is processed as a batch, the records from the batch that are passed to the origin from the pipeline are incorporated into a single response.

In the generated response, records that were passed by microservice destinations in the pipeline are included as a list in a data attribute. Error records passed by the Send Response to Origin pipeline error handling are included as a list in an error attribute.

{
"httpStatusCode":<status code>,
"data":[<list of success records>],
"error":[<list of error records>],
"errorMessage": <error message, if any>
}

The WebSocket Server origin processes data differently based on the data format that you select. The origin processes the following types of data:

Binary

Generates a record with a single byte array field at the root of the record.

When the data exceeds the user-defined maximum data size, the origin cannot process the data. Because the record is not created, the origin cannot pass the record to the pipeline to be written as an error record. Instead, the origin generates a stage error.

Datagram

Generates a record for every message. The origin can process collectd messages, NetFlow 5 and NetFlow 9 messages, and the following types of syslog messages:

• RFC 5424
• RFC 3164
• Non-standard common messages, such as RFC 3339 dates with no version digit

When processing NetFlow messages, the stage generates different records based on the NetFlow version. When processing NetFlow 9, the records are generated based on the NetFlow 9 configuration properties. For more information, see NetFlow Data Processing.

Delimited

Generates a record for each delimited line. You can use the following delimited format types:

• Default CSV - File that includes comma-separated values. Ignores empty lines in the file.
• RFC4180 CSV - Comma-separated file that strictly follows RFC4180 guidelines.
• MS Excel CSV - Microsoft Excel comma-separated file.
• MySQL CSV - MySQL comma-separated file.
• PostgreSQL CSV - PostgreSQL comma-separated file.
• PostgreSQL Text - PostgreSQL text file.
• Tab-Separated Values - File that includes tab-separated values.
• Custom - File that uses user-defined delimiter, escape, and quote characters.

You can use a list or list-map root field type for delimited data, optionally including the header information when available. For more information about the root field types, see Delimited Data Root Field Type.

When using a header line, you can allow processing records with additional columns. The additional columns are named using a custom prefix and integers in sequential increasing order, such as _extra_1, _extra_2. When you disallow additional columns when using a header line, records that include additional columns are sent to error.

You can also replace a string constant with null values.

When a record exceeds the maximum record length defined for the origin, the origin processes the object based on the error handling configured for the stage.

JSON

Generates a record for each JSON object. You can process JSON files that include multiple JSON objects or a single JSON array.

When an object exceeds the maximum object length defined for the origin, the origin processes the object based on the error handling configured for the stage.

Log

Generates a record for every log line.

When a line exceeds the user-defined maximum line length, the origin truncates longer lines.

You can include the processed log line as a field in the record. If the log line is truncated, and you request the log line in the record, the origin includes the truncated line.

You can define the log format or type to be read.

Protobuf

Generates a record for every protobuf message. By default, the origin assumes messages contain multiple protobuf messages.

Protobuf messages must match the specified message type and be described in the descriptor file.

When the data for a record exceeds 1 MB, the origin cannot continue processing data in the message. The origin handles the message based on the stage error handling property and continues reading the next message.

For information about generating the descriptor file, see Protobuf Data Format Prerequisites.

SDC Record

Generates a record for every record. Use to process records generated by a Data Collector pipeline using the SDC Record data format.

For error records, the origin provides the original record as read from the origin in the original pipeline, as well as error information that you can use to correct the record.

When processing error records, the origin expects the error file names and contents as generated by the original pipeline.

Text

Generates a record for each line of text or for each section of text based on a custom delimiter.

When a line or section exceeds the maximum line length defined for the origin, the origin truncates it. The origin adds a boolean field named Truncated to indicate if the line was truncated.

For more information about processing text with a custom delimiter, see Text Data Format with Custom Delimiters.

XML

Generates records based on a user-defined delimiter element. Use an XML element directly under the root element or define a simplified XPath expression. If you do not define a delimiter element, the origin treats the XML file as a single record.

Generated records include XML attributes and namespace declarations as fields in the record by default. You can configure the stage to include them in the record as field attributes.

You can include XPath information for each parsed XML element and XML attribute in field attributes. This also places each namespace in an xmlns record header attribute.

Note: Field attributes and record header attributes are written to destination systems automatically only when you use the SDC RPC data format in destinations. For more information about working with field attributes and record header attributes, and how to include them in records, see Field Attributes and Record Header Attributes.

When a record exceeds the user-defined maximum record length, the origin skips the record and continues processing with the next record. It sends the skipped record to the pipeline for error handling.

Use the XML data format to process valid XML documents. For more information about XML processing, see Reading and Processing XML Data.

Tip: If you want to process invalid XML documents, you can try using the text data format with custom delimiters. For more information, see Processing XML Data with Custom Delimiters.