Convert input streams to nmsg format. More...

Enumerations
enum	nmsg_input_type { nmsg_input_type_stream , nmsg_input_type_pcap , nmsg_input_type_pres , nmsg_input_type_callback , nmsg_input_type_json }
	An enum identifying the underlying implementation of an nmsg_input_t object. More...

Functions
nmsg_input_t	nmsg_input_open_file (int fd)
	Initialize a new NMSG stream input from a byte-stream file source.

nmsg_input_t	nmsg_input_open_sock (int fd)
	Initialize a new NMSG stream input from a datagram socket source.

nmsg_input_t	nmsg_input_open_zmq (void *s)
	Initialize a new NMSG stream input from a ZMQ socket source.

nmsg_input_t	nmsg_input_open_zmq_endpoint (void zmq_ctx, const char ep)
	Create an ZMQ socket and initialize a new NMSG stream input from it.

nmsg_input_t	nmsg_input_open_callback (nmsg_cb_message_read cb, void *user)
	Initialize a new nmsg input closure.

nmsg_input_t	nmsg_input_open_null (void)
	Initialize a new "null source" NMSG stream input.

nmsg_input_t	nmsg_input_open_pres (int fd, nmsg_msgmod_t msgmod)
	Initialize a new NMSG presentation form input from a file descriptor.

nmsg_input_t	nmsg_input_open_json (int fd)
	Initialize a new NMSG JSON form input from a file descriptor.

nmsg_input_t	nmsg_input_open_pcap (nmsg_pcap_t pcap, nmsg_msgmod_t msgmod)
	Initialize a new NMSG pcap input from a pcap descriptor.

nmsg_res	nmsg_input_close (nmsg_input_t *input)
	Close an nmsg_input_t object and release all associated resources.

nmsg_res	nmsg_input_loop (nmsg_input_t input, int count, nmsg_cb_message cb, void *user)
	Loop over an input stream and call a user-provided function for each payload.

void	nmsg_input_breakloop (nmsg_input_t input)
	Break out of an nmsg_input_loop() early.

nmsg_res	nmsg_input_read (nmsg_input_t input, nmsg_message_t *msg)
	Read one NMSG message from an input stream.

nmsg_res	nmsg_input_read_null (nmsg_input_t input, uint8_t buf, size_t buf_len, struct timespec ts, nmsg_message_t *msg, size_t n_msg)
	Read zero, one, or more NMSG messages from a "null source" input.

void	nmsg_input_set_filter_msgtype (nmsg_input_t input, unsigned vid, unsigned msgtype)
	Filter an nmsg_input_t for a given vendor ID / message type.

nmsg_res	nmsg_input_set_filter_msgtype_byname (nmsg_input_t input, const char vname, const char mname)
	Filter an nmsg_input_t for a given vendor ID / message type.

void	nmsg_input_set_filter_source (nmsg_input_t input, unsigned source)
	Set a source filter for input NMSG payloads.

void	nmsg_input_set_filter_operator (nmsg_input_t input, unsigned operator_)
	Set an operator filter for input NMSG payloads.

void	nmsg_input_set_filter_group (nmsg_input_t input, unsigned group)
	Set a group filter for input NMSG payloads.

nmsg_res	nmsg_input_set_blocking_io (nmsg_input_t input, bool flag)
	Configure non-blocking I/O for a stream input.

nmsg_res	nmsg_input_set_byte_rate (nmsg_input_t input, size_t rate)
	Set the target ingress byte rate for a stream input.

nmsg_res	nmsg_input_set_verify_seqsrc (nmsg_input_t input, bool verify)
	Enable or disable seqsrc verification on an NMSG stream nmsg_input_t object.

nmsg_res	nmsg_input_get_count_container_received (nmsg_input_t input, uint64_t *count)
	For UDP datagram socket nmsg_input_t objects, retrieve the total number of NMSG containers that have been received since the nmsg_input_t object was created.

nmsg_res	nmsg_input_get_count_container_dropped (nmsg_input_t input, uint64_t *count)
	For UDP datagram socket nmsg_input_t objects, retrieve the total number of NMSG containers that been dropped since the nmsg_input_t object was created.

Detailed Description

Convert input streams to nmsg format.

Nmsg can import data into a stream of payloads from several different input sources:

 \li Wire-format NMSG containers which contain one or more binary
 payloads that can be read from file or datagram socket sources. This is
 the native NMSG interchange format.

 \li libpcap packets from a pcap savefile or network interface that will
 be reassembled into IP datagrams and passed to a message format specific
 function for conversion into nmsg payloads.

 \li Presentation format data (ASCII lines) read from a file, converted
 by a message format specific function into nmsg payloads.

MP:

