Compare commits
8 Commits
Author | SHA1 | Date | |
---|---|---|---|
fa6d73301e
|
|||
bfe62799d3
|
|||
ede912eb7b
|
|||
ff8f128fe8
|
|||
1971f1396f
|
|||
bf6c184d12
|
|||
770f5c5c64
|
|||
623c021689
|
@@ -46,6 +46,7 @@ var (
|
|||||||
TypeCursorTokenDecode = NewType("CURSOR_TOKEN_DECODE", langext.Ptr(500))
|
TypeCursorTokenDecode = NewType("CURSOR_TOKEN_DECODE", langext.Ptr(500))
|
||||||
TypeMongoFilter = NewType("MONGO_FILTER", langext.Ptr(500))
|
TypeMongoFilter = NewType("MONGO_FILTER", langext.Ptr(500))
|
||||||
TypeMongoReflection = NewType("MONGO_REFLECTION", langext.Ptr(500))
|
TypeMongoReflection = NewType("MONGO_REFLECTION", langext.Ptr(500))
|
||||||
|
TypeMongoInvalidOpt = NewType("MONGO_INVALIDOPT", langext.Ptr(500))
|
||||||
|
|
||||||
TypeWrap = NewType("Wrap", nil)
|
TypeWrap = NewType("Wrap", nil)
|
||||||
|
|
||||||
|
@@ -169,14 +169,32 @@ func (ee *ExErr) ShortLog(evt *zerolog.Event) {
|
|||||||
|
|
||||||
// RecursiveMessage returns the message to show
|
// RecursiveMessage returns the message to show
|
||||||
// = first error (top-down) that is not wrapping/foreign/empty
|
// = first error (top-down) that is not wrapping/foreign/empty
|
||||||
|
// = lowest level error (that is not empty)
|
||||||
|
// = fallback to self.message
|
||||||
func (ee *ExErr) RecursiveMessage() string {
|
func (ee *ExErr) RecursiveMessage() string {
|
||||||
|
|
||||||
|
// ==== [1] ==== first error (top-down) that is not wrapping/foreign/empty
|
||||||
|
|
||||||
for curr := ee; curr != nil; curr = curr.OriginalError {
|
for curr := ee; curr != nil; curr = curr.OriginalError {
|
||||||
if curr.Message != "" && curr.Category != CatWrap && curr.Category != CatForeign {
|
if curr.Message != "" && curr.Category != CatWrap && curr.Category != CatForeign {
|
||||||
return curr.Message
|
return curr.Message
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// fallback to self
|
// ==== [2] ==== lowest level error (that is not empty)
|
||||||
|
|
||||||
|
deepestMsg := ""
|
||||||
|
for curr := ee; curr != nil; curr = curr.OriginalError {
|
||||||
|
if curr.Message != "" {
|
||||||
|
deepestMsg = curr.Message
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if deepestMsg != "" {
|
||||||
|
return deepestMsg
|
||||||
|
}
|
||||||
|
|
||||||
|
// ==== [3] ==== fallback to self.message
|
||||||
|
|
||||||
return ee.Message
|
return ee.Message
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@@ -1,5 +1,5 @@
|
|||||||
package goext
|
package goext
|
||||||
|
|
||||||
const GoextVersion = "0.0.307"
|
const GoextVersion = "0.0.315"
|
||||||
|
|
||||||
const GoextVersionTimestamp = "2023-11-09T10:00:01+0100"
|
const GoextVersionTimestamp = "2023-11-12T03:10:55+0100"
|
||||||
|
@@ -5,6 +5,7 @@ import (
|
|||||||
"go.mongodb.org/mongo-driver/bson/bsoncodec"
|
"go.mongodb.org/mongo-driver/bson/bsoncodec"
|
||||||
"go.mongodb.org/mongo-driver/bson/bsontype"
|
"go.mongodb.org/mongo-driver/bson/bsontype"
|
||||||
"go.mongodb.org/mongo-driver/bson/primitive"
|
"go.mongodb.org/mongo-driver/bson/primitive"
|
||||||
|
"gogs.mikescher.com/BlackForestBytes/goext/langext"
|
||||||
"gogs.mikescher.com/BlackForestBytes/goext/rfctime"
|
"gogs.mikescher.com/BlackForestBytes/goext/rfctime"
|
||||||
"reflect"
|
"reflect"
|
||||||
)
|
)
|
||||||
@@ -18,9 +19,21 @@ func CreateGoExtBsonRegistry() *bsoncodec.Registry {
|
|||||||
rb.RegisterTypeDecoder(reflect.TypeOf(rfctime.RFC3339NanoTime{}), rfctime.RFC3339NanoTime{})
|
rb.RegisterTypeDecoder(reflect.TypeOf(rfctime.RFC3339NanoTime{}), rfctime.RFC3339NanoTime{})
|
||||||
rb.RegisterTypeDecoder(reflect.TypeOf(&rfctime.RFC3339NanoTime{}), rfctime.RFC3339NanoTime{})
|
rb.RegisterTypeDecoder(reflect.TypeOf(&rfctime.RFC3339NanoTime{}), rfctime.RFC3339NanoTime{})
|
||||||
|
|
||||||
|
rb.RegisterTypeDecoder(reflect.TypeOf(rfctime.UnixTime{}), rfctime.UnixTime{})
|
||||||
|
rb.RegisterTypeDecoder(reflect.TypeOf(&rfctime.UnixTime{}), rfctime.UnixTime{})
|
||||||
|
|
||||||
|
rb.RegisterTypeDecoder(reflect.TypeOf(rfctime.UnixMilliTime{}), rfctime.UnixMilliTime{})
|
||||||
|
rb.RegisterTypeDecoder(reflect.TypeOf(&rfctime.UnixMilliTime{}), rfctime.UnixMilliTime{})
|
||||||
|
|
||||||
|
rb.RegisterTypeDecoder(reflect.TypeOf(rfctime.UnixNanoTime{}), rfctime.UnixNanoTime{})
|
||||||
|
rb.RegisterTypeDecoder(reflect.TypeOf(&rfctime.UnixNanoTime{}), rfctime.UnixNanoTime{})
|
||||||
|
|
||||||
rb.RegisterTypeDecoder(reflect.TypeOf(rfctime.Date{}), rfctime.Date{})
|
rb.RegisterTypeDecoder(reflect.TypeOf(rfctime.Date{}), rfctime.Date{})
|
||||||
rb.RegisterTypeDecoder(reflect.TypeOf(&rfctime.Date{}), rfctime.Date{})
|
rb.RegisterTypeDecoder(reflect.TypeOf(&rfctime.Date{}), rfctime.Date{})
|
||||||
|
|
||||||
|
rb.RegisterTypeDecoder(reflect.TypeOf(rfctime.SecondsF64(0)), rfctime.SecondsF64(0))
|
||||||
|
rb.RegisterTypeDecoder(reflect.TypeOf(langext.Ptr(rfctime.SecondsF64(0))), rfctime.SecondsF64(0))
|
||||||
|
|
||||||
bsoncodec.DefaultValueEncoders{}.RegisterDefaultEncoders(rb)
|
bsoncodec.DefaultValueEncoders{}.RegisterDefaultEncoders(rb)
|
||||||
bsoncodec.DefaultValueDecoders{}.RegisterDefaultDecoders(rb)
|
bsoncodec.DefaultValueDecoders{}.RegisterDefaultDecoders(rb)
|
||||||
|
|
||||||
|
@@ -2,7 +2,14 @@ package rfctime
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"go.mongodb.org/mongo-driver/bson"
|
||||||
|
"go.mongodb.org/mongo-driver/bson/bsoncodec"
|
||||||
|
"go.mongodb.org/mongo-driver/bson/bsonrw"
|
||||||
|
"go.mongodb.org/mongo-driver/bson/bsontype"
|
||||||
"gogs.mikescher.com/BlackForestBytes/goext/timeext"
|
"gogs.mikescher.com/BlackForestBytes/goext/timeext"
|
||||||
|
"reflect"
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -54,6 +61,63 @@ func (d SecondsF64) MarshalJSON() ([]byte, error) {
|
|||||||
return json.Marshal(secs)
|
return json.Marshal(secs)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (d *SecondsF64) UnmarshalBSONValue(bt bsontype.Type, data []byte) error {
|
||||||
|
if bt == bson.TypeNull {
|
||||||
|
// we can't set nil in UnmarshalBSONValue (so we use default(struct))
|
||||||
|
// Use mongoext.CreateGoExtBsonRegistry if you need to unmarsh pointer values
|
||||||
|
// https://stackoverflow.com/questions/75167597
|
||||||
|
// https://jira.mongodb.org/browse/GODRIVER-2252
|
||||||
|
*d = SecondsF64(0)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if bt != bson.TypeDouble {
|
||||||
|
return errors.New(fmt.Sprintf("cannot unmarshal %v into SecondsF64", bt))
|
||||||
|
}
|
||||||
|
var tt float64
|
||||||
|
err := bson.RawValue{Type: bt, Value: data}.Unmarshal(&tt)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
*d = SecondsF64(tt)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d SecondsF64) MarshalBSONValue() (bsontype.Type, []byte, error) {
|
||||||
|
return bson.MarshalValue(d.Seconds())
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d SecondsF64) DecodeValue(dc bsoncodec.DecodeContext, vr bsonrw.ValueReader, val reflect.Value) error {
|
||||||
|
if val.Kind() == reflect.Ptr && val.IsNil() {
|
||||||
|
if !val.CanSet() {
|
||||||
|
return errors.New("ValueUnmarshalerDecodeValue")
|
||||||
|
}
|
||||||
|
val.Set(reflect.New(val.Type().Elem()))
|
||||||
|
}
|
||||||
|
|
||||||
|
tp, src, err := bsonrw.Copier{}.CopyValueToBytes(vr)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if val.Kind() == reflect.Ptr && len(src) == 0 {
|
||||||
|
val.Set(reflect.Zero(val.Type()))
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
err = d.UnmarshalBSONValue(tp, src)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if val.Kind() == reflect.Ptr {
|
||||||
|
val.Set(reflect.ValueOf(&d))
|
||||||
|
} else {
|
||||||
|
val.Set(reflect.ValueOf(d))
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func NewSecondsF64(t time.Duration) SecondsF64 {
|
func NewSecondsF64(t time.Duration) SecondsF64 {
|
||||||
return SecondsF64(t)
|
return SecondsF64(t)
|
||||||
}
|
}
|
||||||
|
@@ -2,6 +2,13 @@ package rfctime
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"go.mongodb.org/mongo-driver/bson"
|
||||||
|
"go.mongodb.org/mongo-driver/bson/bsoncodec"
|
||||||
|
"go.mongodb.org/mongo-driver/bson/bsonrw"
|
||||||
|
"go.mongodb.org/mongo-driver/bson/bsontype"
|
||||||
|
"reflect"
|
||||||
"strconv"
|
"strconv"
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
@@ -59,6 +66,63 @@ func (t *UnixTime) UnmarshalText(data []byte) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (t *UnixTime) UnmarshalBSONValue(bt bsontype.Type, data []byte) error {
|
||||||
|
if bt == bson.TypeNull {
|
||||||
|
// we can't set nil in UnmarshalBSONValue (so we use default(struct))
|
||||||
|
// Use mongoext.CreateGoExtBsonRegistry if you need to unmarsh pointer values
|
||||||
|
// https://stackoverflow.com/questions/75167597
|
||||||
|
// https://jira.mongodb.org/browse/GODRIVER-2252
|
||||||
|
*t = UnixTime{}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if bt != bson.TypeDateTime {
|
||||||
|
return errors.New(fmt.Sprintf("cannot unmarshal %v into UnixTime", bt))
|
||||||
|
}
|
||||||
|
var tt time.Time
|
||||||
|
err := bson.RawValue{Type: bt, Value: data}.Unmarshal(&tt)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
*t = UnixTime(tt)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t UnixTime) MarshalBSONValue() (bsontype.Type, []byte, error) {
|
||||||
|
return bson.MarshalValue(time.Time(t))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t UnixTime) DecodeValue(dc bsoncodec.DecodeContext, vr bsonrw.ValueReader, val reflect.Value) error {
|
||||||
|
if val.Kind() == reflect.Ptr && val.IsNil() {
|
||||||
|
if !val.CanSet() {
|
||||||
|
return errors.New("ValueUnmarshalerDecodeValue")
|
||||||
|
}
|
||||||
|
val.Set(reflect.New(val.Type().Elem()))
|
||||||
|
}
|
||||||
|
|
||||||
|
tp, src, err := bsonrw.Copier{}.CopyValueToBytes(vr)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if val.Kind() == reflect.Ptr && len(src) == 0 {
|
||||||
|
val.Set(reflect.Zero(val.Type()))
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
err = t.UnmarshalBSONValue(tp, src)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if val.Kind() == reflect.Ptr {
|
||||||
|
val.Set(reflect.ValueOf(&t))
|
||||||
|
} else {
|
||||||
|
val.Set(reflect.ValueOf(t))
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func (t UnixTime) Serialize() string {
|
func (t UnixTime) Serialize() string {
|
||||||
return strconv.FormatInt(t.Time().Unix(), 10)
|
return strconv.FormatInt(t.Time().Unix(), 10)
|
||||||
}
|
}
|
||||||
|
@@ -2,6 +2,13 @@ package rfctime
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"go.mongodb.org/mongo-driver/bson"
|
||||||
|
"go.mongodb.org/mongo-driver/bson/bsoncodec"
|
||||||
|
"go.mongodb.org/mongo-driver/bson/bsonrw"
|
||||||
|
"go.mongodb.org/mongo-driver/bson/bsontype"
|
||||||
|
"reflect"
|
||||||
"strconv"
|
"strconv"
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
@@ -59,6 +66,63 @@ func (t *UnixMilliTime) UnmarshalText(data []byte) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (t *UnixMilliTime) UnmarshalBSONValue(bt bsontype.Type, data []byte) error {
|
||||||
|
if bt == bson.TypeNull {
|
||||||
|
// we can't set nil in UnmarshalBSONValue (so we use default(struct))
|
||||||
|
// Use mongoext.CreateGoExtBsonRegistry if you need to unmarsh pointer values
|
||||||
|
// https://stackoverflow.com/questions/75167597
|
||||||
|
// https://jira.mongodb.org/browse/GODRIVER-2252
|
||||||
|
*t = UnixMilliTime{}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if bt != bson.TypeDateTime {
|
||||||
|
return errors.New(fmt.Sprintf("cannot unmarshal %v into UnixMilliTime", bt))
|
||||||
|
}
|
||||||
|
var tt time.Time
|
||||||
|
err := bson.RawValue{Type: bt, Value: data}.Unmarshal(&tt)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
*t = UnixMilliTime(tt)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t UnixMilliTime) MarshalBSONValue() (bsontype.Type, []byte, error) {
|
||||||
|
return bson.MarshalValue(time.Time(t))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t UnixMilliTime) DecodeValue(dc bsoncodec.DecodeContext, vr bsonrw.ValueReader, val reflect.Value) error {
|
||||||
|
if val.Kind() == reflect.Ptr && val.IsNil() {
|
||||||
|
if !val.CanSet() {
|
||||||
|
return errors.New("ValueUnmarshalerDecodeValue")
|
||||||
|
}
|
||||||
|
val.Set(reflect.New(val.Type().Elem()))
|
||||||
|
}
|
||||||
|
|
||||||
|
tp, src, err := bsonrw.Copier{}.CopyValueToBytes(vr)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if val.Kind() == reflect.Ptr && len(src) == 0 {
|
||||||
|
val.Set(reflect.Zero(val.Type()))
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
err = t.UnmarshalBSONValue(tp, src)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if val.Kind() == reflect.Ptr {
|
||||||
|
val.Set(reflect.ValueOf(&t))
|
||||||
|
} else {
|
||||||
|
val.Set(reflect.ValueOf(t))
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func (t UnixMilliTime) Serialize() string {
|
func (t UnixMilliTime) Serialize() string {
|
||||||
return strconv.FormatInt(t.Time().UnixMilli(), 10)
|
return strconv.FormatInt(t.Time().UnixMilli(), 10)
|
||||||
}
|
}
|
||||||
|
@@ -2,6 +2,13 @@ package rfctime
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"go.mongodb.org/mongo-driver/bson"
|
||||||
|
"go.mongodb.org/mongo-driver/bson/bsoncodec"
|
||||||
|
"go.mongodb.org/mongo-driver/bson/bsonrw"
|
||||||
|
"go.mongodb.org/mongo-driver/bson/bsontype"
|
||||||
|
"reflect"
|
||||||
"strconv"
|
"strconv"
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
@@ -59,6 +66,63 @@ func (t *UnixNanoTime) UnmarshalText(data []byte) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (t *UnixNanoTime) UnmarshalBSONValue(bt bsontype.Type, data []byte) error {
|
||||||
|
if bt == bson.TypeNull {
|
||||||
|
// we can't set nil in UnmarshalBSONValue (so we use default(struct))
|
||||||
|
// Use mongoext.CreateGoExtBsonRegistry if you need to unmarsh pointer values
|
||||||
|
// https://stackoverflow.com/questions/75167597
|
||||||
|
// https://jira.mongodb.org/browse/GODRIVER-2252
|
||||||
|
*t = UnixNanoTime{}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if bt != bson.TypeDateTime {
|
||||||
|
return errors.New(fmt.Sprintf("cannot unmarshal %v into UnixNanoTime", bt))
|
||||||
|
}
|
||||||
|
var tt time.Time
|
||||||
|
err := bson.RawValue{Type: bt, Value: data}.Unmarshal(&tt)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
*t = UnixNanoTime(tt)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t UnixNanoTime) MarshalBSONValue() (bsontype.Type, []byte, error) {
|
||||||
|
return bson.MarshalValue(time.Time(t))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t UnixNanoTime) DecodeValue(dc bsoncodec.DecodeContext, vr bsonrw.ValueReader, val reflect.Value) error {
|
||||||
|
if val.Kind() == reflect.Ptr && val.IsNil() {
|
||||||
|
if !val.CanSet() {
|
||||||
|
return errors.New("ValueUnmarshalerDecodeValue")
|
||||||
|
}
|
||||||
|
val.Set(reflect.New(val.Type().Elem()))
|
||||||
|
}
|
||||||
|
|
||||||
|
tp, src, err := bsonrw.Copier{}.CopyValueToBytes(vr)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if val.Kind() == reflect.Ptr && len(src) == 0 {
|
||||||
|
val.Set(reflect.Zero(val.Type()))
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
err = t.UnmarshalBSONValue(tp, src)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if val.Kind() == reflect.Ptr {
|
||||||
|
val.Set(reflect.ValueOf(&t))
|
||||||
|
} else {
|
||||||
|
val.Set(reflect.ValueOf(t))
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func (t UnixNanoTime) Serialize() string {
|
func (t UnixNanoTime) Serialize() string {
|
||||||
return strconv.FormatInt(t.Time().UnixNano(), 10)
|
return strconv.FormatInt(t.Time().UnixNano(), 10)
|
||||||
}
|
}
|
||||||
|
@@ -1,33 +1,28 @@
|
|||||||
package syncext
|
package syncext
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
|
||||||
"gogs.mikescher.com/BlackForestBytes/goext/langext"
|
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
type AtomicBool struct {
|
type Atomic[T any] struct {
|
||||||
v bool
|
v T
|
||||||
listener map[string]chan bool
|
lock sync.RWMutex
|
||||||
lock sync.Mutex
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewAtomicBool(value bool) *AtomicBool {
|
func NewAtomic[T any](value T) *Atomic[T] {
|
||||||
return &AtomicBool{
|
return &Atomic[T]{
|
||||||
v: value,
|
v: value,
|
||||||
listener: make(map[string]chan bool),
|
lock: sync.RWMutex{},
|
||||||
lock: sync.Mutex{},
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *AtomicBool) Get() bool {
|
func (a *Atomic[T]) Get() T {
|
||||||
a.lock.Lock()
|
a.lock.RLock()
|
||||||
defer a.lock.Unlock()
|
defer a.lock.RUnlock()
|
||||||
return a.v
|
return a.v
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *AtomicBool) Set(value bool) bool {
|
func (a *Atomic[T]) Set(value T) T {
|
||||||
a.lock.Lock()
|
a.lock.Lock()
|
||||||
defer a.lock.Unlock()
|
defer a.lock.Unlock()
|
||||||
|
|
||||||
@@ -35,79 +30,5 @@ func (a *AtomicBool) Set(value bool) bool {
|
|||||||
|
|
||||||
a.v = value
|
a.v = value
|
||||||
|
|
||||||
for k, v := range a.listener {
|
|
||||||
select {
|
|
||||||
case v <- value:
|
|
||||||
// message sent
|
|
||||||
default:
|
|
||||||
// no receiver on channel
|
|
||||||
delete(a.listener, k)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return oldValue
|
return oldValue
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *AtomicBool) Wait(waitFor bool) {
|
|
||||||
_ = a.WaitWithContext(context.Background(), waitFor)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *AtomicBool) WaitWithTimeout(timeout time.Duration, waitFor bool) error {
|
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
|
||||||
defer cancel()
|
|
||||||
return a.WaitWithContext(ctx, waitFor)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *AtomicBool) WaitWithContext(ctx context.Context, waitFor bool) error {
|
|
||||||
if err := ctx.Err(); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
if a.Get() == waitFor {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
uuid, _ := langext.NewHexUUID()
|
|
||||||
|
|
||||||
waitchan := make(chan bool)
|
|
||||||
|
|
||||||
a.lock.Lock()
|
|
||||||
a.listener[uuid] = waitchan
|
|
||||||
a.lock.Unlock()
|
|
||||||
defer func() {
|
|
||||||
a.lock.Lock()
|
|
||||||
delete(a.listener, uuid)
|
|
||||||
a.lock.Unlock()
|
|
||||||
}()
|
|
||||||
|
|
||||||
for {
|
|
||||||
if err := ctx.Err(); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
timeOut := 1024 * time.Millisecond
|
|
||||||
|
|
||||||
if dl, ok := ctx.Deadline(); ok {
|
|
||||||
timeOutMax := dl.Sub(time.Now())
|
|
||||||
if timeOutMax <= 0 {
|
|
||||||
timeOut = 0
|
|
||||||
} else if 0 < timeOutMax && timeOutMax < timeOut {
|
|
||||||
timeOut = timeOutMax
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if v, ok := ReadChannelWithTimeout(waitchan, timeOut); ok {
|
|
||||||
if v == waitFor {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
if err := ctx.Err(); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
if a.Get() == waitFor {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
113
syncext/bool.go
Normal file
113
syncext/bool.go
Normal file
@@ -0,0 +1,113 @@
|
|||||||
|
package syncext
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"gogs.mikescher.com/BlackForestBytes/goext/langext"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
type AtomicBool struct {
|
||||||
|
v bool
|
||||||
|
listener map[string]chan bool
|
||||||
|
lock sync.Mutex
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewAtomicBool(value bool) *AtomicBool {
|
||||||
|
return &AtomicBool{
|
||||||
|
v: value,
|
||||||
|
listener: make(map[string]chan bool),
|
||||||
|
lock: sync.Mutex{},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *AtomicBool) Get() bool {
|
||||||
|
a.lock.Lock()
|
||||||
|
defer a.lock.Unlock()
|
||||||
|
return a.v
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *AtomicBool) Set(value bool) bool {
|
||||||
|
a.lock.Lock()
|
||||||
|
defer a.lock.Unlock()
|
||||||
|
|
||||||
|
oldValue := a.v
|
||||||
|
|
||||||
|
a.v = value
|
||||||
|
|
||||||
|
for k, v := range a.listener {
|
||||||
|
select {
|
||||||
|
case v <- value:
|
||||||
|
// message sent
|
||||||
|
default:
|
||||||
|
// no receiver on channel
|
||||||
|
delete(a.listener, k)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return oldValue
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *AtomicBool) Wait(waitFor bool) {
|
||||||
|
_ = a.WaitWithContext(context.Background(), waitFor)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *AtomicBool) WaitWithTimeout(timeout time.Duration, waitFor bool) error {
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||||
|
defer cancel()
|
||||||
|
return a.WaitWithContext(ctx, waitFor)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *AtomicBool) WaitWithContext(ctx context.Context, waitFor bool) error {
|
||||||
|
if err := ctx.Err(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if a.Get() == waitFor {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
uuid, _ := langext.NewHexUUID()
|
||||||
|
|
||||||
|
waitchan := make(chan bool)
|
||||||
|
|
||||||
|
a.lock.Lock()
|
||||||
|
a.listener[uuid] = waitchan
|
||||||
|
a.lock.Unlock()
|
||||||
|
defer func() {
|
||||||
|
a.lock.Lock()
|
||||||
|
delete(a.listener, uuid)
|
||||||
|
a.lock.Unlock()
|
||||||
|
}()
|
||||||
|
|
||||||
|
for {
|
||||||
|
if err := ctx.Err(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
timeOut := 1024 * time.Millisecond
|
||||||
|
|
||||||
|
if dl, ok := ctx.Deadline(); ok {
|
||||||
|
timeOutMax := dl.Sub(time.Now())
|
||||||
|
if timeOutMax <= 0 {
|
||||||
|
timeOut = 0
|
||||||
|
} else if 0 < timeOutMax && timeOutMax < timeOut {
|
||||||
|
timeOut = timeOutMax
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if v, ok := ReadChannelWithTimeout(waitchan, timeOut); ok {
|
||||||
|
if v == waitFor {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
if err := ctx.Err(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if a.Get() == waitFor {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
101
wmo/queryFind.go
101
wmo/queryFind.go
@@ -4,18 +4,52 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"go.mongodb.org/mongo-driver/bson"
|
"go.mongodb.org/mongo-driver/bson"
|
||||||
"go.mongodb.org/mongo-driver/mongo"
|
"go.mongodb.org/mongo-driver/mongo"
|
||||||
|
"go.mongodb.org/mongo-driver/mongo/options"
|
||||||
"gogs.mikescher.com/BlackForestBytes/goext/exerr"
|
"gogs.mikescher.com/BlackForestBytes/goext/exerr"
|
||||||
"gogs.mikescher.com/BlackForestBytes/goext/langext"
|
"gogs.mikescher.com/BlackForestBytes/goext/langext"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (c *Coll[TData]) Find(ctx context.Context, filter bson.M) ([]TData, error) {
|
func (c *Coll[TData]) Find(ctx context.Context, filter bson.M, opts ...*options.FindOptions) ([]TData, error) {
|
||||||
|
|
||||||
pipeline := mongo.Pipeline{}
|
pipeline := mongo.Pipeline{}
|
||||||
pipeline = append(pipeline, bson.D{{Key: "$match", Value: filter}})
|
pipeline = append(pipeline, bson.D{{Key: "$match", Value: filter}})
|
||||||
|
|
||||||
|
for _, opt := range opts {
|
||||||
|
if opt != nil && opt.Sort != nil {
|
||||||
|
pipeline = append(pipeline, bson.D{{Key: "$sort", Value: opt.Sort}})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, opt := range opts {
|
||||||
|
if opt != nil && opt.Skip != nil {
|
||||||
|
pipeline = append(pipeline, bson.D{{Key: "$skip", Value: *opt.Skip}})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, opt := range opts {
|
||||||
|
if opt != nil && opt.Limit != nil {
|
||||||
|
pipeline = append(pipeline, bson.D{{Key: "$limit", Value: *opt.Limit}})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
pipeline = langext.ArrConcat(pipeline, c.extraModPipeline)
|
pipeline = langext.ArrConcat(pipeline, c.extraModPipeline)
|
||||||
|
|
||||||
cursor, err := c.coll.Aggregate(ctx, pipeline)
|
for _, opt := range opts {
|
||||||
|
if opt != nil && opt.Projection != nil {
|
||||||
|
pipeline = append(pipeline, bson.D{{Key: "$project", Value: opt.Projection}})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
convOpts := make([]*options.AggregateOptions, 0, len(opts))
|
||||||
|
for _, v := range opts {
|
||||||
|
vConv, err := convertFindOpt(v)
|
||||||
|
if err != nil {
|
||||||
|
return nil, exerr.Wrap(err, "mongo-aggregation failed").Any("pipeline", pipeline).Str("collection", c.Name()).Build()
|
||||||
|
}
|
||||||
|
convOpts = append(convOpts, vConv)
|
||||||
|
}
|
||||||
|
|
||||||
|
cursor, err := c.coll.Aggregate(ctx, pipeline, convOpts...)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, exerr.Wrap(err, "mongo-aggregation failed").Any("pipeline", pipeline).Str("collection", c.Name()).Build()
|
return nil, exerr.Wrap(err, "mongo-aggregation failed").Any("pipeline", pipeline).Str("collection", c.Name()).Build()
|
||||||
}
|
}
|
||||||
@@ -27,3 +61,66 @@ func (c *Coll[TData]) Find(ctx context.Context, filter bson.M) ([]TData, error)
|
|||||||
|
|
||||||
return res, nil
|
return res, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// converts FindOptions to AggregateOptions
|
||||||
|
func convertFindOpt(v *options.FindOptions) (*options.AggregateOptions, error) {
|
||||||
|
if v == nil {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
r := options.Aggregate()
|
||||||
|
|
||||||
|
if v.AllowDiskUse != nil {
|
||||||
|
r.SetAllowDiskUse(*v.AllowDiskUse)
|
||||||
|
}
|
||||||
|
if v.AllowPartialResults != nil {
|
||||||
|
return nil, exerr.New(exerr.TypeMongoInvalidOpt, "Invalid option 'AllowPartialResults' (cannot convert to AggregateOptions)").Build()
|
||||||
|
}
|
||||||
|
if v.BatchSize != nil {
|
||||||
|
r.SetBatchSize(*v.BatchSize)
|
||||||
|
}
|
||||||
|
if v.Collation != nil {
|
||||||
|
r.SetCollation(v.Collation)
|
||||||
|
}
|
||||||
|
if v.Comment != nil {
|
||||||
|
r.SetComment(*v.Comment)
|
||||||
|
}
|
||||||
|
if v.CursorType != nil {
|
||||||
|
return nil, exerr.New(exerr.TypeMongoInvalidOpt, "Invalid option 'CursorType' (cannot convert to AggregateOptions)").Build()
|
||||||
|
}
|
||||||
|
if v.Hint != nil {
|
||||||
|
r.SetHint(v.Hint)
|
||||||
|
}
|
||||||
|
if v.Max != nil {
|
||||||
|
return nil, exerr.New(exerr.TypeMongoInvalidOpt, "Invalid option 'Max' (cannot convert to AggregateOptions)").Build()
|
||||||
|
}
|
||||||
|
if v.MaxAwaitTime != nil {
|
||||||
|
r.SetMaxAwaitTime(*v.MaxAwaitTime)
|
||||||
|
}
|
||||||
|
if v.MaxTime != nil {
|
||||||
|
r.SetMaxTime(*v.MaxTime)
|
||||||
|
}
|
||||||
|
if v.Min != nil {
|
||||||
|
return nil, exerr.New(exerr.TypeMongoInvalidOpt, "Invalid option 'Min' (cannot convert to AggregateOptions)").Build()
|
||||||
|
}
|
||||||
|
if v.NoCursorTimeout != nil {
|
||||||
|
return nil, exerr.New(exerr.TypeMongoInvalidOpt, "Invalid option 'NoCursorTimeout' (cannot convert to AggregateOptions)").Build()
|
||||||
|
}
|
||||||
|
if v.OplogReplay != nil {
|
||||||
|
return nil, exerr.New(exerr.TypeMongoInvalidOpt, "Invalid option 'OplogReplay' (cannot convert to AggregateOptions)").Build()
|
||||||
|
}
|
||||||
|
if v.ReturnKey != nil {
|
||||||
|
return nil, exerr.New(exerr.TypeMongoInvalidOpt, "Invalid option 'ReturnKey' (cannot convert to AggregateOptions)").Build()
|
||||||
|
}
|
||||||
|
if v.ShowRecordID != nil {
|
||||||
|
return nil, exerr.New(exerr.TypeMongoInvalidOpt, "Invalid option 'ShowRecordID' (cannot convert to AggregateOptions)").Build()
|
||||||
|
}
|
||||||
|
if v.Snapshot != nil {
|
||||||
|
return nil, exerr.New(exerr.TypeMongoInvalidOpt, "Invalid option 'Snapshot' (cannot convert to AggregateOptions)").Build()
|
||||||
|
}
|
||||||
|
if v.Let != nil {
|
||||||
|
r.SetLet(v.Let)
|
||||||
|
}
|
||||||
|
|
||||||
|
return r, nil
|
||||||
|
}
|
||||||
|
@@ -4,7 +4,6 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"go.mongodb.org/mongo-driver/bson"
|
"go.mongodb.org/mongo-driver/bson"
|
||||||
"go.mongodb.org/mongo-driver/mongo"
|
"go.mongodb.org/mongo-driver/mongo"
|
||||||
ct "gogs.mikescher.com/BlackForestBytes/goext/cursortoken"
|
|
||||||
"gogs.mikescher.com/BlackForestBytes/goext/exerr"
|
"gogs.mikescher.com/BlackForestBytes/goext/exerr"
|
||||||
"gogs.mikescher.com/BlackForestBytes/goext/langext"
|
"gogs.mikescher.com/BlackForestBytes/goext/langext"
|
||||||
pag "gogs.mikescher.com/BlackForestBytes/goext/pagination"
|
pag "gogs.mikescher.com/BlackForestBytes/goext/pagination"
|
||||||
@@ -21,18 +20,15 @@ func (c *Coll[TData]) Paginate(ctx context.Context, filter pag.Filter, page int,
|
|||||||
|
|
||||||
pipelineSort := mongo.Pipeline{}
|
pipelineSort := mongo.Pipeline{}
|
||||||
pipelineFilter := mongo.Pipeline{}
|
pipelineFilter := mongo.Pipeline{}
|
||||||
pf1 := "_id"
|
sort := bson.D{}
|
||||||
pd1 := ct.SortASC
|
|
||||||
|
|
||||||
if filter != nil {
|
if filter != nil {
|
||||||
pipelineFilter = filter.FilterQuery()
|
pipelineFilter = filter.FilterQuery()
|
||||||
pf1, pd1 = filter.Pagination()
|
sort = filter.Sort()
|
||||||
}
|
}
|
||||||
|
|
||||||
if pd1 == ct.SortASC {
|
if len(sort) != 0 {
|
||||||
pipelineSort = append(pipelineSort, bson.D{{Key: "$sort", Value: bson.D{{Key: pf1, Value: +1}}}})
|
pipelineSort = append(pipelineSort, bson.D{{Key: "$sort", Value: sort}})
|
||||||
} else if pd1 == ct.SortDESC {
|
|
||||||
pipelineSort = append(pipelineSort, bson.D{{Key: "$sort", Value: bson.D{{Key: pf1, Value: -1}}}})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pipelinePaginate := mongo.Pipeline{}
|
pipelinePaginate := mongo.Pipeline{}
|
||||||
@@ -44,7 +40,7 @@ func (c *Coll[TData]) Paginate(ctx context.Context, filter pag.Filter, page int,
|
|||||||
}
|
}
|
||||||
|
|
||||||
pipelineCount := mongo.Pipeline{}
|
pipelineCount := mongo.Pipeline{}
|
||||||
pipelinePaginate = append(pipelinePaginate, bson.D{{Key: "$count", Value: "count"}})
|
pipelineCount = append(pipelineCount, bson.D{{Key: "$count", Value: "count"}})
|
||||||
|
|
||||||
pipelineList := langext.ArrConcat(mongo.Pipeline{}, pipelineFilter, pipelinePaginate, c.extraModPipeline)
|
pipelineList := langext.ArrConcat(mongo.Pipeline{}, pipelineFilter, pipelinePaginate, c.extraModPipeline)
|
||||||
pipelineTotalCount := langext.ArrConcat(mongo.Pipeline{}, pipelineFilter, pipelineCount)
|
pipelineTotalCount := langext.ArrConcat(mongo.Pipeline{}, pipelineFilter, pipelineCount)
|
||||||
|
Reference in New Issue
Block a user