liven

package module
v0.0.0-...-1c08215 Latest Latest
Warning

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

 Go to latest Published: Jul 3, 2026 License: MIT Imports: 9 Imported by: 0

README

liven-go

Go Reference

Go client SDK for LivenDB

package main

import (
    "fmt"
    "github.com/livendb/liven-go"
)

func main() {
    client, err := liven.Connect("127.0.0.1:43121")
    if err != nil { panic(err) }
    defer client.Close()

    client.Insert("users", "u1", map[string]interface{}{"name": "Alice"})
    user, _ := client.Get("users", "u1")
    fmt.Println(user[0].Value.Field("name"))
    // → Alice
}

Installation

go get github.com/livendb/liven-go

Quick Start

1. Connect
import "github.com/livendb/liven-go"

// Default port 43121
client, err := liven.Connect("localhost")

// With auth key
client, err := liven.Connect("localhost:43121?auth_key=your-api-key")
2. CRUD
client.Insert("events", "e1", map[string]interface{}{"type": "click"})
client.Upsert("events", "e1", map[string]interface{}{"type": "click", "value": 99})
client.Update("users", "u1", map[string]interface{}{"last_login": 1234567890})
record, _ := client.Get("users", "u1")
client.Delete("events", "e1")
client.Clear("sessions")
client.DropStream("old_data")
3. Batch
client.InsertMany("events", [][2]interface{}{
    {"e1", map[string]interface{}{"type": "click"}},
    {"e2", map[string]interface{}{"type": "view"}},
})
4. Pipeline
results, err := client.Run(
    liven.NewPipeline("orders").
        Filter(liven.FilterField("amount").Gte(100)).
        Filter(liven.FilterField("status").Eq("completed")).
        Sort("amount", true).
        Limit(10),
)
5. Shorthand
client.Filter("events", liven.FilterField("type").Eq("click"))
client.Limit("events", 50)
client.Count("events")
client.Sort("orders", "total", true)
client.Page("events", 1, 25)
client.PageCursor("feed", "cursor_abc", 25)
client.Map("users", []string{"name", "email"})
client.Window("pageviews", 60000, "count")
client.Group("events", "type", []string{"count", "sum(value)"})
client.Distinct("visitors", "ip_address")
client.VectorFilter("documents", "embedding", []int8{1, 0, -1}, 0.75)
6. Joins
client.Enrich("orders", "users", "user_id")
client.Correlate("events", "sessions", "session_id", 5000)
client.Chain("reviews", "orders", "order_id")
client.Sequence("events", []liven.FilterExpr{
    liven.FilterField("type").Eq("login"),
    liven.FilterField("type").Eq("purchase"),
}, 300000)
7. Mutations
client.RunUpdate(
    liven.NewPipeline("events").Filter(liven.FilterField("status").Eq("pending")),
    map[string]interface{}{"status": "processed"},
)
client.RunDelete(
    liven.NewPipeline("events").Filter(liven.FilterField("type").Eq("temp")),
)
8. Metadata
streams, _ := client.Streams()
status, _ := client.Status()
9. Raw DSL
results, _ := client.Query(`from("users") | count()`)

Development

go test ./...
go run examples/demo.go
go run examples/demo.go --key your-api-key

License

MIT

Documentation

Index

Constants

View Source 
const (
	ProtocolVersion      = 0x01
	DiscriminatorMsgpack = 0x02
	DiscriminatorVector  = 0x03
	HeaderSize           = 5
)

Wire protocol constants.

Variables

This section is empty.

Functions

func Clear 

func Clear(stream string) string

func Delete 

func Delete(stream, key string) string

func Drop 

func Drop(stream string) string

func EncodeFrame 

func EncodeFrame(frame LivenFrame) ([]byte, error)

EncodeFrame encodes a LivenFrame into a framed byte buffer.

func Insert 

func Insert(stream, key string, value interface{}) string

