lcm

README

LCM Go

Performant, native Go LCM implementation with integrated support for protobuf and compression.

Installation

go get go.einride.tech/lcm

Usage

Receiver

	rx, err := lcm.ListenMulticastUDP(
		ctx,
		lcm.WithReceiveInterface("eth0"),
		lcm.WithReceiveProtos(&timestamppb.Timestamp{}),
	)
	if err != nil {
		panic(err) // TODO: Handle error.
	}
	if err := rx.ReceiveProto(ctx); err != nil {
		panic(err) // TODO: Handle error.
	}
	log.Println(rx.ProtoMessage())

Transmitter

	tx, err := lcm.DialMulticastUDP(
		ctx,
		lcm.WithTransmitInterface("eth0"),
	)
	if err != nil {
		panic(err) // TODO: Handle error.
	}
	if err := tx.TransmitProto(ctx, &timestamppb.Now()); err != nil {
		panic(err) // TODO: Handle error.
	}

Notable features

Protobuf messages

Protobuf messages can be transmitted and received, with encoding and decoding handled by the LCM stack.

Compression

The library can handle compression and decompression of messages at the channel-level, with the compression scheme indicated by a query-parameter on the channel name (similar to HTTP URLs).

For example an LZ4 compressed message transmitted over a channel named google.protobuf.Timestamp?z=lz4 will be automatically decompressed.

BPF filtering

When specifying a set of channels to receive from, the library will attempt to use BPF filters to only receive messages from those channels from the kernel.

However, since there is a limit of 255 instructions on BPF filters, if there are too many channels, it will fallback and listening to everything.

Notable missing features

Fragmented messages

This library currently does not support fragmented messages.

Documentation

Overview

Package lcm provides primitives for LCM communication.

Index

• Constants
• func DefaultMulticastIP() net.IP
• type Compressor
• type Decompressor
• type Message
• type Receiver
  • func ListenMulticastUDP(ctx context.Context, receiverOpts ...ReceiverOption) (*Receiver, error)
  • func (r *Receiver) Close() error
  • func (r *Receiver) DestinationAddress() net.IP
  • func (r *Receiver) InterfaceIndex() int
  • func (r *Receiver) Message() *Message
  • func (r *Receiver) ProtoMessage() proto.Message
  • func (r *Receiver) Receive(ctx context.Context) error
  • func (r *Receiver) ReceiveProto(ctx context.Context) error
  • func (r *Receiver) SourceAddress() net.IP
• type ReceiverOption
  • func WithReceiveAddress(ip net.IP) ReceiverOption
  • func WithReceiveBPF(program []bpf.Instruction) ReceiverOption
  • func WithReceiveBatchSize(n int) ReceiverOption
  • func WithReceiveBufferSize(n int) ReceiverOption
  • func WithReceiveInterface(interfaceName string) ReceiverOption
  • func WithReceivePort(port int) ReceiverOption
  • func WithReceiveProtos(msgs ...proto.Message) ReceiverOption
• type Transmitter
  • func DialMulticastUDP(ctx context.Context, transmitterOpts ...TransmitterOption) (*Transmitter, error)
  • func (t *Transmitter) Close() error
  • func (t *Transmitter) Transmit(ctx context.Context, channel string, data []byte) error
  • func (t *Transmitter) TransmitProto(ctx context.Context, m proto.Message) error
  • func (t *Transmitter) TransmitProtoOnChannel(ctx context.Context, channel string, m proto.Message) error
• type TransmitterOption
  • func WithTransmitAddress(addr *net.UDPAddr) TransmitterOption
  • func WithTransmitCompression(compressor Compressor, channels ...string) TransmitterOption
  • func WithTransmitCompressionProto(compressor Compressor, msgs ...proto.Message) TransmitterOption
  • func WithTransmitInterface(interfaceName string) TransmitterOption
  • func WithTransmitMulticastLoopback(b bool) TransmitterOption
  • func WithTransmitTTL(ttl int) TransmitterOption

Examples

• Receiver
• Transmitter

