mcap

package module
Version: v0.0.0-...-926774a Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: May 24, 2022 License: Apache-2.0 Imports: 14 Imported by: 2

README

mcap

Note: This library is experimental and will change without warning until finalization of the spec.

An experimental library for writing and reading MCAP files in go.

API Reference

API documentation can be found here.

Documentation

Index

Constants

This section is empty.

Variables

View Source
var ErrBadMagic = errors.New("not an mcap file")

ErrBadMagic indicates the lexer has detected invalid magic bytes.

View Source
var ErrChunkTooLarge = errors.New("chunk exceeds configured maximum size")
View Source
var ErrLengthOutOfRange = errors.New("length out of int32 range")
View Source
var ErrNestedChunk = errors.New("detected nested chunk")

ErrNestedChunk indicates the lexer has detected a nested chunk.

View Source
var ErrRecordTooLarge = errors.New("record exceeds configured maximum size")
View Source
var ErrUnknownSchema = errors.New("unknown schema")

ErrUnknownSchema is returned when a schema ID is not known to the writer.

View Source
var Magic = []byte{0x89, 'M', 'C', 'A', 'P', 0x30, '\r', '\n'}

Magic is the magic number for an MCAP file.

Functions

func Version

func Version() string

Types

type Attachment

type Attachment struct {
	LogTime     uint64
	CreateTime  uint64
	Name        string
	ContentType string
	Data        []byte
	CRC         uint32
}

Attachment records contain auxiliary artifacts such as text, core dumps, calibration data, or other arbitrary data. Attachment records must not appear within a chunk.

func ParseAttachment

func ParseAttachment(buf []byte) (*Attachment, error)

ParseAttachment parses an attachment record.

type AttachmentIndex

type AttachmentIndex struct {
	Offset      uint64
	Length      uint64
	LogTime     uint64
	CreateTime  uint64
	DataSize    uint64
	Name        string
	ContentType string
}

AttachmentIndex records contain the location of attachments in the file. An AttachmentIndex record exists for every Attachment in the file.

func ParseAttachmentIndex

func ParseAttachmentIndex(buf []byte) (*AttachmentIndex, error)

ParseAttachmentIndex parses an attachment index record.

type Channel

type Channel struct {
	ID              uint16
	SchemaID        uint16
	Topic           string
	MessageEncoding string
	Metadata        map[string]string
}

Channel records define encoded streams of messages on topics. Channel records are uniquely identified within a file by their channel ID. A Channel record must occur at least once in the file prior to any message referring to its channel ID. Any two channel records sharing a common ID must be identical.

func ParseChannel

func ParseChannel(buf []byte) (*Channel, error)

ParseChannel parses a channel record.

type Chunk

type Chunk struct {
	MessageStartTime uint64
	MessageEndTime   uint64
	UncompressedSize uint64
	UncompressedCRC  uint32
	Compression      string
	Records          []byte
}

Chunk records each contain a batch of Schema, Channel, and Message records. The batch of records contained in a chunk may be compressed or uncompressed. All messages in the chunk must reference channels recorded earlier in the file (in a previous chunk or earlier in the current chunk).

func ParseChunk

func ParseChunk(buf []byte) (*Chunk, error)

ParseChunk parses a chunk record.

type ChunkIndex

type ChunkIndex struct {
	MessageStartTime    uint64
	MessageEndTime      uint64
	ChunkStartOffset    uint64
	ChunkLength         uint64
	MessageIndexOffsets map[uint16]uint64
	MessageIndexLength  uint64
	Compression         CompressionFormat
	CompressedSize      uint64
	UncompressedSize    uint64
}

ChunkIndex records contain the location of a Chunk record and its associated MessageIndex records. A ChunkIndex record exists for every Chunk in the file.

func ParseChunkIndex

func ParseChunkIndex(buf []byte) (*ChunkIndex, error)

ParseChunkIndex parses a chunk index record.

type CompressionFormat

type CompressionFormat string

CompressionFormat represents a supported chunk compression format.

const (
	// CompressionZSTD represents zstd compression.
	CompressionZSTD CompressionFormat = "zstd"
	// CompressionLZ4 represents lz4 compression.
	CompressionLZ4 CompressionFormat = "lz4"
	// CompressionNone represents no compression.
	CompressionNone CompressionFormat = ""
)

func (CompressionFormat) String

func (c CompressionFormat) String() string

String converts a compression format to a string for display.

type DataEnd

type DataEnd struct {
	DataSectionCRC uint32
}

DataEnd indicates the end of the data section.

func ParseDataEnd

func ParseDataEnd(buf []byte) (*DataEnd, error)

ParseDataEnd parses a data end record.

type Footer struct {
	SummaryStart       uint64
	SummaryOffsetStart uint64
	SummaryCRC         uint32
}