func InsertBatch 

func InsertBatch(stream string, batch [][2]interface{}) string

func ListStreams 

func ListStreams() string

func Status 

func Status() string

func Update 

func Update(stream, key string, value interface{}) string

func Upsert 

func Upsert(stream, key string, value interface{}) string

func UpsertBatch 

func UpsertBatch(stream string, batch [][2]interface{}) string

Types

type DataValue 

type DataValue struct {
	Type  string      `msgpack:"-" json:"type"`
	Value interface{} `msgpack:"-" json:"value,omitempty"`
}

func DVArr 

func DVArr(v []DataValue) DataValue

func DVBin 

func DVBin(v []byte) DataValue

func DVBool 

func DVBool(v bool) DataValue

func DVFlt 

func DVFlt(v float64) DataValue

func DVInt 

func DVInt(v int64) DataValue

func DVNull 

func DVNull() DataValue

Standard DataValue constructors.

func DVObj 

func DVObj(v map[string]DataValue) DataValue

func DVStr 

func DVStr(v string) DataValue

func DVUInt 

func DVUInt(v uint64) DataValue

func DVVec 

func DVVec(v []int8) DataValue

func (DataValue) Field 

func (dv DataValue) Field(name string) interface{}

Field extracts a named field from an Object DataValue, or from a String DataValue containing JSON.

type FieldFilter 

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

FieldFilter is an intermediate builder returned by FilterField.

func FilterField 

func FilterField(name string) FieldFilter

func (FieldFilter) Between 

func (f FieldFilter) Between(low, high float64) FilterExpr

func (FieldFilter) Contains 

func (f FieldFilter) Contains(value string) FilterExpr

func (FieldFilter) EndsWith 

func (f FieldFilter) EndsWith(value string) FilterExpr

func (FieldFilter) Eq 

func (f FieldFilter) Eq(value interface{}) FilterExpr

func (FieldFilter) Gt 

func (f FieldFilter) Gt(value float64) FilterExpr

func (FieldFilter) Gte 

func (f FieldFilter) Gte(value float64) FilterExpr

func (FieldFilter) In 

func (f FieldFilter) In(values []interface{}) FilterExpr

func (FieldFilter) Lt 

func (f FieldFilter) Lt(value float64) FilterExpr

func (FieldFilter) Lte 

func (f FieldFilter) Lte(value float64) FilterExpr

func (FieldFilter) Ne 

func (f FieldFilter) Ne(value interface{}) FilterExpr

func (FieldFilter) StartsWith 

func (f FieldFilter) StartsWith(value string) FilterExpr

type FilterExpr 

type FilterExpr struct {
	Kind     string      `msgpack:"kind"`
	Field    string      `msgpack:"field,omitempty"`
	Operator string      `msgpack:"operator,omitempty"`
	Value    interface{} `msgpack:"value,omitempty"`
	Left     *FilterExpr `msgpack:"left,omitempty"`
	Right    *FilterExpr `msgpack:"right,omitempty"`
	Expr     *FilterExpr `msgpack:"expr,omitempty"`
}

func FilterAnd 

func FilterAnd(filters ...FilterExpr) FilterExpr

func FilterNot 

func FilterNot(filter FilterExpr) FilterExpr

func FilterOr 

func FilterOr(filters ...FilterExpr) FilterExpr

type LivenClient 

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

LivenClient is a native wire protocol client for LivenDB.

Usage: 

client, err := liven.Connect("127.0.0.1:43121")
if err != nil { ... }
defer client.Close()

client.Insert("users", "u1", map[string]interface{}{"name": "Alice"})
records, _ := client.Get("users", "u1")

func Connect 

func Connect(addr string) (*LivenClient, error)

Connect connects to a Liven server over TCP. Supports optional auth_key via URL query: "host:port?auth_key=xxx"

func (*LivenClient) Chain 

func (c *LivenClient) Chain(stream, targetStream, joinKey string) ([]LivenRecord, error)

