Compare commits
8 Commits
Author | SHA1 | Date | |
---|---|---|---|
|
27fa6e9173 | ||
bd55a35dc3 | |||
653bd386cc | |||
|
558c6f4d7c | ||
d7dd6fbeb2 | |||
a00cf2c8d9 | |||
|
a3e8ab2492 | ||
06da500ef4 |
@ -1,5 +1,5 @@
|
|||||||
# Micro
|
# Micro
|
||||||
![Coverage](https://img.shields.io/badge/Coverage-44.8%25-yellow)
|
![Coverage](https://img.shields.io/badge/Coverage-44.7%25-yellow)
|
||||||
|
|
||||||
Micro is a standard library for microservices.
|
Micro is a standard library for microservices.
|
||||||
|
|
||||||
|
@ -14,6 +14,7 @@ import (
|
|||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
"go.unistack.org/micro/v3/logger"
|
"go.unistack.org/micro/v3/logger"
|
||||||
"go.unistack.org/micro/v3/metadata"
|
"go.unistack.org/micro/v3/metadata"
|
||||||
|
"go.unistack.org/micro/v3/util/buffer"
|
||||||
)
|
)
|
||||||
|
|
||||||
// always first to have proper check
|
// always first to have proper check
|
||||||
@ -30,11 +31,30 @@ func TestStacktrace(t *testing.T) {
|
|||||||
|
|
||||||
l.Error(ctx, "msg1", errors.New("err"))
|
l.Error(ctx, "msg1", errors.New("err"))
|
||||||
|
|
||||||
if !bytes.Contains(buf.Bytes(), []byte(`slog_test.go:31`)) {
|
if !bytes.Contains(buf.Bytes(), []byte(`slog_test.go:32`)) {
|
||||||
t.Fatalf("logger error not works, buf contains: %s", buf.Bytes())
|
t.Fatalf("logger error not works, buf contains: %s", buf.Bytes())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestDelayedBuffer(t *testing.T) {
|
||||||
|
ctx := context.TODO()
|
||||||
|
buf := bytes.NewBuffer(nil)
|
||||||
|
dbuf := buffer.NewDelayedBuffer(100, 100*time.Millisecond, buf)
|
||||||
|
l := NewLogger(logger.WithLevel(logger.ErrorLevel), logger.WithOutput(dbuf),
|
||||||
|
WithHandlerFunc(slog.NewTextHandler),
|
||||||
|
logger.WithAddStacktrace(true),
|
||||||
|
)
|
||||||
|
if err := l.Init(logger.WithFields("key1", "val1")); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
l.Error(ctx, "msg1", errors.New("err"))
|
||||||
|
time.Sleep(120 * time.Millisecond)
|
||||||
|
if !bytes.Contains(buf.Bytes(), []byte(`key1=val1`)) {
|
||||||
|
t.Fatalf("logger delayed buffer not works, buf contains: %s", buf.Bytes())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestTime(t *testing.T) {
|
func TestTime(t *testing.T) {
|
||||||
ctx := context.TODO()
|
ctx := context.TODO()
|
||||||
buf := bytes.NewBuffer(nil)
|
buf := bytes.NewBuffer(nil)
|
||||||
|
@ -149,7 +149,7 @@ func (m *memory) Register(_ context.Context, s *register.Service, opts ...regist
|
|||||||
m.opts.Logger.Debug(m.opts.Context, fmt.Sprintf("Register added new service: %s, version: %s", s.Name, s.Version))
|
m.opts.Logger.Debug(m.opts.Context, fmt.Sprintf("Register added new service: %s, version: %s", s.Name, s.Version))
|
||||||
}
|
}
|
||||||
m.records[options.Namespace] = srvs
|
m.records[options.Namespace] = srvs
|
||||||
go m.sendEvent(®ister.Result{Action: "create", Service: s})
|
go m.sendEvent(®ister.Result{Action: register.EventCreate, Service: s})
|
||||||
}
|
}
|
||||||
|
|
||||||
var addedNodes bool
|
var addedNodes bool
|
||||||
@ -185,7 +185,7 @@ func (m *memory) Register(_ context.Context, s *register.Service, opts ...regist
|
|||||||
if m.opts.Logger.V(logger.DebugLevel) {
|
if m.opts.Logger.V(logger.DebugLevel) {
|
||||||
m.opts.Logger.Debug(m.opts.Context, fmt.Sprintf("Register added new node to service: %s, version: %s", s.Name, s.Version))
|
m.opts.Logger.Debug(m.opts.Context, fmt.Sprintf("Register added new node to service: %s, version: %s", s.Name, s.Version))
|
||||||
}
|
}
|
||||||
go m.sendEvent(®ister.Result{Action: "update", Service: s})
|
go m.sendEvent(®ister.Result{Action: register.EventUpdate, Service: s})
|
||||||
} else {
|
} else {
|
||||||
// refresh TTL and timestamp
|
// refresh TTL and timestamp
|
||||||
for _, n := range s.Nodes {
|
for _, n := range s.Nodes {
|
||||||
@ -238,7 +238,7 @@ func (m *memory) Deregister(ctx context.Context, s *register.Service, opts ...re
|
|||||||
// is cleanup
|
// is cleanup
|
||||||
if len(version.Nodes) > 0 {
|
if len(version.Nodes) > 0 {
|
||||||
m.records[options.Namespace][s.Name][s.Version] = version
|
m.records[options.Namespace][s.Name][s.Version] = version
|
||||||
go m.sendEvent(®ister.Result{Action: "update", Service: s})
|
go m.sendEvent(®ister.Result{Action: register.EventUpdate, Service: s})
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -246,7 +246,7 @@ func (m *memory) Deregister(ctx context.Context, s *register.Service, opts ...re
|
|||||||
// register and exit
|
// register and exit
|
||||||
if len(versions) == 1 {
|
if len(versions) == 1 {
|
||||||
delete(m.records[options.Namespace], s.Name)
|
delete(m.records[options.Namespace], s.Name)
|
||||||
go m.sendEvent(®ister.Result{Action: "delete", Service: s})
|
go m.sendEvent(®ister.Result{Action: register.EventDelete, Service: s})
|
||||||
|
|
||||||
if m.opts.Logger.V(logger.DebugLevel) {
|
if m.opts.Logger.V(logger.DebugLevel) {
|
||||||
m.opts.Logger.Debug(m.opts.Context, fmt.Sprintf("Register removed service: %s", s.Name))
|
m.opts.Logger.Debug(m.opts.Context, fmt.Sprintf("Register removed service: %s", s.Name))
|
||||||
@ -256,7 +256,7 @@ func (m *memory) Deregister(ctx context.Context, s *register.Service, opts ...re
|
|||||||
|
|
||||||
// there are other versions of the service running, so only remove this version of it
|
// there are other versions of the service running, so only remove this version of it
|
||||||
delete(m.records[options.Namespace][s.Name], s.Version)
|
delete(m.records[options.Namespace][s.Name], s.Version)
|
||||||
go m.sendEvent(®ister.Result{Action: "delete", Service: s})
|
go m.sendEvent(®ister.Result{Action: register.EventDelete, Service: s})
|
||||||
if m.opts.Logger.V(logger.DebugLevel) {
|
if m.opts.Logger.V(logger.DebugLevel) {
|
||||||
m.opts.Logger.Debug(m.opts.Context, fmt.Sprintf("Register removed service: %s, version: %s", s.Name, s.Version))
|
m.opts.Logger.Debug(m.opts.Context, fmt.Sprintf("Register removed service: %s, version: %s", s.Name, s.Version))
|
||||||
}
|
}
|
||||||
|
@ -138,8 +138,6 @@ type ListOptions struct {
|
|||||||
Context context.Context
|
Context context.Context
|
||||||
// Namespace to scope the request to
|
// Namespace to scope the request to
|
||||||
Namespace string
|
Namespace string
|
||||||
// Name filter services by name
|
|
||||||
Name string
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewListOptions returns list options filled by opts
|
// NewListOptions returns list options filled by opts
|
||||||
@ -301,13 +299,6 @@ func ListNamespace(d string) ListOption {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// ListName sets the name for list method to filter needed services
|
|
||||||
func ListName(n string) ListOption {
|
|
||||||
return func(o *ListOptions) {
|
|
||||||
o.Name = n
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Name sets the name
|
// Name sets the name
|
||||||
func Name(n string) Option {
|
func Name(n string) Option {
|
||||||
return func(o *Options) {
|
return func(o *Options) {
|
||||||
@ -315,8 +306,6 @@ func Name(n string) Option {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
type codecKey struct{}
|
|
||||||
|
|
||||||
func Codec(c codec.Codec) Option {
|
func Codec(c codec.Codec) Option {
|
||||||
return func(o *Options) {
|
return func(o *Options) {
|
||||||
o.Codec = c
|
o.Codec = c
|
||||||
|
@ -15,31 +15,31 @@ type Watcher interface {
|
|||||||
// the watcher. Actions can be create, update, delete
|
// the watcher. Actions can be create, update, delete
|
||||||
type Result struct {
|
type Result struct {
|
||||||
// Service holds register service
|
// Service holds register service
|
||||||
Service *Service
|
Service *Service `json:"service,omitempty"`
|
||||||
// Action holds the action
|
// Action holds the action
|
||||||
Action string
|
Action EventType `json:"action,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// EventType defines register event type
|
// EventType defines register event type
|
||||||
type EventType int
|
type EventType int
|
||||||
|
|
||||||
const (
|
const (
|
||||||
// Create is emitted when a new service is registered
|
// EventCreate is emitted when a new service is registered
|
||||||
Create EventType = iota
|
EventCreate EventType = iota
|
||||||
// Delete is emitted when an existing service is deregistered
|
// EventDelete is emitted when an existing service is deregistered
|
||||||
Delete
|
EventDelete
|
||||||
// Update is emitted when an existing service is updated
|
// EventUpdate is emitted when an existing service is updated
|
||||||
Update
|
EventUpdate
|
||||||
)
|
)
|
||||||
|
|
||||||
// String returns human readable event type
|
// String returns human readable event type
|
||||||
func (t EventType) String() string {
|
func (t EventType) String() string {
|
||||||
switch t {
|
switch t {
|
||||||
case Create:
|
case EventCreate:
|
||||||
return "create"
|
return "create"
|
||||||
case Delete:
|
case EventDelete:
|
||||||
return "delete"
|
return "delete"
|
||||||
case Update:
|
case EventUpdate:
|
||||||
return "update"
|
return "update"
|
||||||
default:
|
default:
|
||||||
return "unknown"
|
return "unknown"
|
||||||
@ -49,11 +49,11 @@ func (t EventType) String() string {
|
|||||||
// Event is register event
|
// Event is register event
|
||||||
type Event struct {
|
type Event struct {
|
||||||
// Timestamp is event timestamp
|
// Timestamp is event timestamp
|
||||||
Timestamp time.Time
|
Timestamp time.Time `json:"timestamp,omitempty"`
|
||||||
// Service is register service
|
// Service is register service
|
||||||
Service *Service
|
Service *Service `json:"service,omitempty"`
|
||||||
// ID is register id
|
// ID is register id
|
||||||
ID string
|
ID string `json:"id,omitempty"`
|
||||||
// Type defines type of event
|
// Type defines type of event
|
||||||
Type EventType
|
Type EventType `json:"type,omitempty"`
|
||||||
}
|
}
|
||||||
|
@ -1,27 +0,0 @@
|
|||||||
package buf
|
|
||||||
|
|
||||||
import (
|
|
||||||
"bytes"
|
|
||||||
"io"
|
|
||||||
)
|
|
||||||
|
|
||||||
var _ io.Closer = &Buffer{}
|
|
||||||
|
|
||||||
// Buffer bytes.Buffer wrapper to satisfie io.Closer interface
|
|
||||||
type Buffer struct {
|
|
||||||
*bytes.Buffer
|
|
||||||
}
|
|
||||||
|
|
||||||
// Close reset buffer contents
|
|
||||||
func (b *Buffer) Close() error {
|
|
||||||
b.Buffer.Reset()
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// New creates new buffer that satisfies Closer interface
|
|
||||||
func New(b *bytes.Buffer) *Buffer {
|
|
||||||
if b == nil {
|
|
||||||
b = bytes.NewBuffer(nil)
|
|
||||||
}
|
|
||||||
return &Buffer{b}
|
|
||||||
}
|
|
85
util/buffer/buffer.go
Normal file
85
util/buffer/buffer.go
Normal file
@ -0,0 +1,85 @@
|
|||||||
|
package buffer
|
||||||
|
|
||||||
|
import (
|
||||||
|
"io"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
var _ io.WriteCloser = (*DelayedBuffer)(nil)
|
||||||
|
|
||||||
|
// DelayedBuffer is the buffer that holds items until either the buffer filled or a specified time limit is reached
|
||||||
|
type DelayedBuffer struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
maxWait time.Duration
|
||||||
|
flushTime time.Time
|
||||||
|
buffer chan []byte
|
||||||
|
ticker *time.Ticker
|
||||||
|
w io.Writer
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewDelayedBuffer(size int, maxWait time.Duration, w io.Writer) *DelayedBuffer {
|
||||||
|
b := &DelayedBuffer{
|
||||||
|
buffer: make(chan []byte, size),
|
||||||
|
ticker: time.NewTicker(maxWait),
|
||||||
|
w: w,
|
||||||
|
flushTime: time.Now(),
|
||||||
|
maxWait: maxWait,
|
||||||
|
}
|
||||||
|
b.loop()
|
||||||
|
return b
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *DelayedBuffer) loop() {
|
||||||
|
go func() {
|
||||||
|
for range b.ticker.C {
|
||||||
|
b.mu.Lock()
|
||||||
|
if time.Since(b.flushTime) > b.maxWait {
|
||||||
|
b.flush()
|
||||||
|
}
|
||||||
|
b.mu.Unlock()
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *DelayedBuffer) flush() {
|
||||||
|
bufLen := len(b.buffer)
|
||||||
|
if bufLen > 0 {
|
||||||
|
tmp := make([][]byte, bufLen)
|
||||||
|
for i := 0; i < bufLen; i++ {
|
||||||
|
tmp[i] = <-b.buffer
|
||||||
|
}
|
||||||
|
for _, t := range tmp {
|
||||||
|
_, b.err = b.w.Write(t)
|
||||||
|
}
|
||||||
|
b.flushTime = time.Now()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *DelayedBuffer) Put(items ...[]byte) {
|
||||||
|
b.mu.Lock()
|
||||||
|
for _, item := range items {
|
||||||
|
select {
|
||||||
|
case b.buffer <- item:
|
||||||
|
default:
|
||||||
|
b.flush()
|
||||||
|
b.buffer <- item
|
||||||
|
}
|
||||||
|
}
|
||||||
|
b.mu.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *DelayedBuffer) Close() error {
|
||||||
|
b.mu.Lock()
|
||||||
|
b.flush()
|
||||||
|
close(b.buffer)
|
||||||
|
b.ticker.Stop()
|
||||||
|
b.mu.Unlock()
|
||||||
|
return b.err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *DelayedBuffer) Write(data []byte) (int, error) {
|
||||||
|
b.Put(data)
|
||||||
|
return len(data), b.err
|
||||||
|
}
|
22
util/buffer/buffer_test.go
Normal file
22
util/buffer/buffer_test.go
Normal file
@ -0,0 +1,22 @@
|
|||||||
|
package buffer
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestTimedBuffer(t *testing.T) {
|
||||||
|
buf := bytes.NewBuffer(nil)
|
||||||
|
b := NewDelayedBuffer(100, 300*time.Millisecond, buf)
|
||||||
|
for i := 0; i < 100; i++ {
|
||||||
|
_, _ = b.Write([]byte(`test`))
|
||||||
|
}
|
||||||
|
if buf.Len() != 0 {
|
||||||
|
t.Fatal("delayed write not worked")
|
||||||
|
}
|
||||||
|
time.Sleep(400 * time.Millisecond)
|
||||||
|
if buf.Len() == 0 {
|
||||||
|
t.Fatal("delayed write not worked")
|
||||||
|
}
|
||||||
|
}
|
Loading…
x
Reference in New Issue
Block a user