Clients must ensure synchronized access when reading from an nmsg_input_t object.

Reliability:

Clients must not touch the underlying file descriptor or pcap_t object. Cleanup will be handled by the nmsg_input_close() function.

Resources:

An internal buffer will be allocated and used until an nmsg_input_t object is destroyed.

Definition in file input.h.

Enumeration Type Documentation

◆ nmsg_input_type

enum nmsg_input_type

An enum identifying the underlying implementation of an nmsg_input_t object.

This is used for nmsg_io's close event notification.

Enumerator
nmsg_input_type_stream	NMSG payloads from file or socket.
nmsg_input_type_pcap	pcap packets from file or interface
nmsg_input_type_pres	presentation form
nmsg_input_type_json	JSON form.

Definition at line 54 of file input.h.

Function Documentation

◆ nmsg_input_open_file()

nmsg_input_t nmsg_input_open_file ( int fd )

Initialize a new NMSG stream input from a byte-stream file source.

Parameters

[in] fd Readable file descriptor from a byte-stream source.

Returns: Opaque pointer that is NULL on failure or non-NULL on success.

◆ nmsg_input_open_sock()

nmsg_input_t nmsg_input_open_sock ( int fd )

Initialize a new NMSG stream input from a datagram socket source.

Parameters

[in] fd Readable datagram socket.

Returns: Opaque pointer that is NULL on failure or non-NULL on success.

◆ nmsg_input_open_zmq()

nmsg_input_t nmsg_input_open_zmq ( void * s )

Initialize a new NMSG stream input from a ZMQ socket source.

Parameters

[in] s ZMQ socket.

Returns: Opaque pointer that is NULL on failure or non-NULL on success.

◆ nmsg_input_open_zmq_endpoint()

nmsg_input_t nmsg_input_open_zmq_endpoint	(	void *	zmq_ctx,
		const char *	ep
	)

Create an ZMQ socket and initialize a new NMSG stream input from it.

This function is a wrapper for nmsg_input_open_zmq(). Instead of taking an already initialized ZMQ socket object, it takes an endpoint argument like zmq_connect() and zmq_bind() do which is a string containing a "transport://address" specification and initializes an ZMQ socket object. However, this endpoint string will be munged in order to support additional functionality:

The caller may select between a bound or connected ZMQ socket by appending ",accept" or ",connect" to the endpoint argument. (If not given, this function behaves as if ",connect" was passed.) That is, ",accept" uses zmq_bind() to obtain a ZMQ endpoint, and ",connect" uses zmq_connect().

The caller may additionally select between a SUB socket or a PULL socket by appending ",pubsub" or ",pushpull". (If not given, this function behaves as if ",pubsub" was passed.)

See also: nmsg_output_open_zmq_endpoint()

Parameters

[in]	zmq_ctx	ZMQ context object.
[in]	ep	ZMQ endpoint (with nmsg-specific extensions)

Returns: Opaque pointer that is NULL on failure or non-NULL on success.

◆ nmsg_input_open_callback()

nmsg_input_t nmsg_input_open_callback	(	nmsg_cb_message_read	cb,
		void *	user
	)

Initialize a new nmsg input closure.

This allows a user-provided callback to function as an nmsg input, for instance to participate in an nmsg_io loop. The callback is responsible for creating an nmsg_message_t object and returning it to the caller.

Parameters

[in]	cb	Non-NULL function pointer.
[in]	user	Optionally NULL pointer which will be passed to the callback.

Returns: Opaque pointer that is NULL on failure or non-NULL on success.

◆ nmsg_input_open_null()

nmsg_input_t nmsg_input_open_null ( void )

Initialize a new "null source" NMSG stream input.

A "null source" means the actual gathering of input is not performed by the library but rather by the caller. A "null source" nmsg_input_t thus serves only to hold the state associated with the stream.

Calling nmsg_input_loop() or nmsg_input_read() on a "null source" input will fail. Callers instead need to use nmsg_input_read_null().

Returns: Opaque pointer that is NULL on failure or non-NULL on success.

◆ nmsg_input_open_pres()

nmsg_input_t nmsg_input_open_pres	(	int	fd,
		nmsg_msgmod_t	msgmod
	)

Initialize a new NMSG presentation form input from a file descriptor.

Parameters

[in]	fd	Readable file descriptor.
[in]	msgmod	Handle that implements the desired presentation form to NMSG conversion.

Returns: Opaque pointer that is NULL on failure or non-NULL on success.

◆ nmsg_input_open_json()

nmsg_input_t nmsg_input_open_json ( int fd )

Initialize a new NMSG JSON form input from a file descriptor.

See nmsg_output_open_json for details of the JSON format.

Parameters

[in] fd Readable file descriptor.

