demux

Documentation

Overview

Package demux implements low-level demultiplexing of the stream of messages sent from OpenVPN on the management channel.

OpenVPN's protocol includes two different kinds of message from the OpenVPN process: replies to commands sent by the management client, and asynchronous event notifications.

This package's purpose is to split these messages into two separate streams, so that functions executing command/response sequences can just block on the reply channel while an event loop elsewhere deals with any async events that might show up.

Index

• func Demultiplex(r io.Reader, replyCh, eventCh chan<- []byte)

Examples

• Demultiplex

Functions

func Demultiplex

func Demultiplex(r io.Reader, replyCh, eventCh chan<- []byte)

Demultiplex reads from the given io.Reader, assumed to be the client end of an OpenVPN Management Protocol connection, and splits it into distinct messages from OpenVPN.

It then writes the raw message buffers into either replyCh or eventCh depending on whether each message is a reply to a client command or an asynchronous event notification.

The buffers written to replyCh are entire raw message lines (without the trailing newlines), while the buffers written to eventCh are the raw event strings with the prototcol's leading '>' indicator omitted.

The caller should usually provide buffered channels of sufficient buffer depth so that the reply channel will not be starved by slow event processing.

Once the io.Reader signals EOF, eventCh will be closed, then replyCh will be closed, and then this function will return.

As a special case, if a non-EOF error occurs while reading from the io.Reader then a synthetic "FATAL" event will be written to eventCh before the two buffers are closed and the function returns. This synthetic message will have the error message "Error reading from OpenVPN".

Example

Somewhat-contrived example of blocking for a reply while concurrently processing asynchronous events.

// In a real caller we would have a net.IPConn as our reader,
// but we'll use a bytes reader here as a test.
r := bytes.NewReader([]byte(
	// A reply to a hypothetical command interspersed between
	// two asynchronous events.
	">HOLD:Waiting for hold release\nSUCCESS: foo\n>FATAL:baz\n",
))

// No strong need for buffering on this channel because usually
// a message sender will immediately block waiting for the
// associated response message.
replyCh := make(chan []byte)

// Make sure the event channel buffer is deep enough that slow event
// processing won't significantly delay synchronous replies. If you
// process events quickly, or if you aren't sending any commands
// concurrently with acting on events, then this is not so important.
eventCh := make(chan []byte, 10)

// Start demultiplexing the message stream in the background.
// This goroutine will exit once the reader signals EOF.
go Demultiplex(r, replyCh, eventCh)

// Some coroutine has sent a hypothetical message to OpenVPN,
// and it can directly block until the associated reply arrives.
// The events will be concurrently handled by our event loop
// below while we wait for the reply to show up.
go func() {
	replyMsgBuf := <-replyCh
	fmt.Printf("Command reply: %s\n", string(replyMsgBuf))
}()

// Main event loop deals with the async events as they arrive,
// independently of any commands that are pending.
for msgBuf := range eventCh {
	fmt.Printf("Event: %s\n", string(msgBuf))
}