Skip to content

types

import "github.com/frgrisk/vor-stream/queue/types"

Package types contains functions for converting to VOR Stream data types. This package is recommended for use in node development to ensure consistent and reliable data type conversions. The package includes a generic Cast() function that will attempt to convert any Go type to a VOR Stream data type. It also includes functions for returning typed values from any type so that later type assertion is not required.

Index

Variables

Enum value maps for DataType.

var (
    DataType_name = map[int32]string{
        0:  "UNSPECIFIED",
        1:  "ARRAY",
        2:  "BOOL",
        3:  "CHAR",
        4:  "DATE",
        5:  "DATETIME",
        6:  "INT",
        7:  "NUM",
    }
    DataType_value = map[string]int32{
        "UNSPECIFIED": 0,
        "ARRAY":       1,
        "BOOL":        2,
        "CHAR":        3,
        "DATE":        4,
        "DATETIME":    5,
        "INT":         6,
        "NUM":         7,
    }
)

Extension fields to descriptorpb.FieldOptions.

var (
    // optional vor_stream.queue.types.DataType vor_stream_type = 60001;
    E_VorStreamType = &file_vor_stream_queue_types_datatype_proto_extTypes[0]
)


var File_vor_stream_queue_types_datatype_proto protoreflect.FileDescriptor

func Cast

func Cast(from any, toType DataType) (any, error)

Cast converts the input value (from) to the specified data type (toType). It returns the converted value and an error. If the conversion is successful, the error is nil. If the conversion fails, the error is not nil and the converted value is the zero value of the specified data type. See type-specific conversion functions for more information on how specific data type conversions are handled.

func CastDateToProtoDate

func CastDateToProtoDate(d date.Date) *datepb.Date

CastDateToProtoDate converts a Stream date to a Protobuf date.

func CastDatetimeToProtoTimestamp

func CastDatetimeToProtoTimestamp(dt time.Time) *timestamppb.Timestamp

CastDatetimeToProtoTimestamp converts a Stream datetime to a Protobuf timestamp.

func CastToArray

func CastToArray(from any) ([]float64, error)

CastToArray converts the input value (from) to an array and returns an error if the conversion fails. It uses CastToNum to convert each element of the array to a float64.

func CastToBool

func CastToBool(from any) (bool, error)

CastToBool converts the input value (from) to a boolean value and returns an error if the conversion fails. It's a convenience function that wraps the spf13/cast library's ToBoolE function.

func CastToChar

func CastToChar(from any) (string, error)

CastToChar converts the input value (from) to a string and returns an error if the conversion fails. If the input value is a date.Date object, it converts it to a string in the format "dd/mm/yyyy". If the input value is any other type, it uses the spf13/cast library to convert the input value to a string.

func CastToDate

func CastToDate(from any) (date.Date, error)

CastToDate converts the input value (from) to a date.Date object and returns an error if the conversion fails. If the input value is a time.Time object, it converts it to a date.Date object. If the input value is a date.Date object, it returns the input value. If the input value is nil, it returns the zero value for the date.Date object. If the input value can be converted to an integer, it returns the date.Date object corresponding to the number of days since the epoch (1st January 1970). If the input value is any other type, it converts it to a string and then passes it to the ParseDate function. See that function's documentation for more information.

Example (Date Formats)

CastToDate supports multiple date string formats. See ParseDate documentation for the complete list of supported formats. 

// dd/mm/yyyy format
date1, _ := CastToDate("13/11/2025")
fmt.Println("dd/mm/yyyy:", date1)

// ddmmmyyyy format
date2, _ := CastToDate("13Nov2025")
fmt.Println("ddmmmyyyy:", date2)

// d-mmm-yy format
date3, _ := CastToDate("13-Nov-25")
fmt.Println("d-mmm-yy:", date3)

// ISO 8601 format
date4, _ := CastToDate("2025-11-13")
fmt.Println("ISO 8601:", date4)

// Long format with spaces
date5, _ := CastToDate("   13nov2025  ")
fmt.Println("With spaces:", date5)

