proio

README

proio for Go

API

API documentation is provided by godoc.org

Installation

go-proio and included command-line tools are go get-able. Make sure you have the go compiler installed and set up:

go get github.com/proio-org/go-proio/...

If you do not have the go compiler, you can find pre-compiled binaries for the tools in the releases.

For information on what versions of Go are supported, please see the Travis CI page.

Examples

• Print
• Scan
• Skip
• Push, get, inspect

Documentation

Overview

Example (Print)

event := proio.NewEvent()

parentPDG := int32(443)
parent := &eic.Particle{Pdg: &parentPDG}
parentID := event.AddEntry("Particle", parent)
event.TagEntry(parentID, "MC", "Primary")

child1PDG := int32(11)
child1 := &eic.Particle{Pdg: &child1PDG}
child2PDG := int32(-11)
child2 := &eic.Particle{Pdg: &child2PDG}
childIDs := event.AddEntries("Particle", child1, child2)
for _, id := range childIDs {
	event.TagEntry(id, "MC", "GenStable")
}

parent.Child = append(parent.Child, childIDs...)
child1.Parent = append(child1.Parent, parentID)
child2.Parent = append(child2.Parent, parentID)

fmt.Print(event)

Example (PushGetInspect)

buffer := &bytes.Buffer{}
writer := proio.NewWriter(buffer)

eventOut := proio.NewEvent()

// Create entries and hold onto their IDs for referencing

parentPDG := int32(443)
parent := &eic.Particle{Pdg: &parentPDG}
parentID := eventOut.AddEntry("Particle", parent)
eventOut.TagEntry(parentID, "MC", "Primary")

child1PDG := int32(11)
child1 := &eic.Particle{Pdg: &child1PDG}
child2PDG := int32(-11)
child2 := &eic.Particle{Pdg: &child2PDG}
childIDs := eventOut.AddEntries("Particle", child1, child2)
for _, id := range childIDs {
	eventOut.TagEntry(id, "MC", "GenStable")
}

parent.Child = append(parent.Child, childIDs...)
child1.Parent = append(child1.Parent, parentID)
child2.Parent = append(child2.Parent, parentID)

writer.Push(eventOut)

writer.Flush()

// Event created and serialized, now to deserialize and inspect

reader := proio.NewReader(buffer)
eventIn, _ := reader.Next()

mcParts := eventIn.TaggedEntries("Primary")
fmt.Print(len(mcParts), " Primary particle(s)...\n")
for i, parentID := range mcParts {
	part := eventIn.GetEntry(parentID).(*eic.Particle)
	fmt.Print(i, ". PDG: ", part.GetPdg(), "\n")
	fmt.Print("  ", len(part.Child), " children...\n")
	for j, childID := range part.Child {
		fmt.Print("  ", j, ". PDG: ", eventIn.GetEntry(childID).(*eic.Particle).GetPdg(), "\n")
	}
}

Output:

1 Primary particle(s)...
0. PDG: 443
  2 children...
  0. PDG: 11
  1. PDG: -11

Example (Scan)

buffer := &bytes.Buffer{}
writer := proio.NewWriter(buffer)

pdg := int32(443)
for i := 0; i < 5; i++ {
	event := proio.NewEvent()
	charge := float32(i + 1)
	p := &eic.Particle{
		Pdg:    &pdg,
		Charge: &charge,
	}
	event.AddEntry("Particle", p)
	writer.Push(event)
}
writer.Flush()

reader := proio.NewReader(buffer)

for event := range reader.ScanEvents() {
	fmt.Print(event)
}

Output:

---------- TAG: Particle ----------
ID: 1
Entry type: proio.model.eic.Particle
pdg: 443
charge: 1

---------- TAG: Particle ----------
ID: 1
Entry type: proio.model.eic.Particle
pdg: 443
charge: 2

---------- TAG: Particle ----------
ID: 1
Entry type: proio.model.eic.Particle
pdg: 443
charge: 3

---------- TAG: Particle ----------
ID: 1
Entry type: proio.model.eic.Particle
pdg: 443
charge: 4

---------- TAG: Particle ----------
ID: 1
Entry type: proio.model.eic.Particle
pdg: 443
charge: 5

Example (Skip)

buffer := &bytes.Buffer{}
writer := proio.NewWriter(buffer)

