geek

README

Geek

Geek is a kdb+/q interface for Go (Golang). It comes with a kdb+/q load balancer ConnPool, which uses a buffered channel to handle the connections pool.

• ConnPool.Handle
  • doesn't serialize or deserialize IPC message.
  • before passing on IPC messages, define .geek.user and .geek.ip on the q process
• ConnPool.Sync
  • need a known go type pointer as an parameter

Contents

• Installation
• Quick Start
• Data Types

Installation

1. Install Geek

go get -u github.com/jshinonome/geek

1. Import it in your code

import "github.com/jshinonome/geek"

Quick Start

Load balancer

1. Create an example.go file as below

package main

import "github.com/jshinonome/geek"

func main() {
	// connect to a q process @ 1800
	q1 := geek.QProcess{Port: 1800}
	q1.Dial()
	// connect to a q process @ 1801
	q2 := geek.QProcess{Port: 1801}
	q2.Dial()
	qConnPool := geek.NewConnPool()
	qConnPool.Put(&q1)
	qConnPool.Put(&q2)
	qConnPool.Serving()

	qEngine := geek.DefaultEngine(qConnPool)
	qEngine.Run() // listen and server on :8101
}

1. Start two q processes

q -p 1800

q -p 1801

1. Run the q Engine

go run example.go

1. Start a new q process and run queries(geek.Engine doesn't keep client connection, so don't try to open a handle)

`::8101 ("system";"p") //1800
`::8101 ("system";"p") //1801

Interface with q process

1. Create an example.go file as below

package main

import (
	"fmt"
	"time"

	"github.com/jshinonome/geek"
)

func main() {
	// connect to a q process @ 1800
	q := geek.QProcess{Port: 1800}
	q.Dial()
	sym := "a"
	f := struct {
		Api string
		Sym string
	}{
		"getTrade", sym,
	}
	r := make([]trade, 0)
	err := q.Sync(&r, f)
	if err != nil {
		fmt.Println(err)
	}
	for _, t := range r {
		fmt.Printf("%+v\n", t)
	}
}

type trade struct {
	Time  time.Time `k:"time"`
	Sym   string    `k:"sym"`
	Price float64   `k:"price"`
	Qty   int64     `k:"qty"`
}

1. Start a q process

