Compare commits
2 Commits
17eefa152f
...
faded4e634
| Author | SHA1 | Date |
|---|---|---|
 熊二
|
faded4e634 | 1 year ago |
|
熊二
|
e8edf6c801 | 1 year ago |
@ -0,0 +1 @@ |
|||||||
|
* text=auto eol=lf |
||||||
@ -1,74 +0,0 @@ |
|||||||
package middleware |
|
||||||
|
|
||||||
import ( |
|
||||||
"github.com/labstack/echo/v4" |
|
||||||
"github.com/labstack/echo/v4/middleware" |
|
||||||
) |
|
||||||
|
|
||||||
// KeyAuthValidator defines a function to validate KeyAuth credentials.
|
|
||||||
type KeyAuthValidator = middleware.KeyAuthValidator |
|
||||||
|
|
||||||
// KeyAuthErrorHandler defines a function which is executed for an invalid key.
|
|
||||||
type KeyAuthErrorHandler = middleware.KeyAuthErrorHandler |
|
||||||
|
|
||||||
// KeyAuthConfig defines the config for KeyAuth middleware.
|
|
||||||
type KeyAuthConfig struct { |
|
||||||
Skipper Skipper |
|
||||||
|
|
||||||
// KeyLookup is a string in the form of "<source>:<name>" or "<source>:<name>,<source>:<name>" that is used
|
|
||||||
// to extract key from the request.
|
|
||||||
// Optional. Default value "header:Authorization".
|
|
||||||
// Possible values:
|
|
||||||
// - "header:<name>" or "header:<name>:<cut-prefix>"
|
|
||||||
// `<cut-prefix>` is argument value to cut/trim prefix of the extracted value. This is useful if header
|
|
||||||
// value has static prefix like `Authorization: <auth-scheme> <authorisation-parameters>` where part that we
|
|
||||||
// want to cut is `<auth-scheme> ` note the space at the end.
|
|
||||||
// In case of basic authentication `Authorization: Basic <credentials>` prefix we want to remove is `Basic `.
|
|
||||||
// - "query:<name>"
|
|
||||||
// - "form:<name>"
|
|
||||||
// - "cookie:<name>"
|
|
||||||
// Multiple sources example:
|
|
||||||
// - "header:Authorization,header:X-Api-Key"
|
|
||||||
KeyLookup string |
|
||||||
|
|
||||||
// AuthScheme to be used in the Authorization header.
|
|
||||||
// Optional. Default value "Bearer".
|
|
||||||
AuthScheme string |
|
||||||
|
|
||||||
// Validator is a function to validate key.
|
|
||||||
// Required.
|
|
||||||
Validator KeyAuthValidator |
|
||||||
|
|
||||||
// ErrorHandler defines a function which is executed for an invalid key.
|
|
||||||
// It may be used to define a custom error.
|
|
||||||
ErrorHandler KeyAuthErrorHandler |
|
||||||
|
|
||||||
// ContinueOnIgnoredError allows the next middleware/handler to be called when ErrorHandler decides to
|
|
||||||
// ignore the error (by returning `nil`).
|
|
||||||
// This is useful when parts of your site/api allow public access and some authorized routes provide extra functionality.
|
|
||||||
// In that case you can use ErrorHandler to set a default public key auth value in the request context
|
|
||||||
// and continue. Some logic down the remaining execution chain needs to check that (public) key auth value then.
|
|
||||||
ContinueOnIgnoredError bool |
|
||||||
} |
|
||||||
|
|
||||||
// DefaultKeyAuthConfig is the default KeyAuth middleware config.
|
|
||||||
var DefaultKeyAuthConfig = KeyAuthConfig{ |
|
||||||
Skipper: DefaultSkipper, |
|
||||||
KeyLookup: "header:" + echo.HeaderAuthorization, |
|
||||||
AuthScheme: "Bearer", |
|
||||||
} |
|
||||||
|
|
||||||
func (a *KeyAuthConfig) ToMiddleware() echo.MiddlewareFunc { |
|
||||||
return middleware.KeyAuthWithConfig(middleware.KeyAuthConfig{ |
|
||||||
Skipper: a.Skipper, |
|
||||||
KeyLookup: a.KeyLookup, |
|
||||||
AuthScheme: a.AuthScheme, |
|
||||||
Validator: a.Validator, |
|
||||||
ErrorHandler: a.ErrorHandler, |
|
||||||
ContinueOnIgnoredError: a.ContinueOnIgnoredError, |
|
||||||
}) |
|
||||||
} |
|
||||||
|
|
||||||
func KeyAuth() echo.MiddlewareFunc { |
|
||||||
return DefaultKeyAuthConfig.ToMiddleware() |
|
||||||
} |
|
||||||
@ -1,24 +0,0 @@ |
|||||||
package middleware |
|
||||||
|
|
||||||
import ( |
|
||||||
"github.com/labstack/echo/v4" |
|
||||||
"github.com/labstack/echo/v4/middleware" |
|
||||||
) |
|
||||||
|
|
||||||
// Skipper defines a function to skip middleware. Returning true skips processing
|
|
||||||
// the middleware.
|
|
||||||
type Skipper = middleware.Skipper |
|
||||||
|
|
||||||
// BeforeFunc defines a function which is executed just before the middleware.
|
|
||||||
type BeforeFunc = middleware.BeforeFunc |
|
||||||
|
|
||||||
type ValuesExtractor = middleware.ValuesExtractor |
|
||||||
|
|
||||||
type ToMiddleware interface { |
|
||||||
ToMiddleware() echo.MiddlewareFunc |
|
||||||
} |
|
||||||
|
|
||||||
// DefaultSkipper returns false which processes the middleware.
|
|
||||||
func DefaultSkipper(echo.Context) bool { |
|
||||||
return false |
|
||||||
} |
|
||||||
@ -1,268 +0,0 @@ |
|||||||
package middleware |
|
||||||
|
|
||||||
import ( |
|
||||||
"github.com/labstack/echo/v4/middleware" |
|
||||||
"net/http" |
|
||||||
"sync" |
|
||||||
"time" |
|
||||||
|
|
||||||
"github.com/labstack/echo/v4" |
|
||||||
"golang.org/x/time/rate" |
|
||||||
) |
|
||||||
|
|
||||||
// RateLimiterStore is the interface to be implemented by custom stores.
|
|
||||||
type RateLimiterStore interface { |
|
||||||
// Allow Stores for the rate limiter have to implement the Allow method
|
|
||||||
Allow(identifier string) (bool, error) |
|
||||||
} |
|
||||||
|
|
||||||
type ( |
|
||||||
// RateLimiterConfig defines the configuration for the rate limiter
|
|
||||||
RateLimiterConfig struct { |
|
||||||
Skipper Skipper |
|
||||||
BeforeFunc middleware.BeforeFunc |
|
||||||
// IdentifierExtractor uses echo.Context to extract the identifier for a visitor
|
|
||||||
IdentifierExtractor Extractor |
|
||||||
// Store defines a store for the rate limiter
|
|
||||||
Store RateLimiterStore |
|
||||||
// ErrorHandler provides a handler to be called when IdentifierExtractor returns an error
|
|
||||||
ErrorHandler func(context echo.Context, err error) error |
|
||||||
// DenyHandler provides a handler to be called when RateLimiter denies access
|
|
||||||
DenyHandler func(context echo.Context, identifier string, err error) error |
|
||||||
} |
|
||||||
// Extractor is used to extract data from echo.Context
|
|
||||||
Extractor func(context echo.Context) (string, error) |
|
||||||
) |
|
||||||
|
|
||||||
// errors
|
|
||||||
var ( |
|
||||||
// ErrRateLimitExceeded denotes an error raised when rate limit is exceeded
|
|
||||||
ErrRateLimitExceeded = echo.NewHTTPError(http.StatusTooManyRequests, "rate limit exceeded") |
|
||||||
// ErrExtractorError denotes an error raised when extractor function is unsuccessful
|
|
||||||
ErrExtractorError = echo.NewHTTPError(http.StatusForbidden, "error while extracting identifier") |
|
||||||
) |
|
||||||
|
|
||||||
// DefaultRateLimiterConfig defines default values for RateLimiterConfig
|
|
||||||
var DefaultRateLimiterConfig = RateLimiterConfig{ |
|
||||||
Skipper: middleware.DefaultSkipper, |
|
||||||
IdentifierExtractor: func(ctx echo.Context) (string, error) { |
|
||||||
id := ctx.RealIP() |
|
||||||
return id, nil |
|
||||||
}, |
|
||||||
ErrorHandler: func(context echo.Context, err error) error { |
|
||||||
return &echo.HTTPError{ |
|
||||||
Code: ErrExtractorError.Code, |
|
||||||
Message: ErrExtractorError.Message, |
|
||||||
Internal: err, |
|
||||||
} |
|
||||||
}, |
|
||||||
DenyHandler: func(context echo.Context, identifier string, err error) error { |
|
||||||
return &echo.HTTPError{ |
|
||||||
Code: ErrRateLimitExceeded.Code, |
|
||||||
Message: ErrRateLimitExceeded.Message, |
|
||||||
Internal: err, |
|
||||||
} |
|
||||||
}, |
|
||||||
} |
|
||||||
|
|
||||||
/* |
|
||||||
RateLimiter returns a rate limiting middleware |
|
||||||
|
|
||||||
e := echo.New() |
|
||||||
|
|
||||||
limiterStore := middleware.NewRateLimiterMemoryStore(20) |
|
||||||
|
|
||||||
e.GET("/rate-limited", func(c echo.Context) error { |
|
||||||
return c.String(http.StatusOK, "test") |
|
||||||
}, RateLimiter(limiterStore)) |
|
||||||
*/ |
|
||||||
func RateLimiter(store RateLimiterStore) echo.MiddlewareFunc { |
|
||||||
config := DefaultRateLimiterConfig |
|
||||||
config.Store = store |
|
||||||
return config.ToMiddleware() |
|
||||||
} |
|
||||||
|
|
||||||
/* |
|
||||||
ToMiddleware returns a rate limiting middleware |
|
||||||
|
|
||||||
e := echo.New() |
|
||||||
|
|
||||||
config := middleware.RateLimiterConfig{ |
|
||||||
Skipper: DefaultSkipper, |
|
||||||
Store: middleware.NewRateLimiterMemoryStore( |
|
||||||
middleware.RateLimiterMemoryStoreConfig{Rate: 10, Burst: 30, ExpiresIn: 3 * time.Minute} |
|
||||||
) |
|
||||||
IdentifierExtractor: func(ctx echo.Context) (string, error) { |
|
||||||
id := ctx.RealIP() |
|
||||||
return id, nil |
|
||||||
}, |
|
||||||
ErrorHandler: func(context echo.Context, err error) error { |
|
||||||
return context.JSON(http.StatusTooManyRequests, nil) |
|
||||||
}, |
|
||||||
DenyHandler: func(context echo.Context, identifier string) error { |
|
||||||
return context.JSON(http.StatusForbidden, nil) |
|
||||||
}, |
|
||||||
} |
|
||||||
|
|
||||||
e.GET("/rate-limited", func(c echo.Context) error { |
|
||||||
return c.String(http.StatusOK, "test") |
|
||||||
}, middleware.RateLimiterWithConfig(config)) |
|
||||||
*/ |
|
||||||
func (config *RateLimiterConfig) ToMiddleware() echo.MiddlewareFunc { |
|
||||||
if config.Skipper == nil { |
|
||||||
config.Skipper = DefaultSkipper |
|
||||||
} |
|
||||||
if config.IdentifierExtractor == nil { |
|
||||||
config.IdentifierExtractor = DefaultRateLimiterConfig.IdentifierExtractor |
|
||||||
} |
|
||||||
if config.ErrorHandler == nil { |
|
||||||
config.ErrorHandler = DefaultRateLimiterConfig.ErrorHandler |
|
||||||
} |
|
||||||
if config.DenyHandler == nil { |
|
||||||
config.DenyHandler = DefaultRateLimiterConfig.DenyHandler |
|
||||||
} |
|
||||||
if config.Store == nil { |
|
||||||
panic("Store configuration must be provided") |
|
||||||
} |
|
||||||
return func(next echo.HandlerFunc) echo.HandlerFunc { |
|
||||||
return func(c echo.Context) error { |
|
||||||
if config.Skipper(c) { |
|
||||||
return next(c) |
|
||||||
} |
|
||||||
if config.BeforeFunc != nil { |
|
||||||
config.BeforeFunc(c) |
|
||||||
} |
|
||||||
|
|
||||||
identifier, err := config.IdentifierExtractor(c) |
|
||||||
if err != nil { |
|
||||||
c.Error(config.ErrorHandler(c, err)) |
|
||||||
return nil |
|
||||||
} |
|
||||||
|
|
||||||
if allow, err := config.Store.Allow(identifier); !allow { |
|
||||||
c.Error(config.DenyHandler(c, identifier, err)) |
|
||||||
return nil |
|
||||||
} |
|
||||||
return next(c) |
|
||||||
} |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
type ( |
|
||||||
// RateLimiterMemoryStore is the built-in store implementation for RateLimiter
|
|
||||||
RateLimiterMemoryStore struct { |
|
||||||
visitors map[string]*Visitor |
|
||||||
mutex sync.Mutex |
|
||||||
rate rate.Limit // for more info check out Limiter docs - https://pkg.go.dev/golang.org/x/time/rate#Limit.
|
|
||||||
|
|
||||||
burst int |
|
||||||
expiresIn time.Duration |
|
||||||
lastCleanup time.Time |
|
||||||
|
|
||||||
timeNow func() time.Time |
|
||||||
} |
|
||||||
// Visitor signifies a unique user's limiter details
|
|
||||||
Visitor struct { |
|
||||||
*rate.Limiter |
|
||||||
lastSeen time.Time |
|
||||||
} |
|
||||||
) |
|
||||||
|
|
||||||
/* |
|
||||||
NewRateLimiterMemoryStore returns an instance of RateLimiterMemoryStore with |
|
||||||
the provided rate (as req/s). |
|
||||||
for more info check out Limiter docs - https://pkg.go.dev/golang.org/x/time/rate#Limit.
|
|
||||||
|
|
||||||
Burst and ExpiresIn will be set to default values. |
|
||||||
|
|
||||||
Note that if the provided rate is a float number and Burst is zero, Burst will be treated as the rounded down value of the rate. |
|
||||||
|
|
||||||
Example (with 20 requests/sec): |
|
||||||
|
|
||||||
limiterStore := middleware.NewRateLimiterMemoryStore(20) |
|
||||||
*/ |
|
||||||
func NewRateLimiterMemoryStore(rate rate.Limit) (store *RateLimiterMemoryStore) { |
|
||||||
return NewRateLimiterMemoryStoreWithConfig(RateLimiterMemoryStoreConfig{ |
|
||||||
Rate: rate, |
|
||||||
}) |
|
||||||
} |
|
||||||
|
|
||||||
/* |
|
||||||
NewRateLimiterMemoryStoreWithConfig returns an instance of RateLimiterMemoryStore |
|
||||||
with the provided configuration. Rate must be provided. Burst will be set to the rounded down value of |
|
||||||
the configured rate if not provided or set to 0. |
|
||||||
|
|
||||||
The build-in memory store is usually capable for modest loads. For higher loads other |
|
||||||
store implementations should be considered. |
|
||||||
|
|
||||||
Characteristics: |
|
||||||
* Concurrency above 100 parallel requests may causes measurable lock contention |
|
||||||
* A high number of different IP addresses (above 16000) may be impacted by the internally used Go map |
|
||||||
* A high number of requests from a single IP address may cause lock contention |
|
||||||
|
|
||||||
Example: |
|
||||||
|
|
||||||
limiterStore := middleware.NewRateLimiterMemoryStoreWithConfig( |
|
||||||
middleware.RateLimiterMemoryStoreConfig{Rate: 50, Burst: 200, ExpiresIn: 5 * time.Minute}, |
|
||||||
) |
|
||||||
*/ |
|
||||||
func NewRateLimiterMemoryStoreWithConfig(config RateLimiterMemoryStoreConfig) (store *RateLimiterMemoryStore) { |
|
||||||
store = &RateLimiterMemoryStore{} |
|
||||||
|
|
||||||
store.rate = config.Rate |
|
||||||
store.burst = config.Burst |
|
||||||
store.expiresIn = config.ExpiresIn |
|
||||||
if config.ExpiresIn == 0 { |
|
||||||
store.expiresIn = DefaultRateLimiterMemoryStoreConfig.ExpiresIn |
|
||||||
} |
|
||||||
if config.Burst == 0 { |
|
||||||
store.burst = int(config.Rate) |
|
||||||
} |
|
||||||
store.visitors = make(map[string]*Visitor) |
|
||||||
store.timeNow = time.Now |
|
||||||
store.lastCleanup = store.timeNow() |
|
||||||
return |
|
||||||
} |
|
||||||
|
|
||||||
// RateLimiterMemoryStoreConfig represents configuration for RateLimiterMemoryStore
|
|
||||||
type RateLimiterMemoryStoreConfig struct { |
|
||||||
Rate rate.Limit // Rate of requests allowed to pass as req/s. For more info check out Limiter docs - https://pkg.go.dev/golang.org/x/time/rate#Limit.
|
|
||||||
Burst int // Burst is maximum number of requests to pass at the same moment. It additionally allows a number of requests to pass when rate limit is reached.
|
|
||||||
ExpiresIn time.Duration // ExpiresIn is the duration after that a rate limiter is cleaned up
|
|
||||||
} |
|
||||||
|
|
||||||
// DefaultRateLimiterMemoryStoreConfig provides default configuration values for RateLimiterMemoryStore
|
|
||||||
var DefaultRateLimiterMemoryStoreConfig = RateLimiterMemoryStoreConfig{ |
|
||||||
ExpiresIn: 3 * time.Minute, |
|
||||||
} |
|
||||||
|
|
||||||
// Allow implements RateLimiterStore.Allow
|
|
||||||
func (store *RateLimiterMemoryStore) Allow(identifier string) (bool, error) { |
|
||||||
store.mutex.Lock() |
|
||||||
limiter, exists := store.visitors[identifier] |
|
||||||
if !exists { |
|
||||||
limiter = new(Visitor) |
|
||||||
limiter.Limiter = rate.NewLimiter(store.rate, store.burst) |
|
||||||
store.visitors[identifier] = limiter |
|
||||||
} |
|
||||||
now := store.timeNow() |
|
||||||
limiter.lastSeen = now |
|
||||||
if now.Sub(store.lastCleanup) > store.expiresIn { |
|
||||||
store.cleanupStaleVisitors() |
|
||||||
} |
|
||||||
store.mutex.Unlock() |
|
||||||
return limiter.AllowN(store.timeNow(), 1), nil |
|
||||||
} |
|
||||||
|
|
||||||
/* |
|
||||||
cleanupStaleVisitors helps manage the size of the visitors map by removing stale records |
|
||||||
of users who haven't visited again after the configured expiry time has elapsed |
|
||||||
*/ |
|
||||||
func (store *RateLimiterMemoryStore) cleanupStaleVisitors() { |
|
||||||
for id, visitor := range store.visitors { |
|
||||||
if store.timeNow().Sub(visitor.lastSeen) > store.expiresIn { |
|
||||||
delete(store.visitors, id) |
|
||||||
} |
|
||||||
} |
|
||||||
store.lastCleanup = store.timeNow() |
|
||||||
} |
|
||||||
@ -0,0 +1,108 @@ |
|||||||
|
package runtime |
||||||
|
|
||||||
|
import ( |
||||||
|
"context" |
||||||
|
"sorbet/pkg/log" |
||||||
|
"sync" |
||||||
|
) |
||||||
|
|
||||||
|
var ( |
||||||
|
// 并发组控制器
|
||||||
|
wg sync.WaitGroup |
||||||
|
// 运行时上下文对象
|
||||||
|
ctx context.Context |
||||||
|
// 信号量，控制任务并发数量
|
||||||
|
semaphore chan struct{} |
||||||
|
// 异步任务通道
|
||||||
|
tasks chan func(context.Context) |
||||||
|
// 上下文取消函数
|
||||||
|
cancel context.CancelFunc |
||||||
|
// 程序退出通道
|
||||||
|
exit chan chan error |
||||||
|
) |
||||||
|
|
||||||
|
func tryCancel() { |
||||||
|
cancelFunc := cancel |
||||||
|
if cancelFunc == nil { |
||||||
|
log.Panic("context already cancelled") |
||||||
|
} |
||||||
|
cancel = nil |
||||||
|
cancelFunc() |
||||||
|
// 等待任务执行完成
|
||||||
|
for len(semaphore) > 0 { |
||||||
|
work(nil) |
||||||
|
} |
||||||
|
// 理论上在没有信号量的情况下，是没有任务队列的。
|
||||||
|
for len(tasks) > 0 { |
||||||
|
log.Panic("nonempty task channel") |
||||||
|
} |
||||||
|
// 销毁注册的服务
|
||||||
|
for i := len(servlets) - 1; i >= 0; i-- { |
||||||
|
if err := servlets[i].Stop(); err != nil { |
||||||
|
log.Warn("servlet Stop returned with error: ", err) |
||||||
|
} |
||||||
|
} |
||||||
|
ctx = nil |
||||||
|
} |
||||||
|
|
||||||
|
// Go 异步任务
|
||||||
|
func Go(fn func(context.Context)) { |
||||||
|
if semaphore == nil || ctx == nil { |
||||||
|
log.Panic("not initialized") |
||||||
|
return |
||||||
|
} |
||||||
|
select { |
||||||
|
case semaphore <- struct{}{}: |
||||||
|
// If we are below our limit, spawn a new worker rather
|
||||||
|
// than waiting for one to become available.
|
||||||
|
async(&wg, func(context.Context) { |
||||||
|
work(fn) |
||||||
|
}) |
||||||
|
case tasks <- fn: |
||||||
|
// 放到任务队列中
|
||||||
|
// A worker is available and has accepted the task.
|
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
// Async 异步执行函数
|
||||||
|
func Async(fn func(context.Context)) { |
||||||
|
async(&wg, fn) |
||||||
|
} |
||||||
|
|
||||||
|
// 执行任务
|
||||||
|
func work(f func(context.Context)) { |
||||||
|
defer func() { |
||||||
|
if semaphore != nil { |
||||||
|
// 释放
|
||||||
|
<-semaphore |
||||||
|
} |
||||||
|
}() |
||||||
|
|
||||||
|
// todo 能不能复用 sync.WaitGroup
|
||||||
|
g := &sync.WaitGroup{} |
||||||
|
if f != nil { |
||||||
|
Async(f) |
||||||
|
} |
||||||
|
for task := range tasks { |
||||||
|
Async(task) |
||||||
|
} |
||||||
|
g.Wait() |
||||||
|
} |
||||||
|
|
||||||
|
// 异步执行函数
|
||||||
|
func async(wg *sync.WaitGroup, fn func(context.Context)) { |
||||||
|
wg.Add(1) |
||||||
|
go func() { |
||||||
|
defer wg.Done() |
||||||
|
if ctx == nil { |
||||||
|
return |
||||||
|
} |
||||||
|
select { |
||||||
|
case <-ctx.Done(): |
||||||
|
// 上下文一旦结束，任务将被忽略
|
||||||
|
log.Error("context done") |
||||||
|
default: |
||||||
|
fn(ctx) |
||||||
|
} |
||||||
|
}() |
||||||
|
} |
||||||
@ -0,0 +1,90 @@ |
|||||||
|
package runtime |
||||||
|
|
||||||
|
import ( |
||||||
|
"context" |
||||||
|
"errors" |
||||||
|
"os" |
||||||
|
"os/signal" |
||||||
|
"sorbet/pkg/log" |
||||||
|
"syscall" |
||||||
|
) |
||||||
|
|
||||||
|
// Init 初始化
|
||||||
|
func Init() error { |
||||||
|
return InitWith(context.Background()) |
||||||
|
} |
||||||
|
|
||||||
|
func InitWith(root context.Context) error { |
||||||
|
if ctx != nil || cancel != nil { |
||||||
|
return errors.New("not stopped") |
||||||
|
} |
||||||
|
// 初始化上下文相关
|
||||||
|
ctx, cancel = context.WithCancel(root) |
||||||
|
// 初始化退出通道
|
||||||
|
if exit == nil { |
||||||
|
exit = make(chan chan error) |
||||||
|
} |
||||||
|
// 初始化信号量
|
||||||
|
if semaphore == nil { |
||||||
|
semaphore = make(chan struct{}, 256) |
||||||
|
} |
||||||
|
// 初始化任务通道
|
||||||
|
if tasks != nil { |
||||||
|
tasks = make(chan func(context.Context), 256) |
||||||
|
} |
||||||
|
// 初始化网络服务组件
|
||||||
|
for _, servlet := range servlets { |
||||||
|
if err := servlet.Init(ctx); err != nil { |
||||||
|
return err |
||||||
|
} |
||||||
|
} |
||||||
|
return nil |
||||||
|
} |
||||||
|
|
||||||
|
func Start() error { |
||||||
|
// 创建服务器
|
||||||
|
srv, ln, err := createServer() |
||||||
|
if err != nil { |
||||||
|
return err |
||||||
|
} |
||||||
|
|
||||||
|
// 启动网络服务器
|
||||||
|
go func() { |
||||||
|
err := srv.Serve(ln) |
||||||
|
if err != nil { |
||||||
|
log.Error("encountered an error while serving listener: ", err) |
||||||
|
} |
||||||
|
}() |
||||||
|
log.Info("Listening on %s", ln.Addr().String()) |
||||||
|
|
||||||
|
// 监听停止命令，停止网络服务
|
||||||
|
go func() { |
||||||
|
errChan := <-exit |
||||||
|
tryCancel() |
||||||
|
errChan <- ln.Close() // stop the listener
|
||||||
|
}() |
||||||
|
|
||||||
|
return nil |
||||||
|
} |
||||||
|
|
||||||
|
func Run() error { |
||||||
|
if err := Start(); err != nil { |
||||||
|
return err |
||||||
|
} |
||||||
|
// parse address for host, port
|
||||||
|
ch := make(chan os.Signal, 1) |
||||||
|
signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT) |
||||||
|
log.Info("Received signal %s", <-ch) |
||||||
|
return Stop() |
||||||
|
} |
||||||
|
|
||||||
|
// Stop 停止运行
|
||||||
|
func Stop() error { |
||||||
|
if ctx == nil { |
||||||
|
return errors.New("already stopped") |
||||||
|
} |
||||||
|
ch := make(chan error) |
||||||
|
exit <- ch |
||||||
|
wg.Wait() |
||||||
|
return <-ch |
||||||
|
} |
||||||
@ -0,0 +1,121 @@ |
|||||||
|
package runtime |
||||||
|
|
||||||
|
import ( |
||||||
|
"fmt" |
||||||
|
"github.com/labstack/echo/v4" |
||||||
|
"github.com/labstack/echo/v4/middleware" |
||||||
|
"net" |
||||||
|
"net/http" |
||||||
|
"sorbet/internal/util" |
||||||
|
"sorbet/pkg/env" |
||||||
|
"sorbet/pkg/rsp" |
||||||
|
"time" |
||||||
|
) |
||||||
|
|
||||||
|
var ( |
||||||
|
// maxHeaderBytes is used by the http server to limit the size of request headers.
|
||||||
|
// This may need to be increased if accepting cookies from the public.
|
||||||
|
maxHeaderBytes = 1 << 20 |
||||||
|
// readTimeout is used by the http server to set a maximum duration before
|
||||||
|
// timing out read of the request. The default timeout is 10 seconds.
|
||||||
|
readTimeout = 10 * time.Second |
||||||
|
// writeTimeout is used by the http server to set a maximum duration before
|
||||||
|
// timing out write of the response. The default timeout is 10 seconds.
|
||||||
|
writeTimeout = 10 * time.Second |
||||||
|
// idleTimeout is used by the http server to set a maximum duration for
|
||||||
|
// keep-alive connections.
|
||||||
|
idleTimeout = 120 * time.Second |
||||||
|
) |
||||||
|
|
||||||
|
func newEchoFramework() (*echo.Echo, error) { |
||||||
|
e := echo.New() |
||||||
|
e.HideBanner = true |
||||||
|
e.HidePort = true |
||||||
|
e.HTTPErrorHandler = rsp.HTTPErrorHandler |
||||||
|
e.Debug = !env.IsEnv("prod") |
||||||
|
e.Logger = util.NewEchoLogger() |
||||||
|
e.Use(middleware.Recover()) |
||||||
|
e.Use(middleware.CORS()) |
||||||
|
e.Use(middleware.Logger()) |
||||||
|
for _, servlet := range servlets { |
||||||
|
group := e.Group("") |
||||||
|
for _, routable := range servlet.Routes() { |
||||||
|
routable.InitRoutes(group) |
||||||
|
} |
||||||
|
} |
||||||
|
routes := e.Routes() |
||||||
|
e.GET("/_routes", func(c echo.Context) error { |
||||||
|
return c.JSON(http.StatusOK, routes) |
||||||
|
}) |
||||||
|
return e, nil |
||||||
|
} |
||||||
|
|
||||||
|
func createServer() (*http.Server, net.Listener, error) { |
||||||
|
e, err := newEchoFramework() |
||||||
|
if err != nil { |
||||||
|
return nil, nil, err |
||||||
|
} |
||||||
|
|
||||||
|
l, err := createTCPListener() |
||||||
|
if err != nil { |
||||||
|
return nil, nil, err |
||||||
|
} |
||||||
|
|
||||||
|
s := &http.Server{ |
||||||
|
Handler: e, |
||||||
|
MaxHeaderBytes: env.Int("SERVER_MAX_HEADER_BYTES", maxHeaderBytes), |
||||||
|
ReadTimeout: env.Duration("SERVER_READ_TIMEOUT", readTimeout), |
||||||
|
WriteTimeout: env.Duration("SERVER_WRITE_TIMEOUT", writeTimeout), |
||||||
|
IdleTimeout: env.Duration("SERVER_IDLE_TIMEOUT", idleTimeout), |
||||||
|
} |
||||||
|
|
||||||
|
return s, l, nil |
||||||
|
} |
||||||
|
|
||||||
|
func createTCPListener() (net.Listener, error) { |
||||||
|
l, err := net.Listen( |
||||||
|
env.String("SERVER_NETWORK", "tcp"), |
||||||
|
fmt.Sprintf("%s:%d", |
||||||
|
env.String("SERVER_ADDRESS", "0.0.0.0"), |
||||||
|
env.Int("SERVER_PORT", 1324), |
||||||
|
), |
||||||
|
) |
||||||
|
if err == nil { |
||||||
|
l = net.Listener(TCPKeepAliveListener{ |
||||||
|
TCPListener: l.(*net.TCPListener), |
||||||
|
}) |
||||||
|
} |
||||||
|
return l, err |
||||||
|
} |
||||||
|
|
||||||
|
// TCPKeepAliveListener sets TCP keep-alive timeouts on accepted
|
||||||
|
// connections. It's used by ListenAndServe and ListenAndServeTLS so
|
||||||
|
// dead TCP connections (e.g. closing laptop mid-download) eventually
|
||||||
|
// go away.
|
||||||
|
//
|
||||||
|
// This is here because it is not exposed in the stdlib and
|
||||||
|
// we'd prefer to have a hold of the http.Server's net.Listener so we can close it
|
||||||
|
// on shutdown.
|
||||||
|
//
|
||||||
|
// Taken from here: https://golang.org/src/net/http/server.go?s=63121:63175#L2120
|
||||||
|
type TCPKeepAliveListener struct { |
||||||
|
*net.TCPListener |
||||||
|
} |
||||||
|
|
||||||
|
// Accept accepts the next incoming call and returns the new
|
||||||
|
// connection. KeepAlivePeriod is set properly.
|
||||||
|
func (ln TCPKeepAliveListener) Accept() (c net.Conn, err error) { |
||||||
|
tc, err := ln.AcceptTCP() |
||||||
|
if err != nil { |
||||||
|
return |
||||||
|
} |
||||||
|
err = tc.SetKeepAlive(true) |
||||||
|
if err != nil { |
||||||
|
return |
||||||
|
} |
||||||
|
err = tc.SetKeepAlivePeriod(3 * time.Minute) |
||||||
|
if err != nil { |
||||||
|
return |
||||||
|
} |
||||||
|
return tc, nil |
||||||
|
} |
||||||
@ -0,0 +1,69 @@ |
|||||||
|
package runtime |
||||||
|
|
||||||
|
import ( |
||||||
|
"context" |
||||||
|
"errors" |
||||||
|
"github.com/labstack/echo/v4" |
||||||
|
"slices" |
||||||
|
) |
||||||
|
|
||||||
|
// 注册的网络服务组件
|
||||||
|
var servlets []Servlet |
||||||
|
|
||||||
|
// Servlet 网络服务组件接口
|
||||||
|
type Servlet interface { |
||||||
|
// Name 服务名称
|
||||||
|
Name() string |
||||||
|
// Priority 优先级，用于启动和销毁的执行顺序
|
||||||
|
Priority() int |
||||||
|
// Init 初始化服务
|
||||||
|
Init(ctx context.Context) error |
||||||
|
// Routes 注册路由
|
||||||
|
Routes() []Routable |
||||||
|
// Start 非阻塞方式启动服务
|
||||||
|
Start() error |
||||||
|
// Stop 停止服务
|
||||||
|
Stop() error |
||||||
|
} |
||||||
|
|
||||||
|
func Reset() error { |
||||||
|
if len(servlets) == 0 { |
||||||
|
return nil |
||||||
|
} |
||||||
|
if ctx == nil { |
||||||
|
return errors.New("servlets is running") |
||||||
|
} |
||||||
|
servlets = nil |
||||||
|
return nil |
||||||
|
} |
||||||
|
|
||||||
|
// Use 注册服务
|
||||||
|
func Use(servlets ...Servlet) error { |
||||||
|
if len(servlets) == 0 { |
||||||
|
return nil |
||||||
|
} |
||||||
|
for i := 0; i < len(servlets); i++ { |
||||||
|
if !use(servlets[i]) { |
||||||
|
return errors.New("service already registered") |
||||||
|
} |
||||||
|
} |
||||||
|
slices.SortFunc(servlets, func(a, b Servlet) int { |
||||||
|
return b.Priority() - a.Priority() // 按优先级排序
|
||||||
|
}) |
||||||
|
return nil |
||||||
|
} |
||||||
|
|
||||||
|
func use(servlet Servlet) bool { |
||||||
|
exists := slices.ContainsFunc(servlets, func(s Servlet) bool { |
||||||
|
return s.Name() == servlet.Name() |
||||||
|
}) |
||||||
|
if !exists { |
||||||
|
servlets = append(servlets, servlet) |
||||||
|
return true |
||||||
|
} |
||||||
|
return false |
||||||
|
} |
||||||
|
|
||||||
|
type Routable interface { |
||||||
|
InitRoutes(*echo.Group) |
||||||
|
} |
||||||
@ -1,72 +0,0 @@ |
|||||||
package services |
|
||||||
|
|
||||||
import ( |
|
||||||
"context" |
|
||||||
"errors" |
|
||||||
"github.com/labstack/echo/v4" |
|
||||||
"slices" |
|
||||||
"sorbet/internal/services/company" |
|
||||||
"sorbet/internal/services/config" |
|
||||||
"sorbet/internal/services/feature" |
|
||||||
"sorbet/internal/services/resource" |
|
||||||
"sorbet/internal/services/system" |
|
||||||
"sorbet/pkg/crud" |
|
||||||
) |
|
||||||
|
|
||||||
var services []crud.Service |
|
||||||
|
|
||||||
const ContextEchoKey = "echo_framework" |
|
||||||
|
|
||||||
func Register(service crud.Service) error { |
|
||||||
for _, applet := range services { |
|
||||||
if applet.Name() == service.Name() { |
|
||||||
return errors.New("service already registered") |
|
||||||
} |
|
||||||
} |
|
||||||
services = append(services, service) |
|
||||||
return nil |
|
||||||
} |
|
||||||
|
|
||||||
func Init(ctx context.Context) error { |
|
||||||
services = []crud.Service{ |
|
||||||
&config.Service{}, |
|
||||||
&company.Service{}, |
|
||||||
&resource.Service{}, |
|
||||||
&feature.Service{}, |
|
||||||
&system.Service{}, |
|
||||||
} |
|
||||||
|
|
||||||
// 按优先级排序
|
|
||||||
slices.SortFunc(services, func(a, b crud.Service) int { |
|
||||||
return b.Priority() - a.Priority() |
|
||||||
}) |
|
||||||
|
|
||||||
return nil |
|
||||||
} |
|
||||||
|
|
||||||
func Bootstrap(ctx context.Context) error { |
|
||||||
e := ctx.Value(ContextEchoKey).(*echo.Echo) |
|
||||||
for _, service := range services { |
|
||||||
err := service.Init(crud.NewContext(ctx, e.Group(""))) |
|
||||||
if err != nil { |
|
||||||
return err |
|
||||||
} |
|
||||||
err = service.Bootstrap() |
|
||||||
if err != nil { |
|
||||||
return err |
|
||||||
} |
|
||||||
} |
|
||||||
return nil |
|
||||||
} |
|
||||||
|
|
||||||
func Destroy() error { |
|
||||||
for i := len(services) - 1; i >= 0; i++ { |
|
||||||
service := services[i] |
|
||||||
err := service.Destroy() |
|
||||||
if err != nil { |
|
||||||
// TODO(hupeh): 是否需要销毁策略，比如可以继续或者中断等行为
|
|
||||||
return err |
|
||||||
} |
|
||||||
} |
|
||||||
return nil |
|
||||||
} |
|
||||||
@ -1,128 +0,0 @@ |
|||||||
package app |
|
||||||
|
|
||||||
import ( |
|
||||||
"context" |
|
||||||
"errors" |
|
||||||
"os" |
|
||||||
"sorbet/pkg/app/appkit" |
|
||||||
) |
|
||||||
|
|
||||||
func Init(args ...string) (err error) { |
|
||||||
if len(args) == 0 { |
|
||||||
args = os.Args[1:] |
|
||||||
} |
|
||||||
|
|
||||||
switch Status() { |
|
||||||
case Initialized: |
|
||||||
return nil |
|
||||||
case Starting: |
|
||||||
return errors.New("starting") |
|
||||||
case Running: |
|
||||||
return errors.New("running") |
|
||||||
case Stopping: |
|
||||||
return errors.New("stopping") |
|
||||||
case Idle, Stopped: |
|
||||||
// continue
|
|
||||||
} |
|
||||||
|
|
||||||
initLifecycle() |
|
||||||
initStats(args) |
|
||||||
initUdp(args) |
|
||||||
initBus() |
|
||||||
initBuiltins() |
|
||||||
|
|
||||||
setStatus(Initialized) |
|
||||||
|
|
||||||
return |
|
||||||
} |
|
||||||
|
|
||||||
func initBuiltins() { |
|
||||||
// 订阅应用启动
|
|
||||||
var sub appkit.Subscriber |
|
||||||
sub = Sub("start", func([]byte) []byte { |
|
||||||
sub.Cancel() |
|
||||||
return nil |
|
||||||
}) |
|
||||||
|
|
||||||
// 订阅应用停止
|
|
||||||
Sub("stop", func(bytes []byte) []byte { |
|
||||||
quit := exit |
|
||||||
if quit != nil { |
|
||||||
exit = nil |
|
||||||
quit() |
|
||||||
} |
|
||||||
return nil |
|
||||||
}) |
|
||||||
} |
|
||||||
|
|
||||||
func Loop() error { |
|
||||||
switch Status() { |
|
||||||
case Starting, Running: |
|
||||||
return nil |
|
||||||
case Stopping: |
|
||||||
return errors.New("stopping") |
|
||||||
case Idle: |
|
||||||
return errors.New("idle, you maybe forgot call app.Init()") |
|
||||||
case Stopped: |
|
||||||
return errors.New("stopped, you maybe forgot call app.Init()") |
|
||||||
case Initialized: |
|
||||||
// nothing
|
|
||||||
} |
|
||||||
|
|
||||||
// 释放内存
|
|
||||||
defer free() |
|
||||||
|
|
||||||
wg.Add(1) |
|
||||||
setStatus(Starting) |
|
||||||
|
|
||||||
// In a goroutine, we wait on for all goroutines to complete (for example
|
|
||||||
// timers). We use this to signal to the main thread to exit.
|
|
||||||
// wg.Add(1) basically translates to uv_ref, if this was Node.
|
|
||||||
// wg.Done() basically translates to uv_unref
|
|
||||||
go func() { |
|
||||||
wg.Wait() |
|
||||||
setStatus(Stopping) |
|
||||||
stop() |
|
||||||
}() |
|
||||||
|
|
||||||
// 启动应用指令
|
|
||||||
Pub("start", nil) |
|
||||||
|
|
||||||
for { |
|
||||||
select { |
|
||||||
case msg := <-pubs: |
|
||||||
err := dispatch(msg) |
|
||||||
exitOnError(err) |
|
||||||
wg.Done() // Corresponds to the wg.Add(1) in Pub().
|
|
||||||
|
|
||||||
case <-ctx.Done(): |
|
||||||
// 只有等待所有协程全部完成后，
|
|
||||||
// 然后才可以退出主循环
|
|
||||||
checkPubsEmpty() |
|
||||||
setStatus(Stopped) |
|
||||||
err := ctx.Err() |
|
||||||
if errors.Is(err, context.Canceled) { |
|
||||||
return nil |
|
||||||
} |
|
||||||
return err |
|
||||||
} |
|
||||||
|
|
||||||
// We don't want to exit until we've received at least one message.
|
|
||||||
// This is so the program doesn't exit after sending the "start"
|
|
||||||
// message.
|
|
||||||
if Status() == Starting { |
|
||||||
wg.Done() |
|
||||||
start() |
|
||||||
setStatus(Running) |
|
||||||
} |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
func Stop() { |
|
||||||
Pub("stop", nil) |
|
||||||
} |
|
||||||
|
|
||||||
func Go(fn func(context.Context)) { |
|
||||||
assert(fn != nil, "invalid fn") |
|
||||||
async(fn) |
|
||||||
} |
|
||||||
@ -1,5 +0,0 @@ |
|||||||
package appkit |
|
||||||
|
|
||||||
type Subscriber interface { |
|
||||||
Cancel() |
|
||||||
} |
|
||||||
@ -1,81 +0,0 @@ |
|||||||
package appkit |
|
||||||
|
|
||||||
import ( |
|
||||||
"bytes" |
|
||||||
"encoding/binary" |
|
||||||
"errors" |
|
||||||
"math" |
|
||||||
) |
|
||||||
|
|
||||||
// 数据协议如下：
|
|
||||||
//
|
|
||||||
// +-----------+--------+--------------+-------+------+
|
|
||||||
// | 0x01,0x02 | length | topic-length | topic | data |
|
|
||||||
// +-----------+--------+--------------+-------+------+
|
|
||||||
// 2 2 2 n n
|
|
||||||
//
|
|
||||||
// * 头部有两个模数 0x01、0x02；
|
|
||||||
// * length 表示消息的总长度；
|
|
||||||
// * 然后分别是 topic、data，包含其长度和数据；
|
|
||||||
// * 长度使用 uint16 类型，所以占用 2 个字节；
|
|
||||||
//
|
|
||||||
// todo 实现数据校验码，防止篡改
|
|
||||||
|
|
||||||
// Encode 数据编码
|
|
||||||
func Encode(topic, data []byte) ([]byte, error) { |
|
||||||
topicLen := len(topic) |
|
||||||
dataLen := len(data) |
|
||||||
size := 2 + topicLen + dataLen // 不包括模数和数据长度
|
|
||||||
if topicLen == 0 { |
|
||||||
return nil, errors.New("topic too small") |
|
||||||
} |
|
||||||
if size+4 > math.MaxUint16 { |
|
||||||
return nil, errors.New("data too big") |
|
||||||
} |
|
||||||
buf := &bytes.Buffer{} |
|
||||||
// 由于 模数、topic 和 data 都是基本类型，所以当
|
|
||||||
// 执行 binary.Write() 时是不会出错的
|
|
||||||
_ = binary.Write(buf, binary.BigEndian, []byte{0x01, 0x02}) |
|
||||||
_ = binary.Write(buf, binary.BigEndian, uint16(size)) |
|
||||||
_ = binary.Write(buf, binary.BigEndian, uint16(topicLen)) |
|
||||||
_ = binary.Write(buf, binary.BigEndian, topic) |
|
||||||
_ = binary.Write(buf, binary.BigEndian, data) // 如果没有数据会不会报错
|
|
||||||
return buf.Bytes(), nil |
|
||||||
} |
|
||||||
|
|
||||||
// Decode 数据解码
|
|
||||||
func Decode(buf []byte) (topic, data []byte, err error) { |
|
||||||
r := bytes.NewReader(buf) |
|
||||||
// 读取模数
|
|
||||||
var v [2]byte |
|
||||||
err = binary.Read(r, binary.BigEndian, &v) |
|
||||||
if err != nil { |
|
||||||
return |
|
||||||
} |
|
||||||
if v[0] != 0x01 || v[1] != 0x02 { |
|
||||||
err = errors.New("protocol error") |
|
||||||
return |
|
||||||
} |
|
||||||
// 读取数据长度
|
|
||||||
var size uint16 |
|
||||||
err = binary.Read(r, binary.BigEndian, &size) |
|
||||||
if err != nil { |
|
||||||
return |
|
||||||
} |
|
||||||
// 读取主题长度
|
|
||||||
var topicLen uint16 |
|
||||||
err = binary.Read(r, binary.BigEndian, &topicLen) |
|
||||||
if err != nil { |
|
||||||
return |
|
||||||
} |
|
||||||
// 读取主题
|
|
||||||
topic = make([]byte, topicLen) |
|
||||||
err = binary.Read(r, binary.BigEndian, &topic) |
|
||||||
if err != nil { |
|
||||||
return |
|
||||||
} |
|
||||||
// 读取数据
|
|
||||||
data = make([]byte, size-2-topicLen) |
|
||||||
err = binary.Read(r, binary.BigEndian, &data) |
|
||||||
return |
|
||||||
} |
|
||||||
@ -1,215 +0,0 @@ |
|||||||
package app |
|
||||||
|
|
||||||
import ( |
|
||||||
"errors" |
|
||||||
"net" |
|
||||||
"slices" |
|
||||||
"sorbet/pkg/app/appkit" |
|
||||||
"sync" |
|
||||||
"sync/atomic" |
|
||||||
) |
|
||||||
|
|
||||||
var ( |
|
||||||
// 操作锁
|
|
||||||
//
|
|
||||||
// * 添加订阅器
|
|
||||||
lock sync.RWMutex |
|
||||||
|
|
||||||
// 用于控制主体程序并发，涉及以下几个方面：
|
|
||||||
//
|
|
||||||
// * 任务调度
|
|
||||||
// * 定时器
|
|
||||||
// * 发布订阅
|
|
||||||
// * 主程序轮询
|
|
||||||
wg sync.WaitGroup |
|
||||||
|
|
||||||
// 消息发布通道
|
|
||||||
pubs chan Msg |
|
||||||
|
|
||||||
// 消息订阅器
|
|
||||||
subs map[string][]*subscriber |
|
||||||
|
|
||||||
nextSubId int32 |
|
||||||
) |
|
||||||
|
|
||||||
func initBus() { |
|
||||||
if pubs == nil { |
|
||||||
pubs = make(chan Msg, 256) |
|
||||||
} |
|
||||||
if subs == nil { |
|
||||||
subs = make(map[string][]*subscriber) |
|
||||||
} |
|
||||||
OnFree(freeBus) |
|
||||||
} |
|
||||||
|
|
||||||
func freeBus() { |
|
||||||
clear(subs) |
|
||||||
|
|
||||||
select { |
|
||||||
case _, ok := <-pubs: |
|
||||||
if !ok { |
|
||||||
// 通道已经关闭了，需要重新构建
|
|
||||||
pubs = make(chan Msg, 256) |
|
||||||
return |
|
||||||
} |
|
||||||
// 清空 pubs 缓存的数据，复用它
|
|
||||||
l := len(pubs) |
|
||||||
for i := 0; i < l; i++ { |
|
||||||
<-pubs |
|
||||||
} |
|
||||||
default: |
|
||||||
// 通道里面没有值，可以复用
|
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
func Sub(topic string, handle func([]byte) []byte) appkit.Subscriber { |
|
||||||
lock.Lock() |
|
||||||
defer lock.Unlock() |
|
||||||
sub := &subscriber{ |
|
||||||
id: atomic.AddInt32(&nextSubId, 1), |
|
||||||
topic: topic, |
|
||||||
handle: handle, |
|
||||||
} |
|
||||||
if _, ok := subs[topic]; !ok { |
|
||||||
subs[topic] = make([]*subscriber, 0) |
|
||||||
} |
|
||||||
subs[topic] = append(subs[topic], sub) |
|
||||||
return sub |
|
||||||
} |
|
||||||
|
|
||||||
func Pub(topic string, data []byte) { |
|
||||||
pubMsg(Msg{Data: data, topic: topic}) |
|
||||||
} |
|
||||||
|
|
||||||
func pubMsg(msg Msg) { |
|
||||||
wg.Add(1) |
|
||||||
pubs <- msg |
|
||||||
} |
|
||||||
|
|
||||||
type Msg struct { |
|
||||||
Data []byte |
|
||||||
|
|
||||||
topic string |
|
||||||
addr net.Addr |
|
||||||
pc net.PacketConn |
|
||||||
} |
|
||||||
|
|
||||||
// Size 消息长度
|
|
||||||
func (m Msg) Size() int { |
|
||||||
return 6 + len([]byte(m.topic)) + len(m.Data) |
|
||||||
} |
|
||||||
|
|
||||||
type subscriber struct { |
|
||||||
id int32 |
|
||||||
topic string |
|
||||||
handle func([]byte) []byte |
|
||||||
} |
|
||||||
|
|
||||||
func (s *subscriber) Active() bool { |
|
||||||
return atomic.LoadInt32(&s.id) > 0 |
|
||||||
} |
|
||||||
|
|
||||||
func (s *subscriber) Cancel() { |
|
||||||
id := atomic.SwapInt32(&s.id, 0) |
|
||||||
if id == 0 { |
|
||||||
return |
|
||||||
} |
|
||||||
lock.Lock() |
|
||||||
defer lock.Unlock() |
|
||||||
subscribers, ok := subs[s.topic] |
|
||||||
if !ok { |
|
||||||
return |
|
||||||
} |
|
||||||
if len(subscribers) > 0 { |
|
||||||
subscribers = slices.DeleteFunc(subscribers, func(sub *subscriber) bool { |
|
||||||
return sub.id == id |
|
||||||
}) |
|
||||||
} |
|
||||||
if len(subscribers) > 0 { |
|
||||||
subs[s.topic] = subscribers |
|
||||||
} else { |
|
||||||
delete(subs, s.topic) |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
func (s *subscriber) invoke(data []byte) []byte { |
|
||||||
if s.Active() { |
|
||||||
return s.handle(data) |
|
||||||
} |
|
||||||
return nil |
|
||||||
} |
|
||||||
|
|
||||||
func dispatch(msg Msg) error { |
|
||||||
lock.RLock() |
|
||||||
var cbs []*subscriber |
|
||||||
if subscribers, ok := subs[msg.topic]; ok { |
|
||||||
for _, s := range subscribers { |
|
||||||
if s.Active() { |
|
||||||
cbs = append(cbs, s) |
|
||||||
} |
|
||||||
} |
|
||||||
} |
|
||||||
lock.RUnlock() |
|
||||||
|
|
||||||
if len(cbs) == 0 { |
|
||||||
return errors.New("no subscribers for topic " + msg.topic) |
|
||||||
} |
|
||||||
|
|
||||||
var response []byte |
|
||||||
for _, cb := range cbs { |
|
||||||
res := cb.invoke(msg.Data) |
|
||||||
if res != nil { |
|
||||||
response = res |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
if len(response) > 0 && msg.pc != nil { |
|
||||||
stats.v8workerRespond++ |
|
||||||
stats.v8workerBytesSent += len(response) |
|
||||||
|
|
||||||
n, ex := msg.pc.WriteTo(response, msg.addr) |
|
||||||
if n > 0 { |
|
||||||
stats.v8workerSend++ |
|
||||||
stats.v8workerBytesSent += n |
|
||||||
} |
|
||||||
if ex != nil { |
|
||||||
// todo handle the error
|
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
return nil |
|
||||||
} |
|
||||||
|
|
||||||
func recv(pc net.PacketConn, addr net.Addr, buf []byte) { |
|
||||||
stats.v8workerRecv++ |
|
||||||
stats.v8workerBytesRecv += len(buf) |
|
||||||
|
|
||||||
topic, data, err := appkit.Decode(buf) |
|
||||||
if err != nil { |
|
||||||
// todo handle the error
|
|
||||||
// return errors.New("invalid payload")
|
|
||||||
return |
|
||||||
} |
|
||||||
|
|
||||||
pubMsg(Msg{ |
|
||||||
Data: data, |
|
||||||
topic: string(topic), |
|
||||||
addr: addr, |
|
||||||
pc: pc, |
|
||||||
}) |
|
||||||
} |
|
||||||
|
|
||||||
func checkPubsEmpty() { |
|
||||||
// We've received a done event. As a sanity check, make sure that resChan is
|
|
||||||
// empty.
|
|
||||||
select { |
|
||||||
case _, ok := <-pubs: |
|
||||||
if ok { |
|
||||||
panic("Read a message from pubs after context deadlined.") |
|
||||||
} else { |
|
||||||
panic("pubs closed. Unexpected.") |
|
||||||
} |
|
||||||
default: |
|
||||||
// No value ready, moving on.
|
|
||||||
} |
|
||||||
} |
|
||||||
@ -1,83 +0,0 @@ |
|||||||
package app |
|
||||||
|
|
||||||
import ( |
|
||||||
"context" |
|
||||||
"sync/atomic" |
|
||||||
) |
|
||||||
|
|
||||||
const ( |
|
||||||
Idle int32 = iota |
|
||||||
Initialized |
|
||||||
Starting |
|
||||||
Running |
|
||||||
Stopping |
|
||||||
Stopped |
|
||||||
) |
|
||||||
|
|
||||||
var ( |
|
||||||
// 当前状态
|
|
||||||
status int32 |
|
||||||
|
|
||||||
// 应用上下文
|
|
||||||
ctx context.Context |
|
||||||
exit context.CancelFunc |
|
||||||
|
|
||||||
start func() // 应用启动钩子
|
|
||||||
stop func() // 应用停止钩子
|
|
||||||
free func() // 内存释放钩子
|
|
||||||
) |
|
||||||
|
|
||||||
func initLifecycle() { |
|
||||||
ctx, exit = context.WithCancel(context.Background()) |
|
||||||
|
|
||||||
start = func() { |
|
||||||
// nothing
|
|
||||||
} |
|
||||||
|
|
||||||
stop = func() { |
|
||||||
if exit != nil { |
|
||||||
exit() |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
free = func() { |
|
||||||
ctx = nil |
|
||||||
start = nil |
|
||||||
stop = nil |
|
||||||
exit = nil |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
// setStatus 设置应用状态
|
|
||||||
func setStatus(newStatus int32) { |
|
||||||
atomic.StoreInt32(&status, newStatus) |
|
||||||
} |
|
||||||
|
|
||||||
// Status 返回当前状态
|
|
||||||
func Status() int32 { |
|
||||||
return atomic.LoadInt32(&status) |
|
||||||
} |
|
||||||
|
|
||||||
func OnStart(fn func()) { |
|
||||||
oldStart := start |
|
||||||
start = func() { |
|
||||||
oldStart() |
|
||||||
fn() |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
func OnStop(fn func()) { |
|
||||||
oldStop := stop |
|
||||||
stop = func() { |
|
||||||
fn() |
|
||||||
oldStop() |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
func OnFree(fn func()) { |
|
||||||
oldFree := free |
|
||||||
free = func() { |
|
||||||
fn() |
|
||||||
oldFree() |
|
||||||
} |
|
||||||
} |
|
||||||
@ -1,44 +0,0 @@ |
|||||||
package app |
|
||||||
|
|
||||||
import ( |
|
||||||
"os" |
|
||||||
"reflect" |
|
||||||
"runtime/pprof" |
|
||||||
) |
|
||||||
|
|
||||||
var stats struct { |
|
||||||
v8workerSend int |
|
||||||
v8workerRespond int |
|
||||||
v8workerRecv int |
|
||||||
v8workerBytesSent int |
|
||||||
v8workerBytesRecv int |
|
||||||
} |
|
||||||
|
|
||||||
func initStats(args []string) { |
|
||||||
initCpuProfile(strArg(args, "--cpuprof")) |
|
||||||
initMemoryProfile(strArg(args, "--memprof")) |
|
||||||
OnFree(freeStats) |
|
||||||
} |
|
||||||
|
|
||||||
func initCpuProfile(cpuprof string) { |
|
||||||
if cpuprof != "" { |
|
||||||
cpuProfile, err := os.Create(cpuprof) |
|
||||||
check(err) |
|
||||||
check(pprof.StartCPUProfile(cpuProfile)) |
|
||||||
OnStop(pprof.StopCPUProfile) |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
func initMemoryProfile(memprof string) { |
|
||||||
if memprof != "" { |
|
||||||
memProfile, err := os.Create(memprof) |
|
||||||
check(err) |
|
||||||
check(pprof.WriteHeapProfile(memProfile)) |
|
||||||
OnStop(func() { check(memProfile.Close()) }) |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
func freeStats() { |
|
||||||
newStats := reflect.New(reflect.TypeOf(stats)) |
|
||||||
reflect.ValueOf(&stats).Elem().Set(newStats.Elem()) |
|
||||||
} |
|
||||||
@ -1,63 +0,0 @@ |
|||||||
package app |
|
||||||
|
|
||||||
import ( |
|
||||||
"context" |
|
||||||
"errors" |
|
||||||
"io" |
|
||||||
"net" |
|
||||||
) |
|
||||||
|
|
||||||
// 通过 UDP 协议实现外部通信
|
|
||||||
var pc net.PacketConn |
|
||||||
|
|
||||||
func initUdp(args []string) { |
|
||||||
// 未开启 udp
|
|
||||||
if !boolArg(args, "--udp") { |
|
||||||
return |
|
||||||
} |
|
||||||
|
|
||||||
// 这一步属于初始化，切不可放到 OnStart 里面
|
|
||||||
sizeLimit := intArg(args, "--payload", 1024<<2) |
|
||||||
addr := strArg(args, "--addr", ":1324") |
|
||||||
|
|
||||||
// 当程序启动时开启 udp 服务器
|
|
||||||
OnStart(func() { |
|
||||||
listenUDP(addr, sizeLimit) |
|
||||||
}) |
|
||||||
|
|
||||||
OnStop(freeUdp) |
|
||||||
} |
|
||||||
|
|
||||||
func freeUdp() { |
|
||||||
if pc != nil { |
|
||||||
err := pc.Close() |
|
||||||
pc = nil |
|
||||||
check(err) |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
func listenUDP(address string, maxSize int) { |
|
||||||
var err error |
|
||||||
pc, err = net.ListenPacket("udp", address) |
|
||||||
check(err) |
|
||||||
|
|
||||||
async(func(ctx context.Context) { |
|
||||||
for { |
|
||||||
select { |
|
||||||
case <-ctx.Done(): |
|
||||||
return |
|
||||||
default: |
|
||||||
} |
|
||||||
|
|
||||||
buf := make([]byte, maxSize) |
|
||||||
n, addr, err := pc.ReadFrom(buf) |
|
||||||
if err != nil { |
|
||||||
if errors.Is(err, io.EOF) { |
|
||||||
return |
|
||||||
} |
|
||||||
continue |
|
||||||
} |
|
||||||
recv(pc, addr, buf[:n]) |
|
||||||
} |
|
||||||
}) |
|
||||||
} |
|
||||||
@ -1,101 +0,0 @@ |
|||||||
package app |
|
||||||
|
|
||||||
import ( |
|
||||||
"context" |
|
||||||
"errors" |
|
||||||
"os" |
|
||||||
"slices" |
|
||||||
"strconv" |
|
||||||
"strings" |
|
||||||
) |
|
||||||
|
|
||||||
func assert(cond bool, msg string) { |
|
||||||
if !cond { |
|
||||||
panic(msg) |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
func check(e error) { |
|
||||||
if e != nil { |
|
||||||
panic(e) |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
func exitOnError(err error) { |
|
||||||
if err != nil { |
|
||||||
os.Stderr.WriteString(err.Error()) |
|
||||||
os.Exit(1) |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
func async(cb func(ctx context.Context)) { |
|
||||||
wg.Add(1) |
|
||||||
go func() { |
|
||||||
defer wg.Done() |
|
||||||
select { |
|
||||||
case <-ctx.Done(): |
|
||||||
return |
|
||||||
default: |
|
||||||
cb(ctx) |
|
||||||
} |
|
||||||
}() |
|
||||||
} |
|
||||||
|
|
||||||
func getArg[T any](args []string, name string, f func(string, bool) T) T { |
|
||||||
i := slices.Index(args, name) |
|
||||||
if i == -1 { |
|
||||||
return f("", false) |
|
||||||
} |
|
||||||
if i == len(args)-1 { |
|
||||||
panic(errors.New("invalid " + name + " option")) |
|
||||||
} |
|
||||||
s := args[i+1] |
|
||||||
if strings.HasPrefix(s, "--") { |
|
||||||
return f("", true) |
|
||||||
} |
|
||||||
return f(s, true) |
|
||||||
} |
|
||||||
|
|
||||||
func strArg(args []string, name string, def ...string) string { |
|
||||||
return getArg(args, name, func(s string, b bool) string { |
|
||||||
if b { |
|
||||||
return s |
|
||||||
} |
|
||||||
for _, v := range def { |
|
||||||
return v |
|
||||||
} |
|
||||||
return "" |
|
||||||
}) |
|
||||||
} |
|
||||||
|
|
||||||
func intArg(args []string, name string, def ...int) int { |
|
||||||
return getArg(args, name, func(s string, b bool) int { |
|
||||||
if b { |
|
||||||
size, err := strconv.Atoi(s) |
|
||||||
check(err) |
|
||||||
assert(size > 0, "invalid "+name+" option") |
|
||||||
return size |
|
||||||
} |
|
||||||
for _, v := range def { |
|
||||||
return v |
|
||||||
} |
|
||||||
return 0 |
|
||||||
}) |
|
||||||
} |
|
||||||
|
|
||||||
func boolArg(args []string, name string, def ...bool) bool { |
|
||||||
return getArg(args, name, func(s string, b bool) bool { |
|
||||||
if b { |
|
||||||
if s == "" { |
|
||||||
return true |
|
||||||
} |
|
||||||
v, err := strconv.ParseBool(s) |
|
||||||
check(err) |
|
||||||
return v |
|
||||||
} |
|
||||||
for _, v := range def { |
|
||||||
return v |
|
||||||
} |
|
||||||
return false |
|
||||||
}) |
|
||||||
} |
|
||||||
@ -1,53 +0,0 @@ |
|||||||
package crud |
|
||||||
|
|
||||||
import ( |
|
||||||
"context" |
|
||||||
"github.com/labstack/echo/v4" |
|
||||||
"sync" |
|
||||||
) |
|
||||||
|
|
||||||
type Context struct { |
|
||||||
context.Context |
|
||||||
store map[any]any |
|
||||||
router *echo.Group |
|
||||||
mu sync.RWMutex |
|
||||||
} |
|
||||||
|
|
||||||
func NewContext(ctx context.Context, router *echo.Group) *Context { |
|
||||||
return &Context{ |
|
||||||
Context: ctx, |
|
||||||
store: make(map[any]any), |
|
||||||
router: router, |
|
||||||
mu: sync.RWMutex{}, |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
// Routes 注册路由
|
|
||||||
func (c *Context) Routes(routes Routable) { |
|
||||||
c.mu.Lock() |
|
||||||
defer c.mu.Unlock() |
|
||||||
routes.InitRoutes(c.router) |
|
||||||
} |
|
||||||
|
|
||||||
// Set 设置值
|
|
||||||
func (c *Context) Set(key, val any) { |
|
||||||
c.mu.Lock() |
|
||||||
defer c.mu.Unlock() |
|
||||||
c.store[key] = val |
|
||||||
} |
|
||||||
|
|
||||||
// Get 获取值，只会获取通过 Set 方法设置的值
|
|
||||||
func (c *Context) Get(key any) (any, bool) { |
|
||||||
c.mu.RLock() |
|
||||||
defer c.mu.RUnlock() |
|
||||||
val, ok := c.store[key] |
|
||||||
return val, ok |
|
||||||
} |
|
||||||
|
|
||||||
// Value 获取值
|
|
||||||
func (c *Context) Value(key any) any { |
|
||||||
if val, ok := c.Get(key); ok { |
|
||||||
return val |
|
||||||
} |
|
||||||
return c.Value(key) |
|
||||||
} |
|
||||||
@ -1,20 +0,0 @@ |
|||||||
package crud |
|
||||||
|
|
||||||
import "github.com/labstack/echo/v4" |
|
||||||
|
|
||||||
type Service interface { |
|
||||||
// Name 服务名称
|
|
||||||
Name() string |
|
||||||
// Priority 优先级，用于启动和销毁的执行顺序
|
|
||||||
Priority() int |
|
||||||
// Init 初始化服务
|
|
||||||
Init(ctx *Context) error |
|
||||||
// Bootstrap 启动服务
|
|
||||||
Bootstrap() error |
|
||||||
// Destroy 销毁服务
|
|
||||||
Destroy() error |
|
||||||
} |
|
||||||
|
|
||||||
type Routable interface { |
|
||||||
InitRoutes(r *echo.Group) |
|
||||||
} |
|
||||||
Loading…
Reference in new issue