Footer records contain end-of-file information. MCAP files must end with a Footer record.

func ParseFooter

func ParseFooter(buf []byte) (*Footer, error)

ParseFooter parses a footer record.

type Header struct {
	Profile string
	Library string
}

Header is the first record in an MCAP file.

func ParseHeader

func ParseHeader(buf []byte) (*Header, error)

ParseHeader parses a header record.

type Info

type Info struct {
	Statistics        *Statistics
	Channels          map[uint16]*Channel
	Schemas           map[uint16]*Schema
	ChunkIndexes      []*ChunkIndex
	AttachmentIndexes []*AttachmentIndex
	Header            *Header
}

Info represents the result of an "info" operation, for gathering information from the summary section of a file.

func (*Info) ChannelCounts

func (i *Info) ChannelCounts() map[string]uint64

ChannelCounts counts the number of messages on each channel in an Info.

type Lexer

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

Lexer is a low-level reader for mcap files that emits tokenized byte strings without parsing or interpreting them, except in the case of chunks, which may be optionally de-chunked.

func NewLexer

func NewLexer(r io.Reader, opts ...*LexerOptions) (*Lexer, error)

NewLexer returns a new lexer for the given reader.

func (*Lexer) Next

func (l *Lexer) Next(p []byte) (TokenType, []byte, error)

Next returns the next token from the lexer as a byte array. The result will be sliced out of the provided buffer `p`, if p has adequate space. If p does not have adequate space, a new buffer with sufficient size is allocated for the result.

type LexerOptions

type LexerOptions struct {
	// SkipMagic instructs the lexer not to perform validation of the leading magic bytes.
	SkipMagic bool
	// ValidateCRC instructs the lexer to validate CRC checksums for chunks.
	ValidateCRC bool
	// EmitChunks instructs the lexer to emit chunk records without de-chunking.
	// It is incompatible with ValidateCRC.
	EmitChunks bool
	// MaxDecompressedChunkSize defines the maximum size chunk the lexer will
	// decompress. Chunks larger than this will result in an error.
	MaxDecompressedChunkSize int
	// MaxRecordSize defines the maximum size record the lexer will read.
	// Records larger than this will result in an error.
	MaxRecordSize int
}

LexerOptions holds options for the lexer.

type Message

type Message struct {
	ChannelID   uint16
	Sequence    uint32
	LogTime     uint64
	PublishTime uint64
	Data        []byte
}

Message records encode a single timestamped message on a channel. The message encoding and schema must match that of the Channel record corresponding to the message's channel ID.

func ParseMessage

func ParseMessage(buf []byte) (*Message, error)

ParseMessage parses a message record.

type MessageIndex

type MessageIndex struct {
	ChannelID uint16
	Records   []MessageIndexEntry
	// contains filtered or unexported fields
}

MessageIndex records allow readers to locate individual records within a chunk by timestamp. A sequence of Message Index records occurs immediately after each chunk. Exactly one Message Index record must exist in the sequence for every channel on which a message occurs inside the chunk.

func ParseMessageIndex

func ParseMessageIndex(buf []byte) (*MessageIndex, error)

ParseMessageIndex parses a message index record.

func (*MessageIndex) Add

func (idx *MessageIndex) Add(timestamp uint64, offset uint64)

Add an entry to the message index.

func (*MessageIndex) Entries

func (idx *MessageIndex) Entries() []MessageIndexEntry

Entries lists the entries in the message index.

func (*MessageIndex) Insort

func (idx *MessageIndex) Insort()

Insort sorts the records of a MessageIndex record by timestamp, using an insertion sort. This can be advantageous as MessageIndex records are often nearly or fully-sorted already.

func (*MessageIndex) Reset

func (idx *MessageIndex) Reset()

Reset resets the MessageIndex to an empty state, to enable reuse.

type MessageIndexEntry

type MessageIndexEntry struct {
	Timestamp uint64
	Offset    uint64
}

type MessageIterator

type MessageIterator interface {
	Next([]byte) (*Schema, *Channel, *Message, error)
}

type Metadata

type Metadata struct {
	Name     string
	Metadata map[string]string
}

Metadata records contain arbitrary user data in key-value pairs.

func ParseMetadata

func ParseMetadata(buf []byte) (*Metadata, error)

ParseMetadata parses a metadata record.

type MetadataIndex

type MetadataIndex struct {
	Offset uint64
	Length uint64
	Name   string
}

MetadataIndex records each contain the location of a metadata record within the file.

func ParseMetadataIndex

func ParseMetadataIndex(buf []byte) (*MetadataIndex, error)

ParseMetadataIndex parses a metadata index record.

type OpCode