func (*LivenClient) Clear 

func (c *LivenClient) Clear(stream string) ([]LivenRecord, error)

func (*LivenClient) Close 

func (c *LivenClient) Close() error

Close closes the connection.

func (*LivenClient) Correlate 

func (c *LivenClient) Correlate(stream, sourceStream, joinKey string, withinMs uint64) ([]LivenRecord, error)

func (*LivenClient) Count 

func (c *LivenClient) Count(stream string) ([]LivenRecord, error)

func (*LivenClient) Delete 

func (c *LivenClient) Delete(stream, key string) ([]LivenRecord, error)

func (*LivenClient) Distinct 

func (c *LivenClient) Distinct(stream, field string) ([]LivenRecord, error)

func (*LivenClient) DropStream 

func (c *LivenClient) DropStream(stream string) ([]LivenRecord, error)

func (*LivenClient) Enrich 

func (c *LivenClient) Enrich(stream, sourceStream, joinKey string) ([]LivenRecord, error)

func (*LivenClient) Filter 

func (c *LivenClient) Filter(stream string, filter FilterExpr) ([]LivenRecord, error)

func (*LivenClient) Get 

func (c *LivenClient) Get(stream, key string) ([]LivenRecord, error)

func (*LivenClient) Group 

func (c *LivenClient) Group(stream, field string, aggregations []string) ([]LivenRecord, error)

func (*LivenClient) Insert 

func (c *LivenClient) Insert(stream, key string, value interface{}) ([]LivenRecord, error)

func (*LivenClient) InsertMany 

func (c *LivenClient) InsertMany(stream string, batch [][2]interface{}) ([]LivenRecord, error)

func (*LivenClient) Limit 

func (c *LivenClient) Limit(stream string, count int) ([]LivenRecord, error)

func (*LivenClient) Map 

func (c *LivenClient) Map(stream string, fields []string) ([]LivenRecord, error)

func (*LivenClient) Page 

func (c *LivenClient) Page(stream string, pageNum, pageSize int) ([]LivenRecord, error)

func (*LivenClient) PageCursor 

func (c *LivenClient) PageCursor(stream, cursor string, pageSize int) ([]LivenRecord, error)

func (*LivenClient) Query 

func (c *LivenClient) Query(dsl string) ([]LivenRecord, error)

Query sends a raw LIVEN DSL string and returns parsed records.

func (*LivenClient) Run 

func (c *LivenClient) Run(pipeline *Pipeline) ([]LivenRecord, error)

func (*LivenClient) RunDelete 

func (c *LivenClient) RunDelete(pipeline *Pipeline) ([]LivenRecord, error)

func (*LivenClient) RunListen 

func (c *LivenClient) RunListen(pipeline *Pipeline) ([]LivenRecord, error)

func (*LivenClient) RunUpdate 

func (c *LivenClient) RunUpdate(pipeline *Pipeline, value interface{}) ([]LivenRecord, error)

func (*LivenClient) Sequence 

func (c *LivenClient) Sequence(stream string, steps []FilterExpr, withinMs uint64) ([]LivenRecord, error)

func (*LivenClient) Sort 

func (c *LivenClient) Sort(stream, field string, descending bool) ([]LivenRecord, error)

func (*LivenClient) Status 

func (c *LivenClient) Status() ([]LivenRecord, error)

func (*LivenClient) Streams 

func (c *LivenClient) Streams() ([]LivenRecord, error)

func (*LivenClient) Update 

func (c *LivenClient) Update(stream, key string, value interface{}) ([]LivenRecord, error)

func (*LivenClient) Upsert 

func (c *LivenClient) Upsert(stream, key string, value interface{}) ([]LivenRecord, error)

func (*LivenClient) UpsertMany 

func (c *LivenClient) UpsertMany(stream string, batch [][2]interface{}) ([]LivenRecord, error)

