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"
|
||||
"runtime"
|
||||
"runtime/debug"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"repositories.action2quare.com/ayo/gocommon/flagx"
|
||||
@ -15,6 +16,7 @@ import (
|
||||
|
||||
var stdlogger *log.Logger
|
||||
var UseLogFile = flagx.Bool("logfile", false, "")
|
||||
var _ = flagx.Int("logprefix", 3, "0 : no_prefix, 1 : date, 2 : time, 3 : datetime")
|
||||
|
||||
func init() {
|
||||
binpath, _ := os.Executable()
|
||||
@ -23,7 +25,29 @@ func init() {
|
||||
var outWriter io.Writer
|
||||
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)
|
||||
if len(ext) > 0 {
|
||||
binname = binname[:len(binname)-len(ext)]
|
||||
@ -38,7 +62,11 @@ func init() {
|
||||
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{}) {
|
||||
|
||||
@ -135,7 +135,11 @@ func (pe *prometheusExporter) loop(ctx context.Context) {
|
||||
}
|
||||
|
||||
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 {
|
||||
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) {
|
||||
// mc.db.RunCommand()
|
||||
if len(opts) == 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 {
|
||||
|
||||
36
server.go
36
server.go
@ -40,6 +40,13 @@ func init() {
|
||||
gob.Register([]any{})
|
||||
}
|
||||
|
||||
type ServerMuxInterface interface {
|
||||
http.Handler
|
||||
|
||||
HandleFunc(pattern string, handler func(http.ResponseWriter, *http.Request))
|
||||
Handle(pattern string, handler http.Handler)
|
||||
}
|
||||
|
||||
const (
|
||||
// HTTPStatusReloginRequired : http status를 이걸 받으면 클라이언트는 로그아웃하고 로그인 화면으로 돌아가야 한다.
|
||||
HTTPStatusReloginRequired = 599
|
||||
@ -139,8 +146,19 @@ func isTlsEnabled(fileout ...*string) bool {
|
||||
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 :
|
||||
func NewHTTPServerWithPort(serveMux *http.ServeMux, port int) *Server {
|
||||
func NewHTTPServerWithPort(serveMux ServerMuxInterface, port int) *Server {
|
||||
if isTlsEnabled() && port == 80 {
|
||||
port = 443
|
||||
}
|
||||
@ -148,16 +166,21 @@ func NewHTTPServerWithPort(serveMux *http.ServeMux, port int) *Server {
|
||||
serveMux.HandleFunc(MakeHttpHandlerPattern("welcome"), welcomeHandler)
|
||||
serveMux.HandleFunc(MakeHttpHandlerPattern("lb_health_chceck"), healthCheckHandler)
|
||||
serveMux.HandleFunc(MakeHttpHandlerPattern("lb_health_check"), healthCheckHandler)
|
||||
registUnhandledPattern(serveMux)
|
||||
|
||||
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)
|
||||
|
||||
return server
|
||||
}
|
||||
|
||||
func NewHTTPServer(serveMux *http.ServeMux) *Server {
|
||||
func NewHTTPServer(serveMux ServerMuxInterface) *Server {
|
||||
|
||||
// 시작시 자동으로 enable됨
|
||||
if isTlsEnabled() && *portptr == 80 {
|
||||
@ -481,6 +504,13 @@ func ReadStringFormValue(r url.Values, key string) (string, bool) {
|
||||
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 {
|
||||
Encode(any) error
|
||||
}
|
||||
|
||||
@ -19,7 +19,7 @@ type Authorization struct {
|
||||
// by authorization provider
|
||||
Platform string `bson:"p" json:"p"`
|
||||
Uid string `bson:"u" json:"u"`
|
||||
Email string `bson:"em" json:"em"`
|
||||
Alias string `bson:"al" json:"al"`
|
||||
}
|
||||
|
||||
func (auth *Authorization) ToStrings() []string {
|
||||
@ -27,7 +27,7 @@ func (auth *Authorization) ToStrings() []string {
|
||||
"a", auth.Account.Hex(),
|
||||
"p", auth.Platform,
|
||||
"u", auth.Uid,
|
||||
"em", auth.Email,
|
||||
"al", auth.Alias,
|
||||
"inv", auth.invalidated,
|
||||
}
|
||||
}
|
||||
@ -42,7 +42,7 @@ func MakeAuthrizationFromStringMap(src map[string]string) Authorization {
|
||||
Account: accid,
|
||||
Platform: src["p"],
|
||||
Uid: src["u"],
|
||||
Email: src["em"],
|
||||
Alias: src["al"],
|
||||
invalidated: src["inv"],
|
||||
}
|
||||
}
|
||||
|
||||
@ -266,7 +266,7 @@ func (ws *WebsocketHandler) Cleanup() {
|
||||
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")
|
||||
if *noAuthFlag {
|
||||
serveMux.HandleFunc(url, ws.upgrade_nosession)
|
||||
@ -586,9 +586,11 @@ func upgrade_core(ws *WebsocketHandler, conn *Conn, accid primitive.ObjectID, al
|
||||
ws.connWaitGroup.Done()
|
||||
}()
|
||||
|
||||
logger.Println("wshandler connected :", accid)
|
||||
for {
|
||||
messageType, r, err := c.innerConn.NextReader()
|
||||
if err != nil {
|
||||
logger.Println("wshandler NextReader err :", messageType, err)
|
||||
if ce, ok := err.(*websocket.CloseError); ok {
|
||||
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)
|
||||
}
|
||||
}
|
||||
logger.Println("wshandler disconnected :", accid, c.closeMessage)
|
||||
c.Conn = nil
|
||||
if c.closeMessage != ForceShutdownCloseMessage {
|
||||
ws.connInOutChan <- c
|
||||
|
||||
@ -11,6 +11,7 @@ import (
|
||||
"time"
|
||||
|
||||
"go.mongodb.org/mongo-driver/bson/primitive"
|
||||
"repositories.action2quare.com/ayo/gocommon"
|
||||
"repositories.action2quare.com/ayo/gocommon/logger"
|
||||
"repositories.action2quare.com/ayo/gocommon/session"
|
||||
|
||||
@ -18,7 +19,7 @@ import (
|
||||
)
|
||||
|
||||
type WebsocketPeerHandler interface {
|
||||
RegisterHandlers(serveMux *http.ServeMux, prefix string) error
|
||||
RegisterHandlers(serveMux gocommon.ServerMuxInterface, prefix string) error
|
||||
}
|
||||
|
||||
type peerCtorChannelValue struct {
|
||||
@ -164,7 +165,7 @@ func NewWebsocketPeerHandler[T PeerInterface](consumer session.Consumer, creator
|
||||
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 {
|
||||
serveMux.HandleFunc(prefix, ws.upgrade_noauth)
|
||||
} else {
|
||||
|
||||
Reference in New Issue
Block a user