pdg := int32(443)
for i := 0; i < 5; i++ {
	event := proio.NewEvent()
	charge := float32(i + 1)
	p := &eic.Particle{
		Pdg:    &pdg,
		Charge: &charge,
	}
	event.AddEntry("Particle", p)
	writer.Push(event)
}
writer.Flush()

bytesReader := bytes.NewReader(buffer.Bytes())
reader := proio.NewReader(bytesReader)

reader.Skip(3)
event, _ := reader.Next()
fmt.Print(event)
reader.SeekToStart()
event, _ = reader.Next()
fmt.Print(event)

Output:

---------- TAG: Particle ----------
ID: 1
Entry type: proio.model.eic.Particle
pdg: 443
charge: 4

---------- TAG: Particle ----------
ID: 1
Entry type: proio.model.eic.Particle
pdg: 443
charge: 1

Index

• type Compression
• type Event
  • func NewEvent() *Event
  • func (evt *Event) AddEntries(tag string, entries ...protobuf.Message) []uint64
  • func (evt *Event) AddEntry(tag string, entry protobuf.Message) uint64
  • func (evt *Event) AllEntries() []uint64
  • func (evt *Event) DeleteTag(tag string)
  • func (evt *Event) EntryTags(id uint64) []string
  • func (evt *Event) GetEntry(id uint64) protobuf.Message
  • func (evt *Event) RemoveEntry(id uint64)
  • func (evt *Event) String() string
  • func (evt *Event) TagEntry(id uint64, tags ...string)
  • func (evt *Event) TaggedEntries(tag string) []uint64
  • func (evt *Event) Tags() []string
  • func (evt *Event) UntagEntry(id uint64, tag string)
• type Reader
  • func NewReader(streamReader io.Reader) *Reader
  • func Open(filename string) (*Reader, error)
  • func (rdr *Reader) Close()
  • func (rdr *Reader) Next() (*Event, error)
  • func (rdr *Reader) NextHeader() (*proto.BucketHeader, error)
  • func (rdr *Reader) ScanEvents() <-chan *Event
  • func (rdr *Reader) SeekToStart() error
  • func (rdr *Reader) Skip(nEvents int) (nSkipped int, err error)
  • func (rdr *Reader) StopScan()
• type Writer
  • func Create(filename string) (*Writer, error)
  • func NewWriter(streamWriter io.Writer) *Writer
  • func (wrt *Writer) Close() error
  • func (wrt *Writer) Flush() error
  • func (wrt *Writer) Push(event *Event) error
  • func (wrt *Writer) PushMetadata(name string, data []byte) error
  • func (wrt *Writer) SetCompression(comp Compression) error

Examples

• Package (Print)
• Package (PushGetInspect)
• Package (Scan)
• Package (Skip)

Types

type Compression

type Compression int

const (
	UNCOMPRESSED Compression = iota
	GZIP
	LZ4
	LZMA
)

type Event

type Event struct {
	Err      error
	Metadata map[string][]byte
	// contains filtered or unexported fields
}

Event contains all data for an event, and provides methods for adding and retrieving data.

func NewEvent

func NewEvent() *Event

NewEvent is required for constructing an Event.

func (*Event) AddEntries

func (evt *Event) AddEntries(tag string, entries ...protobuf.Message) []uint64

AddEntries is like AddEntry, except that it is variadic, taking an arbitrary number of entries separated by commas. Additionally, the return value is a slice of IDs.

func (*Event) AddEntry

func (evt *Event) AddEntry(tag string, entry protobuf.Message) uint64

AddEntry takes a single primary tag for an entry and an entry protobuf message, and returns a new ID number for the entry. This ID number can be used to persistently reference the entry. For example, pass the ID TagEntry to add additional tags to the entry.

func (*Event) AllEntries

func (evt *Event) AllEntries() []uint64

AllEntries returns a slice of identifiers for all entries contained in the Event.

func (*Event) DeleteTag

func (evt *Event) DeleteTag(tag string)

DeleteTag takes a tag name as an argument and deletes that tag from the Event if it exists.

func (*Event) EntryTags

func (evt *Event) EntryTags(id uint64) []string

EntryTags does a reverse lookup of tags that point to a given entry ID.

func (*Event) GetEntry

func (evt *Event) GetEntry(id uint64) protobuf.Message

GetEntry retrieves and deserializes an entry corresponding to the given ID number. The deserialized entry is returned. The entry type must be one that has been linked (and therefore initialized) with the current executable, otherwise it is an unknown type and nil is returned.