// Output:
// dd/mm/yyyy: 2025-11-13
// ddmmmyyyy: 2025-11-13
// d-mmm-yy: 2025-11-13
// ISO 8601: 2025-11-13
// With spaces: 2025-11-13
#### Output 
dd/mm/yyyy: 2025-11-13
ddmmmyyyy: 2025-11-13
d-mmm-yy: 2025-11-13
ISO 8601: 2025-11-13
With spaces: 2025-11-13

Example (Invalid)

Values that cannot be converted to a date return an error. 

invalidDate, err := CastToDate("invalid")
fmt.Println(err)
fmt.Println(invalidDate)

// Output:
// date.ParseISO: cannot parse "invalid": incorrect length
// 1970-01-01
#### Output 
date.ParseISO: cannot parse "invalid": incorrect length
1970-01-01

Example (Nil)

Nil is acceptable as input. It returns the zero date \(1970\-01\-01\). 

noDate, err := CastToDate(nil)
fmt.Println(err)
fmt.Println(noDate)
fmt.Println(noDate.Equal(DataType_DATE.Missing().(date.Date)))

// Output:
// <nil>
// 1970-01-01
// true
#### Output 
<nil>
1970-01-01
true

Example (No Type Assertion Needed)

CastToDate accepts any type directly. No type assertions are needed. This demonstrates that you can pass values of any type without explicit type conversion. 

var stringFact any = "13/11/2025"
var intFact any = 19910 // Days since epoch
var nilFact any

// Pass directly to CastToDate - no need for type assertions like stringFact.(string)
dateFromString, err := CastToDate(stringFact)
fmt.Println("String:", dateFromString, err)

dateFromInt, err := CastToDate(intFact)
fmt.Println("Int:", dateFromInt, err)

dateFromNil, err := CastToDate(nilFact)
fmt.Println("Nil:", dateFromNil, err)

// Output:
// String: 2025-11-13 <nil>
// Int: 2024-07-06 <nil>
// Nil: 1970-01-01 <nil>
#### Output 
String: 2025-11-13 <nil>
Int: 2024-07-06 <nil>
Nil: 1970-01-01 <nil>

Example (Time To Date)

CastToDate can convert time.Time to date.Date, preserving the date and dropping time information. 

// time.Time with specific date and time
datetime := time.Date(2025, 11, 13, 14, 30, 45, 0, time.UTC)
dateOnly, _ := CastToDate(datetime)
fmt.Println("From time.Time:", dateOnly)

// time.Time at midnight
midnight := time.Date(2025, 11, 13, 0, 0, 0, 0, time.UTC)
dateFromMidnight, _ := CastToDate(midnight)
fmt.Println("From midnight:", dateFromMidnight)

// Output:
// From time.Time: 2025-11-13
// From midnight: 2025-11-13
#### Output 
From time.Time: 2025-11-13
From midnight: 2025-11-13

func CastToDatetime

func CastToDatetime(from any) (time.Time, error)

CastToDatetime converts the input value (from) to a time.Time object and returns an error if the conversion fails. If the type is a date.Date object, it converts it to a time.Time object with the time set to 00:00:00 UTC. Similarly, if the input can be parsed as a regular date using ParseDate, it will return that date with the time set to 00:00:00 UTC. If the input value is nil, it returns the zero value for the time.Time object. If the input value can be converted to an integer, it returns the time.Time object corresponding to the number of nanoseconds since the epoch (1st January 1970). If the input value is any other type, it uses the spf13/cast library to convert the input value to a time.Time object.

func CastToInt

func CastToInt(from any) (int64, error)

CastToInt converts the input value (from) to an int64 and returns an error if the conversion fails. If the input value is a date.Date object, it returns the number of days since the epoch (1st January 1970). If the input value is a time.Time object, it returns the number of nanoseconds since the epoch. If the input value is any other type, it uses the spf13/cast library to convert the input value to an int64.

func CastToNum

func CastToNum(from any) (float64, error)