Returns: Opaque pointer that is NULL on failure or non-NULL on success.

◆ nmsg_input_open_pcap()

nmsg_input_t nmsg_input_open_pcap	(	nmsg_pcap_t	pcap,
		nmsg_msgmod_t	msgmod
	)

Initialize a new NMSG pcap input from a pcap descriptor.

Parameters

[in]	pcap	Descriptor returned by libpcap. Supported data link types are those supported by nmsg_ipdg_parse_pcap().
[in]	msgmod	Handle that implements the desired IP datagram to NMSG conversion.

Returns: Opaque pointer that is NULL on failure or non-NULL on success.

◆ nmsg_input_close()

nmsg_res nmsg_input_close ( nmsg_input_t * input )

Close an nmsg_input_t object and release all associated resources.

Parameters

[in] input Valid pointer to an nmsg_input_t object.

Returns: nmsg_res_success

◆ nmsg_input_loop()

nmsg_res nmsg_input_loop	(	nmsg_input_t	input,
		int	count,
		nmsg_cb_message	cb,
		void *	user
	)

Loop over an input stream and call a user-provided function for each payload.

Parameters

[in]	input	Valid nmsg_input_t.
[in]	count	Non-negative to indicate a finite number of payloads to process or negative to indicate all available payloads should be processed.
[in]	cb	Non-NULL function pointer that will be called once for each payload.
[in]	user	Optionally NULL pointer which will be passed to the callback.

Returns: Any of nmsg_input_read()'s return values.

◆ nmsg_input_breakloop()

void nmsg_input_breakloop ( nmsg_input_t input )

Break out of an nmsg_input_loop() early.

Parameters

[in] input Valid nmsg_input_t.

◆ nmsg_input_read()

nmsg_res nmsg_input_read	(	nmsg_input_t	input,
		nmsg_message_t *	msg
	)

Read one NMSG message from an input stream.

Parameters

[in]	input	Valid nmsg_input_t.
[out]	msg	Pointer to where an nmsg_message_t object may be stored.

Returns: nmsg_res_success; nmsg_res_failure; nmsg_res_again; nmsg_res_eof; nmsg_res_magic_mismatch; nmsg_res_version_mismatch

◆ nmsg_input_read_null()

nmsg_res nmsg_input_read_null	(	nmsg_input_t	input,
		uint8_t *	buf,
		size_t	buf_len,
		struct timespec *	ts,
		nmsg_message_t **	msg,
		size_t *	n_msg
	)

Read zero, one, or more NMSG messages from a "null source" input.

The caller must supply a buffer containing the serialized NMSG container. This function may return nmsg_res_success with n_msg set to zero, which indicates that the NMSG container contained a fragment.

Parameters

[in]	input	Valid "null source" nmsg_input_t.
[in]	buf	Input buffer containing a serialized NMSG container.
[in]	buf_len	Length of input buffer.
[in]	ts	Current "time". May be NULL to indicate the current wall clock time.
[out]	msg	Pointer to where an array of nmsg_message_t objects may be stored.
[out]	n_msg	Pointer to where the size of the output array will be stored.

Returns: nmsg_res_success; nmsg_res_again; nmsg_res_magic_mismatch; nmsg_res_version_mismatch

◆ nmsg_input_set_filter_msgtype()

void nmsg_input_set_filter_msgtype	(	nmsg_input_t	input,
		unsigned	vid,
		unsigned	msgtype
	)

Filter an nmsg_input_t for a given vendor ID / message type.

NMSG messages whose vid and and msgtype fields do not match the filter will be silently discarded when reading from the input.

Calling this function with vid=0 and msgtype=0 will disable the filter.

Parameters

[in]	input	nmsg_input_t object.
[in]	vid	Vendor ID.
[in]	msgtype	Message type.

◆ nmsg_input_set_filter_msgtype_byname()

nmsg_res nmsg_input_set_filter_msgtype_byname	(	nmsg_input_t	input,
		const char *	vname,
		const char *	mname
	)

Filter an nmsg_input_t for a given vendor ID / message type.

Parameters

[in]	input	nmsg_input_t object.
[in]	vname	Vendor ID name.
[in]	mname	Message type name.

◆ nmsg_input_set_filter_source()

void nmsg_input_set_filter_source	(	nmsg_input_t	input,
		unsigned	source
	)

Set a source filter for input NMSG payloads.

This is only effective for file, socket, and xs inputs. Only NMSG payloads whose source field matches the source filter will be output by nmsg_input_read() or nmsg_input_loop().

Parameters

[in]	input	NMSG stream nmsg_input_t object.
[in]	source	Source ID filter, 0 to disable.

◆ nmsg_input_set_filter_operator()

void nmsg_input_set_filter_operator	(	nmsg_input_t	input,
		unsigned	operator_
	)