func (*Event) RemoveEntry

func (evt *Event) RemoveEntry(id uint64)

RemoveEntry takes an entry id and removes the referenced entry from the Event.

func (*Event) String

func (evt *Event) String() string

func (*Event) TagEntry

func (evt *Event) TagEntry(id uint64, tags ...string)

TagEntry adds additional tags to an entry ID returned by AddEntry.

func (*Event) TaggedEntries

func (evt *Event) TaggedEntries(tag string) []uint64

TaggedEntries returns a slice of ID numbers that are referenced by the given tag.

func (*Event) Tags

func (evt *Event) Tags() []string

Tags returns a list of all tags in the Event.

func (*Event) UntagEntry

func (evt *Event) UntagEntry(id uint64, tag string)

UntagEntry removes the association between a tag and an entry.

type Reader

type Reader struct {
	Err            chan error
	EvtScanBufSize int

	sync.Mutex
	// contains filtered or unexported fields
}

Reader serves to read Events from a stream in the proio format. The Reader is not inherently thread safe, but it conveniently embeds sync.Mutex so that it can be locked and unlocked.

func NewReader

func NewReader(streamReader io.Reader) *Reader

NewReader wraps an existing io.Reader for reading proio Events. Either Open or NewReader should be called to construct a new Reader.

func Open

func Open(filename string) (*Reader, error)

Open opens the given existing file (in read-only mode), returning an error where appropriate. Upon success, a new Reader is created to wrap the file, and returned. Either Open or NewReader should be called to construct a new Reader.

func (*Reader) Close

func (rdr *Reader) Close()

Close closes any file that was opened by the library, and stops any unfinished scans. Close does not close io.Readers passed directly to NewReader.

func (*Reader) Next

func (rdr *Reader) Next() (*Event, error)

Next retrieves the next event from the stream.

func (*Reader) NextHeader

func (rdr *Reader) NextHeader() (*proto.BucketHeader, error)

NextHeader returns the next bucket header from the stream, and discards the bucket payload.

func (*Reader) ScanEvents

func (rdr *Reader) ScanEvents() <-chan *Event

ScanEvents returns a buffered channel of type Event where all of the events in the stream will be pushed. The channel buffer size is defined by Reader.EvtScanBufSize which defaults to 100. The goroutine responsible for fetching events will not break until there are no more events, Reader.StopScan() is called, or Reader.Close() is called. In this scenario, errors are pushed to the Reader.Err channel.

func (*Reader) SeekToStart

func (rdr *Reader) SeekToStart() error

SeekToStart seeks seekable streams to the beginning, and prepares the stream to read from there.

func (*Reader) Skip

func (rdr *Reader) Skip(nEvents int) (nSkipped int, err error)

Skip skips nEvents events. If the return error is nil, nEvents have been skipped.

func (*Reader) StopScan

func (rdr *Reader) StopScan()

StopScan stops all scans initiated by Reader.ScanEvents().

type Writer

type Writer struct {
	sync.Mutex
	// contains filtered or unexported fields
}

Writer serves to write Events into a stream in the proio format. The Writer is not inherently thread safe, but it conveniently embeds sync.Mutex so that it can be locked and unlocked.

func Create

func Create(filename string) (*Writer, error)

Create makes a new file specified by filename, overwriting any existing file, and returns a Writer for the file. Either NewWriter or Create must be used to construct a Writer.

func NewWriter

func NewWriter(streamWriter io.Writer) *Writer

NewWriter takes an io.Writer and wraps it in a new proio Writer. Either NewWriter or Create must be used to construct a Writer.

func (*Writer) Close

func (wrt *Writer) Close() error

Close calls Flush and closes any file that was created by the library. Close does not close io.Writers passed directly to NewWriter.

func (*Writer) Flush

func (wrt *Writer) Flush() error

Flush flushes any of the Writer's bucket contents.

func (*Writer) Push

func (wrt *Writer) Push(event *Event) error

Serialize the given Event. Once this is performed, changes to the Event in memory are not reflected in the output stream.

func (*Writer) PushMetadata

func (wrt *Writer) PushMetadata(name string, data []byte) error

func (*Writer) SetCompression

func (wrt *Writer) SetCompression(comp Compression) error

Set compression type, for example to GZIP or UNCOMPRESSED. This can be called even after writing some events.