CastToNum converts the input value (from) to a float64 and returns an error if the conversion fails. It's a convenience function that wraps the spf13/cast library's ToFloat64E function with the exception that it returns the math.NaN() value if the input value is nil.

func CopyMapToProto

func CopyMapToProto(in map[string]any, out proto.Message) error

CopyMapToProto copies the values from the input map to the output protocol buffer message. It uses reflection to set the fields of the output message based on the keys of the input map. If the key is not found in the output message, it skips that key. If the key is found in the output message, it converts the value from the input map to the appropriate data type and sets the field in the output message. If the conversion fails, it returns an error.

func CopyMapToProtos

func CopyMapToProtos(in map[string]any, outputs ...proto.Message) error

CopyMapToProtos copies values from a map to multiple output proto messages.

func DecodeBatch

func DecodeBatch(env Batch, body []byte, open bool) error

DecodeBatch fills env from one message delivered on env's queue. open is the receive's ok flag, false once the delivery channel has closed. Every batch a producer sends holds at least one row, poison pills included, so a closed channel, a body that does not decode and an empty batch all mean the rest of the input is lost. DecodeBatch reports each as a *LostData.

func Equal

func Equal(a, b any) bool

Equal returns true if a and b are equal.

func IsPoison

func IsPoison(body []byte) bool

IsPoison reports whether body, a marshaled batch, is a poison pill: its first row carries Done__. Every envelope numbers these fields alike, so a publisher holding only the bytes can still tell a lost pill from lost rows.

func MarshalBatch

func MarshalBatch(env Batch) ([]byte, error)

MarshalBatch encodes env, a batch of rows bound for its queue. A batch that does not encode never reaches the broker, so the error is a *LostData naming the batch's BufferLen rows.

func MarshalPoison

func MarshalPoison(env Poisoner) ([]byte, error)

MarshalPoison populates env in place with the poison-pill payload and returns the marshaled wire bytes ready for AMQP publish. All existing fields on env are reset.

func ParseDate

func ParseDate(dt string) (date.Date, error)

ParseDate parses the input date string (dt) into a date.Date object. It attempts to parse the date using several supported formats:

  • "dd/mm/yyyy" (2/1/2006)
  • "ddmmmyyy" (02Jan2006)
  • "d-mmm-yy" (2-Jan-06)
  • "yyyy Qn, Qn yyyy" (2006 Q1, Q1 2006)
  • long date (January 2, 2006)
  • Automatic date parsing

If none of the formats are successful, it returns an error and returns an empty date.Date.

func Sort

func Sort[S ~[]E, E proto.Message](s S, field string) error

Sort sorts the given slice of protocol buffer messages based on the specified field. It extracts field values once using proto reflection, then sorts using native Go types to avoid repeated reflection overhead.

func UnmatchedKeys

func UnmatchedKeys[V any](in map[string]V, outputs ...proto.Message) []string

UnmatchedKeys reports the keys of in that no message in outputs declares as a field, matched the way CopyMapToProto matches: case-insensitively, and through the "Proto" suffix that generated date and datetime fields carry. The result is sorted, and nil when every key matched. CopyMapToProto skips such keys silently on purpose, since the map it copies is a superset of any one queue's columns; a model node calls this on the values its model returned, where a miss means that column is left unwritten.

func ZipMap

func ZipMap(x []string, y []any) map[string]any

ZipMap creates a map from two slices, one of strings and one of any type. The function returns a map with the strings as keys and any values as values. The function panics if the length of the two slices are not equal. This function is useful for creating a map to pass to the CopyMapToProto function.

type Batch

Batch is a queue's on-wire envelope as DecodeBatch and MarshalBatch read it: the generated \<QueueName>Proto types. protoc-gen-go generates GetBufferLen; the queue template emits CacheLen and checks each type against Batch at compile time, so template drift fails the playpen build.

type Batch interface {
    Poisoner
    // GetBufferLen returns how many of the cached rows the batch carries.
    GetBufferLen() int32
    // CacheLen returns how many rows the envelope's cache holds.
    CacheLen() int
}