Constants

const DefaultPort = 7667

DefaultPort is the default LCM port.

Functions

func DefaultMulticastIP

func DefaultMulticastIP() net.IP

DefaultMulticastIP returns the default LCM multicast IP.

Types

type Compressor

type Compressor interface {
	Compress(data []byte) ([]byte, error)
	Name() string
}

Compressor is an interface for an LCM message compressor.

type Decompressor

type Decompressor interface {
	Decompress(data []byte) ([]byte, error)
}

Decompressor is an interface for an LCM message decompressor.

type Message

type Message struct {
	Channel        string
	Params         string
	SequenceNumber uint32
	Data           []byte
}

Message represents an LCM message.

type Receiver

type Receiver struct {
	// contains filtered or unexported fields
}

Receiver represents an LCM Receiver instance.

Not thread-safe.

Example

package main

import (
	"context"
	"log"
	"net"

	"go.einride.tech/lcm"
	"golang.org/x/net/nettest"
	"google.golang.org/protobuf/types/known/timestamppb"
)

func main() {
	ctx := context.Background()
	ifi, err := nettest.RoutedInterface("ip4", net.FlagUp|net.FlagMulticast)
	if err != nil {
		panic(err) // TODO: Handle error.
	}
	rx, err := lcm.ListenMulticastUDP(
		ctx,
		lcm.WithReceiveInterface(ifi.Name),
		lcm.WithReceiveProtos(&timestamppb.Timestamp{}),
	)
	if err != nil {
		panic(err) // TODO: Handle error.
	}
	defer func() {
		if err := rx.Close(); err != nil {
			panic(err) // TODO: Handle error.
		}
	}()
	log.Printf("listening on: %s\n", ifi.Name)
	for {
		if err := rx.ReceiveProto(ctx); err != nil {
			panic(err) // TODO: Handle error.
		}
		log.Printf("received: %v", rx.ProtoMessage())
	}
}

func ListenMulticastUDP

func ListenMulticastUDP(ctx context.Context, receiverOpts ...ReceiverOption) (*Receiver, error)

ListenMulticastUDP returns a Receiver configured with the provided options.

func (*Receiver) Close

func (r *Receiver) Close() error

Close the receiver connection after leaving all joined multicast groups.

func (*Receiver) DestinationAddress

func (r *Receiver) DestinationAddress() net.IP

DestinationAddress returns the destination address of the last received message.

func (*Receiver) InterfaceIndex

func (r *Receiver) InterfaceIndex() int

InterfaceIndex returns the interface index of the last received message.

func (*Receiver) Message

func (r *Receiver) Message() *Message

Message returns the last received message.

func (*Receiver) ProtoMessage

func (r *Receiver) ProtoMessage() proto.Message

ProtoMessage returns the last received proto message.

func (*Receiver) Receive

func (r *Receiver) Receive(ctx context.Context) error

Receive an LCM message.

If the provided context has a deadline, it will be propagated to the underlying read operation.

func (*Receiver) ReceiveProto

func (r *Receiver) ReceiveProto(ctx context.Context) error

Receive a proto LCM message. The channel is assumed to be a fully-qualified message name.

func (*Receiver) SourceAddress

func (r *Receiver) SourceAddress() net.IP

SourceAddress returns the source address of the last received message.

type ReceiverOption

type ReceiverOption func(*receiverOptions)

ReceiverOption configures an LCM receiver.

func WithReceiveAddress

func WithReceiveAddress(ip net.IP) ReceiverOption

WithReceiveAddress a multicast group address to receive from.

Provide this option multiple times to join multiple multicast groups.

func WithReceiveBPF

func WithReceiveBPF(program []bpf.Instruction) ReceiverOption

WithReceiveBPF configures the Berkely Packet Filter to set on the receiver socket.

Ineffectual in non-Linux environments.

func WithReceiveBatchSize

func WithReceiveBatchSize(n int) ReceiverOption

WithReceiveBatchSize configures the max number of messages to receive from the kernel in a single batch.

func WithReceiveBufferSize