q -p 1800
trade:([]time:"p"$.z.D-til 10;sym:10?`a`b;price:10?10.0;qty:10?10);
getTrade:{select from trade where sym=x};
.z.pg:{[x]0N!(`zpg;x);value x};

1. Run go program

go run example.go

Data Types

Basic Types

Intend to support only these 6 types

kdb	go
long	int64
float	float64
char	byte
symbol	string
timestamp	time.Time
boolean	bool

Other Types

go struct without k tags -> k mixed list

struct {
	Func   string
	Param1 string
	Param2 string
}

go map -> k dictionary

map[string]bool

go struct with k tags(keys' name) -> k dictionary

struct {
	Key1 string    `k:"sym"`
	Key2 time.Time `k:"time"`
	Key3 float64   `k:"qty"`
}

go list of a struct with k tags(column names) -> k table

[]struct {
	Sym  string    `k:"sym"`
	Time time.Time `k:"time"`
	Qty  float64   `k:"qty"`
}

Documentation

Index

• Constants
• Variables
• func Compress(msg []byte) []byte
• func Decompress(cMsg []byte) []byte
• func PeekAPI(msg []byte) string
• type Authenticator
• type ConnPool
  • func NewConnPool() *ConnPool
  • func (pool *ConnPool) GetQueueCounter() int
  • func (pool *ConnPool) Handle(qClient *QProcess) (int64, int64, string, error)
  • func (pool *ConnPool) Put(q *QProcess)
  • func (pool *ConnPool) Reload()
  • func (pool *ConnPool) Revive()
  • func (pool *ConnPool) Serving() error
  • func (pool *ConnPool) Sync(k interface{}, args interface{}) error
  • func (pool *ConnPool) Validate(API string) error
• type Engine
  • func DefaultEngine(pool *ConnPool) *Engine
  • func (e *Engine) Run() error
• type GeekErr
  • func (ge GeekErr) Error() string
• type HandlerFunc
  • func LoggerWithConfig(conf LoggerConfig) HandlerFunc
• type LogFormatter
• type LogParams
• type LoggerConfig
• type QProcess
  • func (q *QProcess) Async(args interface{}) error
  • func (q *QProcess) Close()
  • func (q *QProcess) Dial() error
  • func (q *QProcess) Discard() error
  • func (q *QProcess) Err(err error) error
  • func (q QProcess) IsConnected() bool
  • func (q *QProcess) Sync(k interface{}, args interface{}) error

Constants

const (
	GeekUser string = ".geek.user"
	GeekIP   string = ".geek.ip"
)

var names used on the q processes

const (
	ASYNC byte = iota
	SYNC
	RESPONSE
)

const (
	K0 byte = 0
	KB byte = 1
	KJ byte = 7
	KF byte = 9
	KC byte = 10
	KS byte = 11
	KP byte = 12

	XT byte = 98
	XD byte = 99
)

const (
	Wj = math.MaxInt64
	Nj = math.MinInt64
)

Variables

var (
	ErrAuth                 = errors.New("geek: wrong credential?")
	ErrVersion              = errors.New("geek: require version 3+ kdb+")
	ErrNotConnected         = errors.New("geek: not connected")
	ErrMaxRetryTimesReached = errors.New("geek: maximum retry times reached")
	ErrNotAllowedAPI        = errors.New("geek: not allowed API")
	ErrInvalidQuery         = errors.New("geek: first item is not a symbol or null symbol")
)

var (
	RB = reflect.TypeOf([]bool{})
	RJ = reflect.TypeOf([]int64{})
	RF = reflect.TypeOf([]float64{})
	RC = reflect.TypeOf([]byte{})
	RS = reflect.TypeOf([]string{})
	RT = reflect.TypeOf([]time.Time{})
	RP = reflect.TypeOf([]timestamppb.Timestamp{})
	R0 = reflect.TypeOf([][]byte{})
)

var (
	Nf = math.NaN()
	Wf = math.Inf(1)
)

var DefaultLogWriter io.Writer = os.Stdout

Functions

func Compress

func Compress(msg []byte) []byte

func Decompress

func Decompress(cMsg []byte) []byte

func PeekAPI

func PeekAPI(msg []byte) string

peek the first symbol(API) in the incoming message

Types

type Authenticator

type Authenticator func(string, string) error

type ConnPool

type ConnPool struct {
	Conns map[int]*QProcess

	Timeout        time.Duration
	ReviveInterval time.Duration
	DeadConns      map[int]bool
	AllowedAPI     map[string]bool
	RetryTimes     int
	// contains filtered or unexported fields
}

func NewConnPool

func NewConnPool() *ConnPool

func (*ConnPool) GetQueueCounter

func (pool *ConnPool) GetQueueCounter() int

func (*ConnPool) Handle

func (pool *ConnPool) Handle(qClient *QProcess) (int64, int64, string, error)

func (*ConnPool) Put

func (pool *ConnPool) Put(q *QProcess)

func (*ConnPool) Reload

func (pool *ConnPool) Reload()

func (*ConnPool) Revive

func (pool *ConnPool) Revive()

func (*ConnPool) Serving

func (pool *ConnPool) Serving() error

func (*ConnPool) Sync

func (pool *ConnPool) Sync(k interface{}, args interface{}) error

func (*ConnPool) Validate

func (pool *ConnPool) Validate(API string) error

type Engine

type Engine struct {
	Port    int
	Auth    Authenticator
	Pool    *ConnPool
	Handler HandlerFunc
}

func DefaultEngine

func DefaultEngine(pool *ConnPool) *Engine

func (*Engine) Run

func (e *Engine) Run() error

type GeekErr

type GeekErr struct {
	Msg string
}

func (GeekErr) Error

func (ge GeekErr) Error() string

type HandlerFunc

type HandlerFunc func(*ConnPool, *QProcess)

func LoggerWithConfig

func LoggerWithConfig(conf LoggerConfig) HandlerFunc

LoggerWithConfig instance a Logger middleware with config.

type LogFormatter

type LogFormatter func(params LogParams) string

type LogParams

type LogParams struct {
	TimeStamp    time.Time
	Duration     time.Duration
	Status       string
	User         string
	ClientIP     string
	API          string
	InputSize    int64
	OutputSize   int64
	ErrorMsg     string
	QueueCounter int
}

type LoggerConfig

type LoggerConfig struct {
	// Optional. Default value is geek.defaultLogFormatter
	Formatter LogFormatter

	// Output is a writer where logs are written.
	// Optional. Default value is geek.DefaultWriter.
	Output io.Writer
}

type QProcess

type QProcess struct {
	Host      string
	Port      int
	User      string
	Password  string
	Timeout   time.Duration
	TLSConfig *tls.Config
	// contains filtered or unexported fields
}

func (*QProcess) Async

func (q *QProcess) Async(args interface{}) error

func (*QProcess) Close

func (q *QProcess) Close()

func (*QProcess) Dial

func (q *QProcess) Dial() error

func (*QProcess) Discard

func (q *QProcess) Discard() error

func (*QProcess) Err

func (q *QProcess) Err(err error) error

func (QProcess) IsConnected

func (q QProcess) IsConnected() bool

func (*QProcess) Sync

func (q *QProcess) Sync(k interface{}, args interface{}) error