sql2q

README

go-sql2q

SQL as Queue

Documentation

Index

• func Compose[T, U, V any](f func(T) U, g func(U) V) func(T) V
• func Error1st(ef []func() error) error
• func ErrorFromBool(ok bool, ng func() error) error
• func IfNg(ng error, f func())
• func IterReduce[T, U any](i Iter[T], init U, reducer func(state U, item T) U) U
• func IterReduceFilter[T, U any](i Iter[T], init U, filter func(T) bool, reducer func(state U, item T) U) U
• func MustOk[T any](t T, e error) T
• func PopLast[T any](s []T) []T
• type Add
• type Cls
• type Cnt
• type Codec
  • func CodecMust(c Codec, e error) Codec
  • func CodecNew(ser Pack, de Unpack) (Codec, error)
  • func (c Codec) Pack(ctx context.Context, msgs []Msg) (Msg, error)
  • func (c Codec) Unpack(ctx context.Context, packed Msg) ([]Msg, error)
• type Counter
  • func CounterNew() Counter
• type Del
• type Get
• type Id
  • func (i Id) AsInteger() int64
• type Iter
  • func IterEmpty[T any]() Iter[T]
  • func IterFromArray[T any](a []T) Iter[T]
  • func IterInts(lbi, ube int) Iter[int]
  • func IterMap[T, U any](i Iter[T], f func(T) U) Iter[U]
  • func (i Iter[T]) Count() int
  • func (i Iter[T]) ToArray() (a []T)
• type Lmt
• type Msg
  • func MsgEmpty() Msg
  • func MsgNew(i int64, dt []byte) Msg
  • func (m Msg) Data() []byte
  • func (m Msg) WithData(dt []byte) Msg
• type Option
  • func OptionEmpty[T any]() Option[T]
  • func OptionMap[T, U any](o Option[T], f func(T) U) Option[U]
  • func OptionNew[T any](t T) Option[T]
  • func (o Option[T]) Empty() bool
  • func (o Option[T]) HasValue() bool
  • func (o Option[T]) OrElse(f func() Option[T]) Option[T]
  • func (o Option[T]) UnwrapOr(alt T) T
  • func (o Option[T]) Value() T
• type Pack
• type Queue
  • func MemQueueNew(lmt int64) (Queue, error)
  • func (q Queue) Close() error
  • func (q Queue) Pop(ctx context.Context, cb func(context.Context, Msg) error) error
  • func (q Queue) PopMany(ctx context.Context, codec Codec, cb func(context.Context, []Msg) error) error
  • func (q Queue) Push(ctx context.Context, data []byte) error
  • func (q Queue) PushMany(ctx context.Context, messages []Msg, codec Codec) error
• type QueueBuilder
  • func (b QueueBuilder) Build() (q Queue, e error)
• type Unpack

Functions

func Compose

func Compose[T, U, V any](f func(T) U, g func(U) V) func(T) V

func Error1st

func Error1st(ef []func() error) error

func ErrorFromBool

func ErrorFromBool(ok bool, ng func() error) error

func IfNg

func IfNg(ng error, f func())

func IterReduce

func IterReduce[T, U any](i Iter[T], init U, reducer func(state U, item T) U) U

func IterReduceFilter

func IterReduceFilter[T, U any](i Iter[T], init U, filter func(T) bool, reducer func(state U, item T) U) U

func MustOk

func MustOk[T any](t T, e error) T

func PopLast

func PopLast[T any](s []T) []T

Types

type Add

type Add func(ctx context.Context, data []byte) error

Add push data to queue.

type Cls

type Cls func() error

Cls closes queue(optional)

type Cnt

type Cnt func(ctx context.Context) (int64, error)

Cnt gets queue count.

type Codec

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

func CodecMust

func CodecMust(c Codec, e error) Codec

func CodecNew

func CodecNew(ser Pack, de Unpack) (Codec, error)

func (Codec) Pack

func (c Codec) Pack(ctx context.Context, msgs []Msg) (Msg, error)

func (Codec) Unpack

func (c Codec) Unpack(ctx context.Context, packed Msg) ([]Msg, error)

type Counter

type Counter func() int64

func CounterNew

func CounterNew() Counter

type Del

type Del func(ctx context.Context, id Id) error

Del removes queue using its unique id.

type Get

type Get func(ctx context.Context) (Msg, error)

Get gets oldest msg from queue.

type Id

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

Id is unique queue identifier.

func (Id) AsInteger

func (i Id) AsInteger() int64

type Iter

type Iter[T any] func() Option[T]

func IterEmpty

func IterEmpty[T any]() Iter[T]

func IterFromArray

func IterFromArray[T any](a []T) Iter[T]

func IterInts

func IterInts(lbi, ube int) Iter[int]

func IterMap

func IterMap[T, U any](i Iter[T], f func(T) U) Iter[U]

func (Iter[T]) Count

func (i Iter[T]) Count() int

func (Iter[T]) ToArray

func (i Iter[T]) ToArray() (a []T)

type Lmt

type Lmt func(ctx context.Context) int64

Lmt gets queue limit.

type Msg

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

Msg is queue data with identifier.

func MsgEmpty

func MsgEmpty() Msg

func MsgNew

func MsgNew(i int64, dt []byte) Msg

func (Msg) Data

func (m Msg) Data() []byte

func (Msg) WithData

func (m Msg) WithData(dt []byte) Msg

type Option

type Option[T any] struct {
	// contains filtered or unexported fields
}

func OptionEmpty

func OptionEmpty[T any]() Option[T]

func OptionMap

func OptionMap[T, U any](o Option[T], f func(T) U) Option[U]

func OptionNew

func OptionNew[T any](t T) Option[T]

func (Option[T]) Empty

func (o Option[T]) Empty() bool

func (Option[T]) HasValue

func (o Option[T]) HasValue() bool

func (Option[T]) OrElse

func (o Option[T]) OrElse(f func() Option[T]) Option[T]

func (Option[T]) UnwrapOr

func (o Option[T]) UnwrapOr(alt T) T

func (Option[T]) Value

func (o Option[T]) Value() T

type Pack

type Pack func(ctx context.Context, msgs []Msg) (Msg, error)

Pack serializes many messages into single msg.

type Queue

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

func MemQueueNew

func MemQueueNew(lmt int64) (Queue, error)

func (Queue) Close

func (q Queue) Close() error

Close closes queue(optional)

func (Queue) Pop

func (q Queue) Pop(ctx context.Context, cb func(context.Context, Msg) error) error

Pop gets msg and remove it if client received. 1. try get msg 2. call callback cb and send msg 3. if callback recv msg, try remove queue

func (Queue) PopMany

func (q Queue) PopMany(ctx context.Context, codec Codec, cb func(context.Context, []Msg) error) error

PopMany pop many messages from single virtual msg.

func (Queue) Push

func (q Queue) Push(ctx context.Context, data []byte) error

Push push msg if queue has enough space. 1. get queue size 2. add msg if queue has enough space

func (Queue) PushMany

func (q Queue) PushMany(ctx context.Context, messages []Msg, codec Codec) error

PushMany push many messages as single msg.

type QueueBuilder

type QueueBuilder struct {
	Add
	Get
	Del
	Cnt
	Lmt
	Cls
}

func (QueueBuilder) Build

func (b QueueBuilder) Build() (q Queue, e error)

type Unpack

type Unpack func(ctx context.Context, packed Msg) ([]Msg, error)

Unpack deserializes single message into many messages.