type DataType

type DataType int32


const (
    DataType_UNSPECIFIED DataType = 0 // Default value
    DataType_ARRAY       DataType = 1
    DataType_BOOL        DataType = 2
    DataType_CHAR        DataType = 3
    DataType_DATE        DataType = 4
    DataType_DATETIME    DataType = 5
    DataType_INT         DataType = 6
    DataType_NUM         DataType = 7
)

func GoTypeToDataType

func GoTypeToDataType(t string) (DataType, error)

GoTypeToDataType converts a Go type to a VOR Stream data type. If the Go type is not supported, it returns an error and the data type is set to the zero value of the DataType type.

func StringToDataType

func StringToDataType(s string) (DataType, error)

StringToDataType converts a string to a VOR Stream data type. If the string is not a valid data type, it returns an error and the data type is set to the zero value of the DataType type.

func (DataType) Descriptor

func (DataType) Descriptor() protoreflect.EnumDescriptor

func (DataType) Enum

func (x DataType) Enum() *DataType

func (DataType) EnumDescriptor

func (DataType) EnumDescriptor() ([]byte, []int)

Deprecated: Use DataType.Descriptor instead.

func (DataType) Missing

func (d DataType) Missing() any

Missing returns the missing value for the given data type. VOR Stream has no NULL: a blank input cell, a value that fails to parse, and an unset field all resolve to the type's missing value below.

  • NUM: NaN
  • CHAR: empty string
  • DATE: 1970-01-01
  • DATETIME: 1970-01-01T00:00:00Z
  • INT: 0
  • BOOL: false
  • ARRAY: empty slice

Only NUM's NaN is collision-free: NaN never occurs in normal numeric data, so it is unambiguously missing, and SQL aggregates skip it. The INT, DATE, and DATETIME sentinels are ordinary values, so a real 0 or a real 1970-01-01 is indistinguishable from missing.

TODO: The missing values for INT, DATE, and DATETIME are arbitrary and could conflict with real data.

func (DataType) Number

func (x DataType) Number() protoreflect.EnumNumber

func (DataType) String

func (x DataType) String() string

func (DataType) Type

func (DataType) Type() protoreflect.EnumType

type LostData

LostData reports rows a node lost on a queue: input it can no longer read, or output it could not hand to the broker. Losing data ends the run (the "Lost data fails the run" ADR in docs/adr), and this error is the line the node's log carries to say why.

type LostData struct {
    // Queue is the queue's name as declared in tables.csv.
    Queue string
    // Lost says which rows went missing, as a clause: "the rest of the input
    // was not read" or "2000 rows were not published".
    Lost string
    // Err is the cause.
    Err error
}

func Unpublished

func Unpublished(queue string, rows int, err error) *LostData

Unpublished reports a batch of rows that never reached queue's broker.

func (*LostData) Error

func (e *LostData) Error() string

func (*LostData) Unwrap

func (e *LostData) Unwrap() error

type Poisoner

Poisoner is the contract a queue envelope must satisfy to be marshaled into a poison-pill payload and routed to its destination queue. The generated \<QueueName>Proto types in playpens implement the methods via emissions from the queue template; proto.Message is satisfied automatically by all generated proto types.

type Poisoner interface {
    proto.Message

    // QueueName returns the lower-case name of the queue this proto
    // type belongs to. Generated <QueueName>Proto types return their
    // queue's declared name in tables.csv.
    QueueName() string

    // Poison overwrites the envelope in place with the on-wire shape of a
    // poison pill: every field is reset, then BufferLen is set to 1 and
    // Cache is set to a single entry with Done__=true. Any prior state
    // on the envelope is discarded. The receiving Iterator copies Cache[0]
    // into the outer queue struct via the generated $fromProto$ macro and
    // then trips its `if q.Done__ { return false }` check, exiting the
    // worker loop without emitting the row.
    Poison()
}

Generated by gomarkdoc