func (*LivenClient) VectorFilter 

func (c *LivenClient) VectorFilter(stream, field string, queryVector []int8, threshold float64) ([]LivenRecord, error)

func (*LivenClient) Window 

func (c *LivenClient) Window(stream string, durationMs uint64, strategy string) ([]LivenRecord, error)

type LivenFrame 

type LivenFrame struct {
	Type            string        `msgpack:"-" json:"type"`
	Query           string        `msgpack:"-" json:"query,omitempty"`
	Records         []LivenRecord `msgpack:"-" json:"records,omitempty"`
	ClientID        string        `msgpack:"-" json:"client_id,omitempty"`
	ProtocolVersion *uint8        `msgpack:"-" json:"protocol_version,omitempty"`
	Message         string        `msgpack:"-" json:"message,omitempty"`
	Values          []int8        `msgpack:"-" json:"values,omitempty"`
}

func ReadFrame 

func ReadFrame(conn net.Conn) (LivenFrame, error)

ReadFrame reads one complete frame from a TCP connection.

type LivenRecord 

type LivenRecord struct {
	SequenceID uint64    `msgpack:"-" json:"sequence_id"`
	Timestamp  int64     `msgpack:"-" json:"timestamp"`
	TypeTag    uint8     `msgpack:"-" json:"type_tag"`
	Flags      uint8     `msgpack:"-" json:"flags"`
	StreamName string    `msgpack:"-" json:"stream_name"`
	Key        string    `msgpack:"-" json:"key"`
	Value      DataValue `msgpack:"-" json:"value"`
}

func (LivenRecord) String 

func (r LivenRecord) String() string

type Pipeline 

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

func NewPipeline 

func NewPipeline(stream string) *Pipeline

func (*Pipeline) Build 

func (p *Pipeline) Build() string

func (*Pipeline) BuildDelete 

func (p *Pipeline) BuildDelete() string

func (*Pipeline) BuildListen 

func (p *Pipeline) BuildListen() string

func (*Pipeline) BuildUpdate 

func (p *Pipeline) BuildUpdate(value interface{}) string

func (*Pipeline) Chain 

func (p *Pipeline) Chain(targetStream, joinKey string) *Pipeline

func (*Pipeline) Correlate 

func (p *Pipeline) Correlate(sourceStream, joinKey string, withinMs uint64) *Pipeline

func (*Pipeline) Count 

func (p *Pipeline) Count() *Pipeline

func (*Pipeline) Distinct 

func (p *Pipeline) Distinct(field string) *Pipeline

func (*Pipeline) Enrich 

func (p *Pipeline) Enrich(sourceStream, joinKey string) *Pipeline

func (*Pipeline) Filter 

func (p *Pipeline) Filter(f FilterExpr) *Pipeline

func (*Pipeline) Get 

func (p *Pipeline) Get(key string) *Pipeline

func (*Pipeline) Group 

func (p *Pipeline) Group(field string, aggregations []string) *Pipeline

func (*Pipeline) Limit 

func (p *Pipeline) Limit(count int) *Pipeline

func (*Pipeline) Map 

func (p *Pipeline) Map(fields []string) *Pipeline

func (*Pipeline) Page 

func (p *Pipeline) Page(pageNum, pageSize int) *Pipeline

func (*Pipeline) PageCursor 

func (p *Pipeline) PageCursor(cursor string, pageSize int) *Pipeline

func (*Pipeline) Sequence 

func (p *Pipeline) Sequence(steps []FilterExpr, withinMs uint64) *Pipeline

func (*Pipeline) Sort 

func (p *Pipeline) Sort(field string, descending bool) *Pipeline

func (*Pipeline) VectorFilter 

func (p *Pipeline) VectorFilter(field string, queryVector []int8, threshold float64) *Pipeline

func (*Pipeline) Window 

func (p *Pipeline) Window(durationMs uint64, strategy string) *Pipeline

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.