Compare commits
12 Commits
new_conn
...
86fac6bbc0
| Author | SHA1 | Date | |
|---|---|---|---|
| 86fac6bbc0 | |||
| 70d3b2507c | |||
| ca5632031c | |||
| 38c5e05d4c | |||
| 7928e69c60 | |||
| 899bae335e | |||
| 8e3d6c28f0 | |||
| ae98abe61d | |||
| 013c89e58d | |||
| dd4928c822 | |||
| 2dafadf949 | |||
| f0a97c4701 |
@ -8,6 +8,7 @@ import (
|
|||||||
"path"
|
"path"
|
||||||
"runtime"
|
"runtime"
|
||||||
"runtime/debug"
|
"runtime/debug"
|
||||||
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"repositories.action2quare.com/ayo/gocommon/flagx"
|
"repositories.action2quare.com/ayo/gocommon/flagx"
|
||||||
@ -15,6 +16,7 @@ import (
|
|||||||
|
|
||||||
var stdlogger *log.Logger
|
var stdlogger *log.Logger
|
||||||
var UseLogFile = flagx.Bool("logfile", false, "")
|
var UseLogFile = flagx.Bool("logfile", false, "")
|
||||||
|
var _ = flagx.Int("logprefix", 3, "0 : no_prefix, 1 : date, 2 : time, 3 : datetime")
|
||||||
|
|
||||||
func init() {
|
func init() {
|
||||||
binpath, _ := os.Executable()
|
binpath, _ := os.Executable()
|
||||||
@ -23,7 +25,29 @@ func init() {
|
|||||||
var outWriter io.Writer
|
var outWriter io.Writer
|
||||||
outWriter = os.Stdout
|
outWriter = os.Stdout
|
||||||
|
|
||||||
if *UseLogFile {
|
args := os.Args
|
||||||
|
useLogFile := false
|
||||||
|
for _, arg := range args {
|
||||||
|
if strings.HasPrefix(arg, "-logfile=") {
|
||||||
|
useLogFile, _ = strconv.ParseBool(arg[9:])
|
||||||
|
break
|
||||||
|
}
|
||||||
|
|
||||||
|
if arg == "-logfile" {
|
||||||
|
useLogFile = true
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
logprefix := 3
|
||||||
|
for _, arg := range args {
|
||||||
|
if strings.HasPrefix(arg, "-logprefix=") {
|
||||||
|
logprefix, _ = strconv.Atoi(arg[11:])
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if useLogFile {
|
||||||
ext := path.Ext(binname)
|
ext := path.Ext(binname)
|
||||||
if len(ext) > 0 {
|
if len(ext) > 0 {
|
||||||
binname = binname[:len(binname)-len(ext)]
|
binname = binname[:len(binname)-len(ext)]
|
||||||
@ -38,7 +62,11 @@ func init() {
|
|||||||
outWriter = io.MultiWriter(outWriter, logFile)
|
outWriter = io.MultiWriter(outWriter, logFile)
|
||||||
}
|
}
|
||||||
|
|
||||||
stdlogger = log.New(outWriter, "", log.LstdFlags)
|
if logprefix < 4 {
|
||||||
|
stdlogger = log.New(outWriter, "", logprefix)
|
||||||
|
} else {
|
||||||
|
stdlogger = log.New(outWriter, "", log.LstdFlags)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func Println(v ...interface{}) {
|
func Println(v ...interface{}) {
|
||||||
|
|||||||
@ -135,7 +135,11 @@ func (pe *prometheusExporter) loop(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if err := prometheus.Register(nextcollector); err != nil {
|
if err := prometheus.Register(nextcollector); err != nil {
|
||||||
logger.Error("prometheus register err :", *nm, err)
|
if _, ok := err.(prometheus.AlreadyRegisteredError); ok {
|
||||||
|
// 이미 등록된 metric. child process를 여럿 실행하면 발생됨
|
||||||
|
} else {
|
||||||
|
logger.Error("prometheus register err :", *nm, err)
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
collector = nextcollector
|
collector = nextcollector
|
||||||
}
|
}
|
||||||
|
|||||||
28
mongo.go
28
mongo.go
@ -154,10 +154,36 @@ func (mc *MongoClient) DropIndex(coll CollectionName, name string) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (mc *MongoClient) Watch(coll CollectionName, pipeline mongo.Pipeline, opts ...*options.ChangeStreamOptions) (*mongo.ChangeStream, error) {
|
func (mc *MongoClient) Watch(coll CollectionName, pipeline mongo.Pipeline, opts ...*options.ChangeStreamOptions) (*mongo.ChangeStream, error) {
|
||||||
|
// mc.db.RunCommand()
|
||||||
if len(opts) == 0 {
|
if len(opts) == 0 {
|
||||||
opts = []*options.ChangeStreamOptions{options.ChangeStream().SetFullDocument(options.UpdateLookup).SetMaxAwaitTime(0)}
|
opts = []*options.ChangeStreamOptions{options.ChangeStream().SetFullDocument(options.UpdateLookup).SetMaxAwaitTime(0)}
|
||||||
}
|
}
|
||||||
return mc.Collection(coll).Watch(mc.ctx, pipeline, opts...)
|
|
||||||
|
stream, err := mc.Collection(coll).Watch(mc.ctx, pipeline, opts...)
|
||||||
|
if err != nil {
|
||||||
|
if mongoErr, ok := err.(mongo.CommandError); ok {
|
||||||
|
logger.Println("MongoClient Watch return err code :", mongoErr, mongoErr.Code)
|
||||||
|
if mongoErr.Code == 40573 {
|
||||||
|
adminDb := mc.db.Client().Database("admin")
|
||||||
|
result := adminDb.RunCommand(mc.ctx, bson.D{
|
||||||
|
{Key: "modifyChangeStreams", Value: 1},
|
||||||
|
{Key: "database", Value: mc.db.Name()},
|
||||||
|
{Key: "collection", Value: coll},
|
||||||
|
{Key: "enable", Value: true},
|
||||||
|
})
|
||||||
|
|
||||||
|
if result.Err() != nil {
|
||||||
|
logger.Println("mc.db.RunCommand failed :", result.Err(), mc.db.Name(), coll)
|
||||||
|
} else {
|
||||||
|
return mc.Collection(coll).Watch(mc.ctx, pipeline, opts...)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
logger.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return stream, err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (mc *MongoClient) Collection(collname CollectionName) *mongo.Collection {
|
func (mc *MongoClient) Collection(collname CollectionName) *mongo.Collection {
|
||||||
|
|||||||
36
server.go
36
server.go
@ -40,6 +40,13 @@ func init() {
|
|||||||
gob.Register([]any{})
|
gob.Register([]any{})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type ServerMuxInterface interface {
|
||||||
|
http.Handler
|
||||||
|
|
||||||
|
HandleFunc(pattern string, handler func(http.ResponseWriter, *http.Request))
|
||||||
|
Handle(pattern string, handler http.Handler)
|
||||||
|
}
|
||||||
|
|
||||||
const (
|
const (
|
||||||
// HTTPStatusReloginRequired : http status를 이걸 받으면 클라이언트는 로그아웃하고 로그인 화면으로 돌아가야 한다.
|
// HTTPStatusReloginRequired : http status를 이걸 받으면 클라이언트는 로그아웃하고 로그인 화면으로 돌아가야 한다.
|
||||||
HTTPStatusReloginRequired = 599
|
HTTPStatusReloginRequired = 599
|
||||||
@ -139,8 +146,19 @@ func isTlsEnabled(fileout ...*string) bool {
|
|||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func registUnhandledPattern(serveMux ServerMuxInterface) {
|
||||||
|
defer func() {
|
||||||
|
recover()
|
||||||
|
}()
|
||||||
|
|
||||||
|
serveMux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
logger.Println("page not found :", r.URL.Path)
|
||||||
|
w.WriteHeader(http.StatusNotFound)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
// NewHTTPServer :
|
// NewHTTPServer :
|
||||||
func NewHTTPServerWithPort(serveMux *http.ServeMux, port int) *Server {
|
func NewHTTPServerWithPort(serveMux ServerMuxInterface, port int) *Server {
|
||||||
if isTlsEnabled() && port == 80 {
|
if isTlsEnabled() && port == 80 {
|
||||||
port = 443
|
port = 443
|
||||||
}
|
}
|
||||||
@ -148,16 +166,21 @@ func NewHTTPServerWithPort(serveMux *http.ServeMux, port int) *Server {
|
|||||||
serveMux.HandleFunc(MakeHttpHandlerPattern("welcome"), welcomeHandler)
|
serveMux.HandleFunc(MakeHttpHandlerPattern("welcome"), welcomeHandler)
|
||||||
serveMux.HandleFunc(MakeHttpHandlerPattern("lb_health_chceck"), healthCheckHandler)
|
serveMux.HandleFunc(MakeHttpHandlerPattern("lb_health_chceck"), healthCheckHandler)
|
||||||
serveMux.HandleFunc(MakeHttpHandlerPattern("lb_health_check"), healthCheckHandler)
|
serveMux.HandleFunc(MakeHttpHandlerPattern("lb_health_check"), healthCheckHandler)
|
||||||
|
registUnhandledPattern(serveMux)
|
||||||
|
|
||||||
server := &Server{
|
server := &Server{
|
||||||
httpserver: &http.Server{Addr: addr, Handler: serveMux},
|
httpserver: &http.Server{
|
||||||
|
Addr: addr,
|
||||||
|
Handler: serveMux,
|
||||||
|
MaxHeaderBytes: 2 << 20, // 2 MB
|
||||||
|
},
|
||||||
}
|
}
|
||||||
server.httpserver.SetKeepAlivesEnabled(true)
|
server.httpserver.SetKeepAlivesEnabled(true)
|
||||||
|
|
||||||
return server
|
return server
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewHTTPServer(serveMux *http.ServeMux) *Server {
|
func NewHTTPServer(serveMux ServerMuxInterface) *Server {
|
||||||
|
|
||||||
// 시작시 자동으로 enable됨
|
// 시작시 자동으로 enable됨
|
||||||
if isTlsEnabled() && *portptr == 80 {
|
if isTlsEnabled() && *portptr == 80 {
|
||||||
@ -481,6 +504,13 @@ func ReadStringFormValue(r url.Values, key string) (string, bool) {
|
|||||||
return strval, len(strval) > 0
|
return strval, len(strval) > 0
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func ReadStringsFormValue(r url.Values, key string) ([]string, bool) {
|
||||||
|
if r.Has(key) {
|
||||||
|
return (map[string][]string)(r)[key], true
|
||||||
|
}
|
||||||
|
return nil, false
|
||||||
|
}
|
||||||
|
|
||||||
type encoder interface {
|
type encoder interface {
|
||||||
Encode(any) error
|
Encode(any) error
|
||||||
}
|
}
|
||||||
|
|||||||
@ -19,7 +19,7 @@ type Authorization struct {
|
|||||||
// by authorization provider
|
// by authorization provider
|
||||||
Platform string `bson:"p" json:"p"`
|
Platform string `bson:"p" json:"p"`
|
||||||
Uid string `bson:"u" json:"u"`
|
Uid string `bson:"u" json:"u"`
|
||||||
Email string `bson:"em" json:"em"`
|
Alias string `bson:"al" json:"al"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func (auth *Authorization) ToStrings() []string {
|
func (auth *Authorization) ToStrings() []string {
|
||||||
@ -27,7 +27,7 @@ func (auth *Authorization) ToStrings() []string {
|
|||||||
"a", auth.Account.Hex(),
|
"a", auth.Account.Hex(),
|
||||||
"p", auth.Platform,
|
"p", auth.Platform,
|
||||||
"u", auth.Uid,
|
"u", auth.Uid,
|
||||||
"em", auth.Email,
|
"al", auth.Alias,
|
||||||
"inv", auth.invalidated,
|
"inv", auth.invalidated,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -42,7 +42,7 @@ func MakeAuthrizationFromStringMap(src map[string]string) Authorization {
|
|||||||
Account: accid,
|
Account: accid,
|
||||||
Platform: src["p"],
|
Platform: src["p"],
|
||||||
Uid: src["u"],
|
Uid: src["u"],
|
||||||
Email: src["em"],
|
Alias: src["al"],
|
||||||
invalidated: src["inv"],
|
invalidated: src["inv"],
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -266,7 +266,7 @@ func (ws *WebsocketHandler) Cleanup() {
|
|||||||
ws.connWaitGroup.Wait()
|
ws.connWaitGroup.Wait()
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ws *WebsocketHandler) RegisterHandlers(serveMux *http.ServeMux, prefix string) error {
|
func (ws *WebsocketHandler) RegisterHandlers(serveMux gocommon.ServerMuxInterface, prefix string) error {
|
||||||
url := gocommon.MakeHttpHandlerPattern(prefix, "ws")
|
url := gocommon.MakeHttpHandlerPattern(prefix, "ws")
|
||||||
if *noAuthFlag {
|
if *noAuthFlag {
|
||||||
serveMux.HandleFunc(url, ws.upgrade_nosession)
|
serveMux.HandleFunc(url, ws.upgrade_nosession)
|
||||||
@ -586,9 +586,11 @@ func upgrade_core(ws *WebsocketHandler, conn *Conn, accid primitive.ObjectID, al
|
|||||||
ws.connWaitGroup.Done()
|
ws.connWaitGroup.Done()
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
logger.Println("wshandler connected :", accid)
|
||||||
for {
|
for {
|
||||||
messageType, r, err := c.innerConn.NextReader()
|
messageType, r, err := c.innerConn.NextReader()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
logger.Println("wshandler NextReader err :", messageType, err)
|
||||||
if ce, ok := err.(*websocket.CloseError); ok {
|
if ce, ok := err.(*websocket.CloseError); ok {
|
||||||
c.closeMessage = ce.Text
|
c.closeMessage = ce.Text
|
||||||
}
|
}
|
||||||
@ -610,6 +612,7 @@ func upgrade_core(ws *WebsocketHandler, conn *Conn, accid primitive.ObjectID, al
|
|||||||
ws.Call(newconn.sender, string(cmd), r)
|
ws.Call(newconn.sender, string(cmd), r)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
logger.Println("wshandler disconnected :", accid, c.closeMessage)
|
||||||
c.Conn = nil
|
c.Conn = nil
|
||||||
if c.closeMessage != ForceShutdownCloseMessage {
|
if c.closeMessage != ForceShutdownCloseMessage {
|
||||||
ws.connInOutChan <- c
|
ws.connInOutChan <- c
|
||||||
|
|||||||
@ -11,6 +11,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"go.mongodb.org/mongo-driver/bson/primitive"
|
"go.mongodb.org/mongo-driver/bson/primitive"
|
||||||
|
"repositories.action2quare.com/ayo/gocommon"
|
||||||
"repositories.action2quare.com/ayo/gocommon/logger"
|
"repositories.action2quare.com/ayo/gocommon/logger"
|
||||||
"repositories.action2quare.com/ayo/gocommon/session"
|
"repositories.action2quare.com/ayo/gocommon/session"
|
||||||
|
|
||||||
@ -18,7 +19,7 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
type WebsocketPeerHandler interface {
|
type WebsocketPeerHandler interface {
|
||||||
RegisterHandlers(serveMux *http.ServeMux, prefix string) error
|
RegisterHandlers(serveMux gocommon.ServerMuxInterface, prefix string) error
|
||||||
}
|
}
|
||||||
|
|
||||||
type peerCtorChannelValue struct {
|
type peerCtorChannelValue struct {
|
||||||
@ -164,7 +165,7 @@ func NewWebsocketPeerHandler[T PeerInterface](consumer session.Consumer, creator
|
|||||||
return wsh
|
return wsh
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ws *websocketPeerHandler[T]) RegisterHandlers(serveMux *http.ServeMux, prefix string) error {
|
func (ws *websocketPeerHandler[T]) RegisterHandlers(serveMux gocommon.ServerMuxInterface, prefix string) error {
|
||||||
if *noAuthFlag {
|
if *noAuthFlag {
|
||||||
serveMux.HandleFunc(prefix, ws.upgrade_noauth)
|
serveMux.HandleFunc(prefix, ws.upgrade_noauth)
|
||||||
} else {
|
} else {
|
||||||
|
|||||||
Reference in New Issue
Block a user