Skip to content

G
goku-api-gateway

Eolink / goku-api-gateway
上一次同步 1 年多

通知 133
Star 2992
Fork 611
G
goku-api-gateway
提交 dbd18ad4 编写于 作者: Y Your Name
浏览文件
操作

Revert "3.1.3版本提交" 

This reverts commit 3364e59b.
上级 7e441718
隐藏空白更改
内联 并排
Showing with 66 addition and 698 deletion
common/database/conn.go
浏览文件 @ dbd18ad4
......@@ -2,6 +2,8 @@ package database
import (
"database/sql"
"io/ioutil"
"strings"
log "github.com/eolinker/goku-api-gateway/goku-log"
......@@ -11,9 +13,15 @@ import (
_ "github.com/mattn/go-sqlite3"
)
var (
defaultDB *sql.DB
)
//InitConnection 初始化数据库连接
func InitConnection(config Config) (*sql.DB, error) {
return getConnection(config)
func InitConnection(config Config) error {
db, e := getConnection(config)
defaultDB = db
return e
}
func getConnection(config Config) (*sql.DB, error) {
......@@ -26,6 +34,7 @@ func getConnection(config Config) (*sql.DB, error) {
}
db.SetMaxOpenConns(1000)
db.SetMaxIdleConns(100)
defaultDB = db
return db, nil
}
log.Info(e)
......@@ -33,17 +42,25 @@ func getConnection(config Config) (*sql.DB, error) {
}
//CheckConnection 检查数据库连接
func CheckConnection(driver string, source string) error {
//GetConnection 获取数据库连接
func GetConnection() *sql.DB {
return defaultDB
}
db, e := sql.Open(driver, source)
defer db.Close()
if e == nil {
if err := db.Ping(); err != nil {
//InitTable 初始化表
func InitTable() error {
content, err := ioutil.ReadFile("sql/goku_ce.sql")
sqls := strings.Split(string(content), ";")
Tx, _ := GetConnection().Begin()
for _, sql := range sqls {
_, err = Tx.Exec(sql)
if err != nil {
Tx.Rollback()
log.Error("InitTable error:",err,"\t sql:",sql)
return err
}
return nil
}
return e
Tx.Commit()
return nil
}
common/database/database.go
浏览文件 @ dbd18ad4
package database
const (
//MysqlDriver mysql驱动器
MysqlDriver = "mysql"
//Sqlite3Driver Sqlite3Driver驱动
Sqlite3Driver = "sqlite3"
)
//Config 数据库配置结构体
type Config interface {
GetDriver() string
......
common/endless/endless.go
浏览文件 @ dbd18ad4
......@@ -4,6 +4,7 @@ import (
"crypto/tls"
"errors"
"fmt"
log "github.com/eolinker/goku-api-gateway/goku-log"
"net"
"net/http"
"os"
......@@ -14,8 +15,6 @@ import (
"sync"
"syscall"
"time"
log "github.com/eolinker/goku-api-gateway/goku-log"
)
const (
......@@ -40,10 +39,9 @@ var (
DefaultMaxHeaderBytes int
DefaultHammerTime time.Duration
isChild bool
socketOrder string
adminServer *endlessServer
nodeServer *endlessServer
isChild bool
socketOrder string
hookableSignals []os.Signal
)
......@@ -73,24 +71,15 @@ func init() {
type endlessServer struct {
http.Server
EndlessListener net.Listener
SignalHooks map[int]map[os.Signal][]func()
tlsInnerListener *endlessListener
wg sync.WaitGroup
sigChan chan os.Signal
isChild bool
state uint8
lock *sync.RWMutex
BeforeBegin func(add string)
runningServersForked bool
}
func SetAdminServer(server *endlessServer) {
adminServer = server
}
func SetNodeServer(server *endlessServer) {
nodeServer = server
EndlessListener net.Listener
SignalHooks map[int]map[os.Signal][]func()
tlsInnerListener *endlessListener
wg sync.WaitGroup
sigChan chan os.Signal
isChild bool
state uint8
lock *sync.RWMutex
BeforeBegin func(add string)
}
/*
......@@ -151,25 +140,17 @@ func NewServer(addr string, handler http.Handler) (srv *endlessServer) {
runningServersOrder = append(runningServersOrder, addr)
runningServers[addr] = srv
return
}
func RestartServer() {
nodeServer.sigChan <- syscall.SIGHUP
}
func StopServer() {
nodeServer.sigChan <- syscall.SIGTERM
return
}
/*F
/*
ListenAndServe listens on the TCP network address addr and then calls Serve
with handler to handle requests on incoming connections. Handler is typically
nil, in which case the DefaultServeMux is used.
*/
func ListenAndServe(addr string, handler http.Handler) error {
server := NewServer(addr, handler)
nodeServer = server
return server.ListenAndServe()
}
......@@ -357,7 +338,7 @@ func (srv *endlessServer) handleSignals() {
log.Info(pid, "Received SIGINT.")
srv.shutdown()
case syscall.SIGTERM:
log.Info(``, "Received SIGTERM.")
log.Info(pid, "Received SIGTERM.")
srv.shutdown()
case syscall.SIGTSTP:
log.Info(pid, "Received SIGTSTP.")
......@@ -437,15 +418,15 @@ func (srv *endlessServer) hammerTime(d time.Duration) {
func (srv *endlessServer) fork() (err error) {
runningServerReg.Lock()
defer runningServerReg.Unlock()
log.Info(1234)
// only one server instance should fork!
if srv.runningServersForked {
if runningServersForked {
return errors.New("Another process already forked. Ignoring this one.")
}
log.Info(5456)
srv.runningServersForked = true
runningServersForked = true
var files = make([]*os.File, len(runningServers))
var orderArgs = make([]string, len(runningServers))
......@@ -462,7 +443,7 @@ func (srv *endlessServer) fork() (err error) {
}
orderArgs[socketPtrOffsetMap[srvPtr.Server.Addr]] = srvPtr.Server.Addr
}
log.Info(123)
env := append(
os.Environ(),
"ENDLESS_CONTINUE=1",
......@@ -477,7 +458,7 @@ func (srv *endlessServer) fork() (err error) {
if len(os.Args) > 1 {
args = os.Args[1:]
}
log.Info(path, args)
cmd := exec.Command(path, args...)
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
......@@ -495,7 +476,6 @@ func (srv *endlessServer) fork() (err error) {
log.Fatalf("Restart: Failed to launch, error: %v", err)
}
srv.runningServersForked = false
return
}
......
common/ioutils/utils.go 已删除 100644 → 0
浏览文件 @ 7e441718
package ioutils
import (
"encoding/binary"
"io"
)
//WriteLField 写域内容
func WriteLField(writer io.Writer, s []byte) (int, error) {
l := uint8(len(s))
e := binary.Write(writer, binary.BigEndian, l)
if e != nil {
return 0, e
}
n, err := writer.Write(s)
if err != nil {
return n + 1, err
}
return n + 1, nil
}
//ReadLField 读取域内容
func ReadLField(reader io.Reader, buf []byte) ([]byte, int, error) {
l := uint8(0)
e := binary.Read(reader, binary.BigEndian, &l)
if e != nil {
return nil, 0, e
}
tmpbuf := buf
if int(l) > len(buf) {
tmpbuf = make([]byte, l, l)
} else {
tmpbuf = buf[:l]
}
_, err := reader.Read(tmpbuf)
if err != nil {
return nil, 0, err
}
return tmpbuf, int(l) + 1, nil
}
common/listener/intercept.go 已删除 100644 → 0
浏览文件 @ 7e441718
package listener
import "sync"
//InterceptFunc 拦截函数
type InterceptFunc func(event interface{}) error
//Intercept 拦截器
type Intercept struct {
callbacks []InterceptFunc
locker sync.RWMutex
}
//NewIntercept 创建拦截器
func NewIntercept() *Intercept {
return &Intercept{
callbacks: nil,
locker: sync.RWMutex{},
}
}
//Add add
func (i *Intercept) Add(f func(v interface{}) error) {
i.locker.Lock()
i.callbacks = append(i.callbacks, InterceptFunc(f))
i.locker.Unlock()
}
//Call call
func (i *Intercept) Call(v interface{}) error {
i.locker.RLock()
fs := i.callbacks
i.locker.RUnlock()
for _, f := range fs {
err := f(v)
if err != nil {
return err
}
}
return nil
}
common/listener/listener.go
浏览文件 @ dbd18ad4
......@@ -5,7 +5,7 @@ import "sync"
//CallbackFunc 回调函数
type CallbackFunc func(event interface{})
//Listener 监听者
//Listener
type Listener struct {
callbacks []CallbackFunc
callbacksOnce []CallbackFunc
......
common/manager/manager.go 已删除 100644 → 0
浏览文件 @ 7e441718
package manager
import "sync"
//Manager manager
type Manager struct {
locker sync.RWMutex
objs map[string]interface{}
}
//NewManager new manager
func NewManager() *Manager {
return &Manager{
locker: sync.RWMutex{},
objs: make(map[string]interface{}),
}
}
//Get get
func (m *Manager) Get(key string) (interface{}, bool) {
m.locker.RLock()
v, has := m.objs[key]
m.locker.RUnlock()
return v, has
}
//Set set
func (m *Manager) Set(key string, value interface{}) {
m.locker.Lock()
m.objs[key] = value
m.locker.Unlock()
}
//Value value
type Value struct {
locker sync.RWMutex
isHas bool
value interface{}
}
//NewValue new value
func NewValue() *Value {
return &Value{
locker: sync.RWMutex{},
isHas: false,
value: nil,
}
}
//Set set
func (v *Value) Set(value interface{}) {
v.locker.Lock()
v.value = value
v.isHas = true
v.locker.Unlock()
}
//Get get
func (v *Value) Get() (interface{}, bool) {
v.locker.RLock()
value, has := v.value, v.isHas
v.locker.RUnlock()
return value, has
}
common/pdao/demo/demo.go 已删除 100644 → 0
浏览文件 @ 7e441718
package main
import (
"fmt"
"github.com/eolinker/goku-api-gateway/common/pdao"
)
type DemoGetDao interface {
GetNameById(id int) string
GetIdByName(name string) int
}
type DemoSetDao interface {
Set(id int,name string)
}
type Demo struct {
names map[string]int
ids map[int]string
}
func NewDemo() *Demo {
return &Demo{
names: make(map[string]int),
ids: make(map[int]string),
}}
func (d *Demo) Set(id int, name string) {
d.names[name]=id
d.ids[id]=name
}
func (d *Demo) GetNameById(id int) string {
return d.ids[id]
}
func (d *Demo) GetIdByName(name string) int {
return d.names[name]
}
var (
getDao DemoGetDao
setDao DemoSetDao
)
func init() {
pdao.Need(&getDao)
pdao.Need(&setDao)
}
func main() {
demo:=NewDemo()
var seter DemoSetDao = demo
pdao.Set(&seter)
var getter DemoGetDao = demo
pdao.Set(&getter)
pdao.Check()
setDao.Set(1,"test")
fmt.Println(getDao.GetNameById(1))
fmt.Println(getDao.GetIdByName("test"))
}
\ No newline at end of file
common/pdao/factory.go 已删除 100644 → 0
浏览文件 @ 7e441718
package pdao
import "database/sql"
//FactoryFunc factoryFunc
type FactoryFunc func(db *sql.DB) (interface{}, error)
//Create create
func (f FactoryFunc) Create(db *sql.DB) (interface{}, error) {
return f(db)
}
//Factory factory
type Factory interface {
Create(db *sql.DB) (interface{}, error)
}
//DBBuilder dbBuilder
type DBBuilder interface {
Build(db *sql.DB) error
}
common/pdao/needs.go 已删除 100644 → 0
浏览文件 @ 7e441718
package pdao
import (
"fmt"
"reflect"
"sync"
)
//NeedsManager needsManager
type NeedsManager struct {
daoInterfaces map[string][]*reflect.Value
lock sync.Mutex
}
//NewNeedsManager 创建新的needsManager
func NewNeedsManager() *NeedsManager {
return &NeedsManager{
daoInterfaces: make(map[string][]*reflect.Value),
lock: sync.Mutex{},
}
}
func (m *NeedsManager) add(key string, v *reflect.Value) {
m.lock.Lock()
m.daoInterfaces[key] = append(m.daoInterfaces[key], v)
m.lock.Unlock()
}
func (m *NeedsManager) set(key string, v reflect.Value) {
m.lock.Lock()
for _, e := range m.daoInterfaces[key] {
e.Set(v)
}
delete(m.daoInterfaces, key)
m.lock.Unlock()
}
func (m *NeedsManager) check() []string {
m.lock.Lock()
r := make([]string, 0, len(m.daoInterfaces))
for pkg := range m.daoInterfaces {
r = append(r, pkg)
}
m.lock.Unlock()
return r
}
//Need 声明
func (m *NeedsManager) Need(p interface{}) {
v := reflect.ValueOf(p)
if v.Kind() != reflect.Ptr {
panic("must ptr")
}
e := v.Elem()
pkg := key(e.Type())
if pkg == "" {
panic("invalid interface")
}
if !e.CanSet() {
panic("invalid interface")
}
m.add(pkg, &e)
}
//Set 注入
func (m *NeedsManager) Set(i interface{}) {
v := reflect.ValueOf(i)
if v.Kind() != reflect.Ptr {
panic("must ptr")
}
e := v.Elem()
pkg := key(e.Type())
if pkg == "" {
panic("invalid interface")
}
m.set(pkg, e)
}
func key(t reflect.Type) string {
return fmt.Sprintf("%s.%s", t.PkgPath(), t.String())
}
//Check 检查是否实现相关dao类
func (m *NeedsManager) Check() error {
rs := m.check()
if len(rs) > 0 {
return fmt.Errorf("need:%v", rs)
}
return nil
}
common/pdao/needs_test.go 已删除 100644 → 0
浏览文件 @ 7e441718
package pdao
import (
"fmt"
"testing"
test1 "github.com/eolinker/goku-api-gateway/common/pdao/test/test/test/test"
)
type T1 struct {
}
func (t *T1) Test4() {
fmt.Println("implement me")
}
type T2 struct {
}
func (t *T2) Test3() {
fmt.Println("implement me")
}
type T3 struct {
}
func (t *T3) Test2() {
fmt.Println("implement me")
}
type T4 struct {
}
func (t *T4) Test1() {
fmt.Println("implement me")
}
func Test(t *testing.T) {
t.Run("all", func(t *testing.T) {
var t1 test1.Test
//var t2 test2.Test = nil
//var t3 test3.Test = nil
//var t4 test4.Test = nil
Need(&t1)
//Need(&t2)
//Need(&t3)
//Need(&t4)
var t1v test1.Test = new(T4)
Set(&t1v)
//Set(test2.Test(new(T3)))
//
//Set(test3.Test(new(T2)))
//Set(test4.Test(new(T1)))
t1.Test1()
//t2.Test2()
//t3.Test3()
//t4.Test4()
Check()
})
}
common/pdao/pdao.go 已删除 100644 → 0
浏览文件 @ 7e441718
package pdao
import (
"database/sql"
"sync"
)
var (
needsManager = NewNeedsManager()
factoryManage = NewFactoryManager()
)
//Need need
func Need(is ...interface{}) {
for _, i := range is {
needsManager.Need(i)
}
}
//Check check
func Check() error {
return needsManager.Check()
}
//RegisterDao 注册dao类
func RegisterDao(driver string, factories ...Factory) {
for _, factory := range factories {
factoryManage.RegisterDao(driver, factory)
}
}
//RegisterDBBuilder 注册dbBuilder
func RegisterDBBuilder(driver string, builders ...DBBuilder) {
for _, builder := range builders {
factoryManage.RegisterDBBuilder(driver, builder)
}
}
//Build build
func Build(driver string, db *sql.DB) error {
return factoryManage.Build(driver, db, needsManager)
}
//FactoryManager 工厂管理者
type FactoryManager struct {
factories map[string][]Factory
builders map[string][]DBBuilder
locker sync.Mutex
}
//NewFactoryManager new工厂管理者
func NewFactoryManager() *FactoryManager {
return &FactoryManager{
factories: make(map[string][]Factory),
builders: make(map[string][]DBBuilder),
locker: sync.Mutex{},
}
}
//RegisterDBBuilder dbBuilder注册器
func (f *FactoryManager) RegisterDBBuilder(driver string, builder DBBuilder) {
f.locker.Lock()
f.builders[driver] = append(f.builders[driver], builder)
f.locker.Unlock()
}
//RegisterDao dao类注册器
func (f *FactoryManager) RegisterDao(driver string, factory Factory) {
f.locker.Lock()
f.factories[driver] = append(f.factories[driver], factory)
f.locker.Unlock()
}
func (f *FactoryManager) get(driver string) []Factory {
f.locker.Lock()
fs := f.factories[driver]
delete(f.factories, driver)
f.locker.Unlock()
return fs
}
func (f *FactoryManager) callBuild(driver string, db *sql.DB) error {
f.locker.Lock()
bs := f.builders[driver]
for _, b := range bs {
err := b.Build(db)
if err != nil {
f.locker.Unlock()
return err
}
}
f.locker.Unlock()
return nil
}
//Build build
func (f *FactoryManager) Build(driver string, db *sql.DB, m *NeedsManager) error {
err := f.callBuild(driver, db)
if err != nil {
return err
}
fs := f.get(driver)
for _, factory := range fs {
i, err := factory.Create(db)
if err != nil {
return err
}
m.Set(i)
}
return nil
}
common/pdao/test/t.go 已删除 100644 → 0
浏览文件 @ 7e441718
package test
type Test interface {
Test4()
}
\ No newline at end of file
common/pdao/test/test/t.go 已删除 100644 → 0
浏览文件 @ 7e441718
package test
type Test interface {
Test3()
}
\ No newline at end of file
common/pdao/test/test/test/t.go 已删除 100644 → 0
浏览文件 @ 7e441718
package test
type Test interface {
Test2()
}
\ No newline at end of file
common/pdao/test/test/test/test/t.go 已删除 100644 → 0
浏览文件 @ 7e441718
package test
type Test interface {
Test1()
}
\ No newline at end of file
common/redis-manager/manager.go
浏览文件 @ dbd18ad4
......@@ -54,18 +54,3 @@ func Get(name string) (Redis, bool) {
return nil, false
}
func del(name string) {
locker.RLock()
defer locker.RUnlock()
delete(redisOfCluster, name)
return
}
//Delete delete
func Delete(name string) {
del(name)
return
}
common/redis-manager/redis.go
浏览文件 @ dbd18ad4
......@@ -5,9 +5,6 @@ import "github.com/go-redis/redis"
const (
//RedisModeCluster cluster模式
RedisModeCluster = "cluster"
//RedisModeSentinel sentinel模式
RedisModeSentinel = "sentinel"
//RedisModeStand stand模式
RedisModeStand = "stand"
)
......
common/redis-manager/redis_conn.go
浏览文件 @ dbd18ad4
package redis_manager
import (
"fmt"
"sort"
"strings"
"sync"
"github.com/go-redis/redis"
)
var (
......@@ -22,77 +17,17 @@ func SetDefault(r Redis) {
//GetConnection 获取redis连接
func GetConnection() Redis {
//if def != nil {
return def
//}
//defLocker.Lock()
//defer defLocker.Unlock()
//
//def = Create(defaultConfig)
//return def
}
//CheckConnection 获取redis连接
func CheckConnection(mode, addrs, password, masters string, dbIndex int) bool {
switch mode {
case RedisModeCluster:
{
a := strings.Split(addrs, ",")
option := &redis.ClusterOptions{
Addrs: a,
Password: password,
PoolSize: 2000,
ReadOnly: true,
RouteByLatency: true,
}
r := redis.NewClusterClient(option)
defer r.Close()
if _, err := r.Ping().Result(); err != nil {
return false
}
return true
}
case RedisModeSentinel:
{
a := strings.Split(addrs, ",")
option := redis.FailoverOptions{
SentinelAddrs: a,
MasterName: masters,
Password: password,
DB: dbIndex,
PoolSize: _PoolSize,
}
r := redis.NewFailoverClient(&option)
defer r.Close()
if _, err := r.Ping().Result(); err != nil {
return false
}
return true
}
case RedisModeStand:
{
a := strings.Split(addrs, ",")
if def != nil {
return def
}
option := redis.RingOptions{
Addrs: make(map[string]string),
Password: password,
DB: dbIndex,
defLocker.Lock()
defer defLocker.Unlock()
PoolSize: _PoolSize,
}
sort.Strings(a)
for i, ad := range a {
option.Addrs[fmt.Sprintf("shad:%d", i)] = ad
}
r := redis.NewRing(&option)
defer r.Close()
if _, err := r.Ping().Result(); err != nil {
return false
}
return true
}
if def != nil {
return def
}
return false
def = Create(defaultConfig)
return def
}
common/telegraph/telegraph.go
浏览文件 @ dbd18ad4
......@@ -5,12 +5,9 @@ import (
"errors"
"sync"
)
var (
//ErrorContextDone context done error
ErrorContextDone = errors.New("context done")
)
//Telegraph telegraph
type Telegraph struct {
value interface{}
......@@ -53,7 +50,7 @@ func (t *Telegraph) get() (string, <-chan struct{}, interface{}) {
}
//Get get
func (t *Telegraph) Get(version string) (interface{}, error) {
func (t *Telegraph) Get(version string) (interface{},error) {
return t.GetWidthContext(context.Background(), version)
}
......@@ -66,21 +63,21 @@ func (t *Telegraph) Close() {
}
//GetWidthContext 获取上下文
func (t *Telegraph) GetWidthContext(ctx context.Context, version string) (interface{}, error) {
func (t *Telegraph) GetWidthContext(ctx context.Context, version string) (interface{} ,error){
v, c, value := t.get()
if v == "" {
// closed
return nil, nil
return nil,nil
}
if version != v {
return value, nil
return value,nil
}
select {
case <-c:
return t.GetWidthContext(ctx, version)
case <-ctx.Done():
return nil, ErrorContextDone
return nil,ErrorContextDone
}
}
common/version/version.go
浏览文件 @ dbd18ad4
package version
//Version 版本号
const Version = "3.1.3"
const Version = "3.1.1"
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或