type OpCode byte
const (
	OpReserved        OpCode = 0x00
	OpHeader          OpCode = 0x01
	OpFooter          OpCode = 0x02
	OpSchema          OpCode = 0x03
	OpChannel         OpCode = 0x04
	OpMessage         OpCode = 0x05
	OpChunk           OpCode = 0x06
	OpMessageIndex    OpCode = 0x07
	OpChunkIndex      OpCode = 0x08
	OpAttachment      OpCode = 0x09
	OpAttachmentIndex OpCode = 0x0A
	OpStatistics      OpCode = 0x0B
	OpMetadata        OpCode = 0x0C
	OpMetadataIndex   OpCode = 0x0D
	OpSummaryOffset   OpCode = 0x0E
	OpDataEnd         OpCode = 0x0F
)

func (OpCode) String

func (c OpCode) String() string

type Reader

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

func NewReader

func NewReader(r io.Reader) (*Reader, error)

func (*Reader) Info

func (r *Reader) Info() (*Info, error)

func (*Reader) Messages

func (r *Reader) Messages(
	start int64,
	end int64,
	topics []string,
	useIndex bool,
) (MessageIterator, error)

type Schema

type Schema struct {
	ID       uint16
	Name     string
	Encoding string
	Data     []byte
}

A Schema record defines an individual schema. Schema records are uniquely identified within a file by their schema ID. A Schema record must occur at least once in the file prior to any Channel referring to its ID. Any two schema records sharing a common ID must be identical.

func ParseSchema

func ParseSchema(buf []byte) (*Schema, error)

ParseSchema parses a schema record.

type Statistics

type Statistics struct {
	MessageCount         uint64
	SchemaCount          uint16
	ChannelCount         uint32
	AttachmentCount      uint32
	MetadataCount        uint32
	ChunkCount           uint32
	MessageStartTime     uint64
	MessageEndTime       uint64
	ChannelMessageCounts map[uint16]uint64
}

Statistics records contain summary information about recorded data. The statistics record is optional, but the file should contain at most one.

func ParseStatistics

func ParseStatistics(buf []byte) (*Statistics, error)

ParseStatistics parses a statistics record.

type SummaryOffset

type SummaryOffset struct {
	GroupOpcode OpCode
	GroupStart  uint64
	GroupLength uint64
}

SummaryOffset records contain the location of records within the summary section. Each SummaryOffset record corresponds to a group of summary records with a common opcode.

func ParseSummaryOffset

func ParseSummaryOffset(buf []byte) (*SummaryOffset, error)

ParseSummaryOffset parses a summary offset record.

type TokenType

type TokenType int

TokenType encodes a type of token from the lexer.

const (
	// TokenHeader represents a header token.
	TokenHeader TokenType = iota
	// TokenFooter represents a footer token.
	TokenFooter
	// TokenSchema represents a schema token.
	TokenSchema
	// TokenChannel represents a channel token.
	TokenChannel
	// TokenMessage represents a message token.
	TokenMessage
	// TokenChunk represents a chunk token.
	TokenChunk
	// TokenMessageIndex represents a message index token.
	TokenMessageIndex
	// TokenChunkIndex represents a chunk index token.
	TokenChunkIndex
	// TokenAttachment represents an attachment token.
	TokenAttachment
	// TokenAttachmentIndex represents an attachment index token.
	TokenAttachmentIndex
	// TokenStatistics represents a statistics token.
	TokenStatistics
	// TokenMetadata represents a metadata token.
	TokenMetadata
	// TokenSummaryOffset represents a summary offset token.
	TokenMetadataIndex
	// TokenDataEnd represents a data end token.
	TokenSummaryOffset
	// 	TokenError represents an error token.
	TokenDataEnd
	// TokenError represents an error token.
	TokenError
)

func (TokenType) String

func (t TokenType) String() string

String converts a token type to its string representation.

type Writer

type Writer struct {
	// Statistics collected over the course of the recording.
	Statistics *Statistics
	// ChunkIndexes created over the course of the recording.
	ChunkIndexes []*ChunkIndex
	// AttachmentIndexes created over the course of the recording.
	AttachmentIndexes []*AttachmentIndex
	// MetadataIndexes created over the course of the recording.
	MetadataIndexes []*MetadataIndex
	// contains filtered or unexported fields
}

Writer is a writer for the MCAP format.

func NewWriter

func NewWriter(w io.Writer, opts *WriterOptions) (*Writer, error)

NewWriter returns a new MCAP writer.

func (*Writer) Close

func (w *Writer) Close() error

Close the writer by closing the active chunk and writing the summary section.

func (*Writer) Offset

func (w *Writer) Offset() uint64

Offset returns the current offset of the writer, or the size of the written file if called after Close().

func (*Writer) WriteAttachment

func (w *Writer) WriteAttachment(a *Attachment) error