Set an operator filter for input NMSG payloads.

This has no effect on non-NMSG inputs. Only NMSG payloads whose operator field matches the operator filter will be output by nmsg_input_read() or nmsg_input_loop().

Parameters

[in]	input	NMSG stream nmsg_input_t object.
[in]	operator_	Operator ID filter, 0 to disable.

◆ nmsg_input_set_filter_group()

void nmsg_input_set_filter_group	(	nmsg_input_t	input,
		unsigned	group
	)

Set a group filter for input NMSG payloads.

This has no effect on non-NMSG inputs. Only NMSG payloads whose group field matches the group filter will be output by nmsg_input_read() or nmsg_input_loop().

Parameters

[in]	input	NMSG stream nmsg_input_t object.
[in]	group	Group ID filter, 0 to disable.

◆ nmsg_input_set_blocking_io()

nmsg_res nmsg_input_set_blocking_io	(	nmsg_input_t	input,
		bool	flag
	)

Configure non-blocking I/O for a stream input.

Parameters

[in]	input	NMSG stream nmsg_input_t object.
[in]	flag	boolean value, true to clear O_NONBLOCK on the underlying file descriptor, false to set O_NONBLOCK.

Returns: nmsg_res_success; nmsg_res_failure

◆ nmsg_input_set_byte_rate()

nmsg_res nmsg_input_set_byte_rate	(	nmsg_input_t	input,
		size_t	rate
	)

Set the target ingress byte rate for a stream input.

If the target byte rate is positive, reading from the input may sleep in order to maintain the target consumption rate.

Setting this value to a non-positive value will disable ingress byte rate control.

Parameters

[in]	input	NMSG stream nmsg_input_t object.
[in]	rate	Target byte rate in bytes/second.

Returns: nmsg_res_success; nmsg_res_failure

◆ nmsg_input_set_verify_seqsrc()

nmsg_res nmsg_input_set_verify_seqsrc	(	nmsg_input_t	input,
		bool	verify
	)

Enable or disable seqsrc verification on an NMSG stream nmsg_input_t object.

Note that for stream nmsg_input_t objects, seqsrc verification is enabled by default.

Parameters

[in]	input	NMSG stream nmsg_input_t object.
[in]	verify	boolean value, true to enable verification, false to disable verification.

Returns: nmsg_res_success; nmsg_res_failure

◆ nmsg_input_get_count_container_received()

nmsg_res nmsg_input_get_count_container_received	(	nmsg_input_t	input,
		uint64_t *	count
	)

For UDP datagram socket nmsg_input_t objects, retrieve the total number of NMSG containers that have been received since the nmsg_input_t object was created.

Parameters

[in]	input	UDP socket based NMSG input object.
[out]	count	Total number of NMSG containers received by the nmsg_input_t object during its lifetime.

Returns: nmsg_res_success; nmsg_res_failure

◆ nmsg_input_get_count_container_dropped()

nmsg_res nmsg_input_get_count_container_dropped	(	nmsg_input_t	input,
		uint64_t *	count
	)

For UDP datagram socket nmsg_input_t objects, retrieve the total number of NMSG containers that been dropped since the nmsg_input_t object was created.

Sequence number tracking must have been previously enabled by a call to nmsg_input_set_verify_seqsrc().

Parameters

[in]	input	UDP socket based NMSG input object.
[out]	count	Number of NMSG containers determined to have been dropped by the nmsg_input_t object since sequence number tracking was enabled.

Returns: nmsg_res_success; nmsg_res_failure

Enumerations

Functions

Detailed Description

Enumeration Type Documentation

◆ nmsg_input_type

Function Documentation

◆ nmsg_input_open_file()

◆ nmsg_input_open_sock()

◆ nmsg_input_open_zmq()

◆ nmsg_input_open_zmq_endpoint()

◆ nmsg_input_open_callback()

◆ nmsg_input_open_null()

◆ nmsg_input_open_pres()

◆ nmsg_input_open_json()

◆ nmsg_input_open_pcap()

◆ nmsg_input_close()

◆ nmsg_input_loop()

◆ nmsg_input_breakloop()

◆ nmsg_input_read()

◆ nmsg_input_read_null()

◆ nmsg_input_set_filter_msgtype()

◆ nmsg_input_set_filter_msgtype_byname()

◆ nmsg_input_set_filter_source()

◆ nmsg_input_set_filter_operator()

◆ nmsg_input_set_filter_group()

◆ nmsg_input_set_blocking_io()

◆ nmsg_input_set_byte_rate()

◆ nmsg_input_set_verify_seqsrc()

◆ nmsg_input_get_count_container_received()

◆ nmsg_input_get_count_container_dropped()