func WithReceiveBufferSize(n int) ReceiverOption

WithReceiveBufferSize configures the kernel read buffer size (in bytes).

func WithReceiveInterface

func WithReceiveInterface(interfaceName string) ReceiverOption

WithReceiveInterface configures the interface to receive on.

func WithReceivePort

func WithReceivePort(port int) ReceiverOption

WithReceivePort configures the port to listen on.

func WithReceiveProtos

func WithReceiveProtos(msgs ...proto.Message) ReceiverOption

type Transmitter

type Transmitter struct {
	// contains filtered or unexported fields
}

Transmitter represents an LCM Transmitter instance.

Example

package main

import (
	"context"
	"log"
	"net"
	"time"

	"go.einride.tech/lcm"
	"go.einride.tech/lcm/compression/lcmlz4"
	"golang.org/x/net/nettest"
	"google.golang.org/protobuf/types/known/timestamppb"
)

func main() {
	ctx := context.Background()
	ifi, err := nettest.RoutedInterface("ip4", net.FlagUp|net.FlagMulticast)
	if err != nil {
		panic(err) // TODO: Handle error.
	}
	tx, err := lcm.DialMulticastUDP(
		ctx,
		lcm.WithTransmitInterface(ifi.Name),
		lcm.WithTransmitCompressionProto(lcmlz4.NewCompressor(), &timestamppb.Timestamp{}),
	)
	if err != nil {
		panic(err) // TODO: Handle error.
	}
	defer func() {
		if err := tx.Close(); err != nil {
			panic(err) // TODO: Handle error.
		}
	}()
	ticker := time.NewTicker(1000 * time.Millisecond)
	log.Printf("transmitting on: %s\n", ifi.Name)
	for range ticker.C {
		if err := tx.TransmitProto(ctx, timestamppb.Now()); err != nil {
			panic(err) // TODO: Handle error.
		}
	}
}

func DialMulticastUDP

func DialMulticastUDP(ctx context.Context, transmitterOpts ...TransmitterOption) (*Transmitter, error)

DialMulticastUDP returns a Transmitter configured with the provided options.

func (*Transmitter) Close

func (t *Transmitter) Close() error

Close the transmitter connection.

func (*Transmitter) Transmit

func (t *Transmitter) Transmit(ctx context.Context, channel string, data []byte) error

Transmit a raw payload.

If the provided context has a deadline, it will be propagated to the underlying write operation.

func (*Transmitter) TransmitProto

func (t *Transmitter) TransmitProto(ctx context.Context, m proto.Message) error

TransmitProto transmits a protobuf message on the channel given by the message's fully-qualified name.

func (*Transmitter) TransmitProtoOnChannel

func (t *Transmitter) TransmitProtoOnChannel(ctx context.Context, channel string, m proto.Message) error

TransmitProto transmits a protobuf message.

type TransmitterOption

type TransmitterOption func(*transmitterOptions)

TransmitterOption configures an LCM transmitter.

func WithTransmitAddress

func WithTransmitAddress(addr *net.UDPAddr) TransmitterOption

WithTransmitAddress configures an address to transmit to.

Provide this option multiple times to transmit to multiple addresses.

func WithTransmitCompression

func WithTransmitCompression(compressor Compressor, channels ...string) TransmitterOption

WithTransmitCompression configures compressor for channels.

func WithTransmitCompressionProto

func WithTransmitCompressionProto(compressor Compressor, msgs ...proto.Message) TransmitterOption

WithTransmitCompressionProto configures compressor for protos.

func WithTransmitInterface

func WithTransmitInterface(interfaceName string) TransmitterOption

WithTransmitInterface configures the interface to transmit on.

func WithTransmitMulticastLoopback

func WithTransmitMulticastLoopback(b bool) TransmitterOption

WithTransmitMulticastLoopback configures multicast loopback on the transmitter socket.

func WithTransmitTTL

func WithTransmitTTL(ttl int) TransmitterOption

WithTransmitTTL configures the multicast TTL on the transmitter socket.