WriteAttachment writes an attachment to the output. Attachment records contain auxiliary artifacts such as text, core dumps, calibration data, or other arbitrary data. Attachment records must not appear within a chunk.

func (*Writer) WriteAttachmentIndex

func (w *Writer) WriteAttachmentIndex(idx *AttachmentIndex) error

WriteAttachmentIndex writes an attachment index record to the output. An Attachment Index record contains the location of an attachment in the file. An Attachment Index record exists for every Attachment record in the file.

func (*Writer) WriteChannel

func (w *Writer) WriteChannel(c *Channel) error

WriteChannel writes a channel info record to the output. Channel Info records are uniquely identified within a file by their channel ID. A Channel Info record must occur at least once in the file prior to any message referring to its channel ID.

func (*Writer) WriteDataEnd

func (w *Writer) WriteDataEnd(e *DataEnd) error

WriteDataEnd writes a data end record to the output. A Data End record indicates the end of the data section.

func (*Writer) WriteFooter

func (w *Writer) WriteFooter(f *Footer) error

WriteFooter writes a footer record to the output. A Footer record contains end-of-file information. It must be the last record in the file. Readers using the index to read the file will begin with by reading the footer and trailing magic.

If opts.IncludeCRC is enabled, the CRC is expected to have been reset after the DataEnd record was written.

func (*Writer) WriteHeader

func (w *Writer) WriteHeader(header *Header) error

WriteHeader writes a header record to the output.

func (*Writer) WriteMessage

func (w *Writer) WriteMessage(m *Message) error

WriteMessage writes a message to the output. A message record encodes a single timestamped message on a channel. The message encoding and schema must match that of the channel info record corresponding to the message's channel ID.

func (*Writer) WriteMessageIndex

func (w *Writer) WriteMessageIndex(idx *MessageIndex) error

WriteMessageIndex writes a message index record to the output. A Message Index record allows readers to locate individual message records within a chunk by their timestamp. A sequence of Message Index records occurs immediately after each chunk. Exactly one Message Index record must exist in the sequence for every channel on which a message occurs inside the chunk.

func (*Writer) WriteMetadata

func (w *Writer) WriteMetadata(m *Metadata) error

WriteMetadata writes a metadata record to the output. A metadata record contains arbitrary user data in key-value pairs.

func (*Writer) WriteMetadataIndex

func (w *Writer) WriteMetadataIndex(idx *MetadataIndex) error

WriteMetadataIndex writes a metadata index record to the output.

func (*Writer) WriteSchema

func (w *Writer) WriteSchema(s *Schema) (err error)

WriteSchema writes a schema record to the output. Schema records are uniquely identified within a file by their schema ID. A Schema record must occur at least once in the file prior to any Channel Info referring to its ID.

func (*Writer) WriteStatistics

func (w *Writer) WriteStatistics(s *Statistics) error

WriteStatistics writes a statistics record to the output. A Statistics record contains summary information about the recorded data. The statistics record is optional, but the file should contain at most one.

func (*Writer) WriteSummaryOffset

func (w *Writer) WriteSummaryOffset(s *SummaryOffset) error

WriteSummaryOffset writes a summary offset record to the output. A Summary Offset record contains the location of records within the summary section. Each Summary Offset record corresponds to a group of summary records with the same opcode.

type WriterOptions

type WriterOptions struct {
	// IncludeCRC specifies whether to compute CRC checksums in the output.
	IncludeCRC bool
	// Chunked specifies whether the file should be chunk-compressed.
	Chunked bool
	// ChunkSize specifies a target chunk size for compressed chunks. This size
	// may be exceeded, for instance in the case of oversized messages.
	ChunkSize int64
	// Compression indicates the compression format to use for chunk compression.
	Compression CompressionFormat

	// SkipMessageIndexing skips the message and chunk indexes for a chunked
	// file.
	SkipMessageIndexing bool

	// SkipStatistics skips the statistics accounting.
	SkipStatistics bool

	// SkipRepeatedSchemas skips the schemas repeated at the end of the file
	SkipRepeatedSchemas bool

	// SkipRepeatedChannelInfos skips the channel infos repeated at the end of
	// the file
	SkipRepeatedChannelInfos bool

	// SkipAttachmentIndex skips indexing for attachments
	SkipAttachmentIndex bool

	// SkipMetadataIndex skips metadata index records.
	SkipMetadataIndex bool

	// SkipChunkIndex skips chunk index records.
	SkipChunkIndex bool

	// SkipSummaryOffsets skips summary offset records.
	SkipSummaryOffsets bool

	// OverrideLibrary causes the default header library to be overridden, not
	// appended to.
	OverrideLibrary bool
}

WriterOptions are options for the MCAP Writer.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL