session consumer로 교체

2023-08-31 21:02:19 +09:00
parent cb5cd280b9
commit fd1502e52a
8 changed files with 106 additions and 110 deletions
--- a/core/group.go
+++ b/core/group.go
@ -53,6 +53,6 @@ func (afc *apiFuncsContainer) call(fn string, w http.ResponseWriter, r *http.Req

 type configDocument map[string]any
 type group interface {
-	Initialize(*subTavern, configDocument) error
+	Initialize(*Tavern, configDocument) error
 	ClientMessageReceived(*wshandler.Sender, wshandler.WebSocketMessageType, any)
 }
--- a/core/group_chat.go
+++ b/core/group_chat.go
@ -45,28 +45,28 @@ type groupChat struct {
 var accidHex func(primitive.ObjectID) string
 var accidstrHex func(string) string

-func (gc *groupChat) Initialize(sub *subTavern, cfg configDocument) error {
+func (gc *groupChat) Initialize(tv *Tavern, cfg configDocument) error {
 	rem, _ := json.Marshal(cfg)
 	if err := json.Unmarshal(rem, &gc.chatConfig); err != nil {
 		return err
 	}

 	gc.enterRoom = func(chanid channelID, accid accountID) {
-		sub.wsh.EnterRoom(sub.region, string(chanid), accid)
+		tv.wsh.EnterRoom(string(chanid), accid)
 	}
 	gc.leaveRoom = func(chanid channelID, accid accountID) {
-		sub.wsh.LeaveRoom(sub.region, string(chanid), accid)
+		tv.wsh.LeaveRoom(string(chanid), accid)
 	}
 	gc.sendUpstreamMessage = func(msg *wshandler.UpstreamMessage) {
-		sub.wsh.SendUpstreamMessage(sub.region, msg)
+		tv.wsh.SendUpstreamMessage(msg)
 	}

-	gc.rh = sub.redison
+	gc.rh = tv.redison

-	sub.apiFuncs.registApiFunction("FetchChattingChannels", gc.FetchChattingChannels)
-	sub.apiFuncs.registApiFunction("BroadcastMessageOnChannel", gc.BroadcastMessageOnChannel)
-	sub.apiFuncs.registApiFunction("QueryPlayerChattingChannel", gc.QueryPlayerChattingChannel)
-	sub.apiFuncs.registApiFunction("SendMessageOnChannel", gc.SendMessageOnChannel)
+	tv.apiFuncs.registApiFunction("FetchChattingChannels", gc.FetchChattingChannels)
+	tv.apiFuncs.registApiFunction("BroadcastMessageOnChannel", gc.BroadcastMessageOnChannel)
+	tv.apiFuncs.registApiFunction("QueryPlayerChattingChannel", gc.QueryPlayerChattingChannel)
+	tv.apiFuncs.registApiFunction("SendMessageOnChannel", gc.SendMessageOnChannel)
 	for name, cfg := range gc.chatConfig.Channels {
 		if cfg.Capacity == 0 {
 			cfg.Capacity = gc.chatConfig.DefaultCapacity
--- a/core/group_party.go
+++ b/core/group_party.go
@ -255,33 +255,33 @@ type groupParty struct {
 	rh                  *gocommon.RedisonHandler
 }

-func (gp *groupParty) Initialize(sub *subTavern, cfg configDocument) error {
+func (gp *groupParty) Initialize(tv *Tavern, cfg configDocument) error {
 	rem, _ := json.Marshal(cfg)
 	err := json.Unmarshal(rem, &gp.partyConfig)
 	if err != nil {
 		return err
 	}

-	gp.rh = sub.redison
+	gp.rh = tv.redison
 	gp.sendUpstreamMessage = func(msg *wshandler.UpstreamMessage) {
-		sub.wsh.SendUpstreamMessage(sub.region, msg)
+		tv.wsh.SendUpstreamMessage(msg)
 	}
 	gp.enterRoom = func(gid groupID, accid accountID) {
-		sub.wsh.EnterRoom(sub.region, gid.Hex(), accid)
+		tv.wsh.EnterRoom(gid.Hex(), accid)
 	}
 	gp.leaveRoom = func(gid groupID, accid accountID) {
-		sub.wsh.LeaveRoom(sub.region, gid.Hex(), accid)
+		tv.wsh.LeaveRoom(gid.Hex(), accid)
 	}

-	sub.apiFuncs.registApiFunction("JoinParty", gp.JoinParty)
-	sub.apiFuncs.registApiFunction("InviteToParty", gp.InviteToParty)
-	sub.apiFuncs.registApiFunction("AcceptPartyInvitation", gp.AcceptPartyInvitation)
-	sub.apiFuncs.registApiFunction("DenyPartyInvitation", gp.DenyPartyInvitation)
-	sub.apiFuncs.registApiFunction("QueryPartyMemberState", gp.QueryPartyMemberState)
-	sub.apiFuncs.registApiFunction("LeaveParty", gp.LeaveParty)
-	sub.apiFuncs.registApiFunction("UpdatePartyMemberDocument", gp.UpdatePartyMemberDocument)
-	sub.apiFuncs.registApiFunction("UpdatePartyDocument", gp.UpdatePartyDocument)
-	sub.apiFuncs.registApiFunction("QueryPartyMembers", gp.QueryPartyMembers)
+	tv.apiFuncs.registApiFunction("JoinParty", gp.JoinParty)
+	tv.apiFuncs.registApiFunction("InviteToParty", gp.InviteToParty)
+	tv.apiFuncs.registApiFunction("AcceptPartyInvitation", gp.AcceptPartyInvitation)
+	tv.apiFuncs.registApiFunction("DenyPartyInvitation", gp.DenyPartyInvitation)
+	tv.apiFuncs.registApiFunction("QueryPartyMemberState", gp.QueryPartyMemberState)
+	tv.apiFuncs.registApiFunction("LeaveParty", gp.LeaveParty)
+	tv.apiFuncs.registApiFunction("UpdatePartyMemberDocument", gp.UpdatePartyMemberDocument)
+	tv.apiFuncs.registApiFunction("UpdatePartyDocument", gp.UpdatePartyDocument)
+	tv.apiFuncs.registApiFunction("QueryPartyMembers", gp.QueryPartyMembers)

 	return nil
 }
--- a/core/tavern.go
+++ b/core/tavern.go
@ -66,7 +66,10 @@ func readBsonDoc(r io.Reader, src any) error {
 }

 type TavernConfig struct {
-	gocommon.RegionStorageConfig `json:",inline"`
+	gocommon.StorageAddr `json:"storage"`
+
+	SessionTTL     int64  `json:"maingate_session_ttl"`
+	SessionStorage string `json:"maingate_session_storage"`

 	Group            map[string]configDocument `json:"tavern_group_types"`
 	MaingateApiToken string                    `json:"maingate_api_token"`
@ -77,15 +80,9 @@ type TavernConfig struct {
 var config TavernConfig

 type Tavern struct {
-	subTaverns []*subTavern
-	wsh        *wshandler.WebsocketHandler
-}
-
-type subTavern struct {
+	wsh         *wshandler.WebsocketHandler
 	mongoClient gocommon.MongoClient
 	redison     *gocommon.RedisonHandler
-	wsh         *wshandler.WebsocketHandler
-	region      string
 	groups      map[string]group
 	apiFuncs    *apiFuncsContainer
 }
@ -124,6 +121,10 @@ func New(context context.Context, wsh *wshandler.WebsocketHandler, inconfig *Tav
 	config.macAddr = macaddr
 	tv := &Tavern{
 		wsh: wsh,
+		apiFuncs: &apiFuncsContainer{
+			normfuncs: make(map[string]apiFuncType),
+			funcs:     make(map[string][]apiFuncType),
+		},
 	}

 	if err = tv.prepare(context); err != nil {
@ -135,92 +136,65 @@ func New(context context.Context, wsh *wshandler.WebsocketHandler, inconfig *Tav
 }

 func (tv *Tavern) Cleanup() {
-	for _, st := range tv.subTaverns {
-		st.mongoClient.Close()
-	}
+	tv.mongoClient.Close()
 }

 func (tv *Tavern) prepare(ctx context.Context) error {
-	for region, addr := range config.RegionStorage {
-		var dbconn gocommon.MongoClient
-		var err error
+	redisClient, err := gocommon.NewRedisClient(config.StorageAddr.Redis["tavern"])
+	if err != nil {
+		return err
+	}

-		redisClient, err := gocommon.NewRedisClient(addr.Redis["tavern"])
-		if err != nil {
+	tv.redison = gocommon.NewRedisonHandler(redisClient.Context(), redisClient)
+
+	groups := make(map[string]group)
+	for typename, cfg := range config.Group {
+		gtype, ok := groupTypeContainer()[typename]
+		if !ok {
+			return fmt.Errorf("%s group type is not valid", typename)
+		}
+
+		if !gtype.Implements(reflect.TypeOf((*group)(nil)).Elem()) {
+			return fmt.Errorf("%s is not implement proper interface", typename)
+		}
+		ptrvalue := reflect.New(gtype.Elem())
+		instance := ptrvalue.Interface().(group)
+		if err := instance.Initialize(tv, cfg); err != nil {
 			return err
 		}
-
-		redison := gocommon.NewRedisonHandler(redisClient.Context(), redisClient)
-
-		sub := &subTavern{
-			wsh:         tv.wsh,
-			mongoClient: dbconn,
-			redison:     redison,
-			region:      region,
-			apiFuncs: &apiFuncsContainer{
-				normfuncs: make(map[string]apiFuncType),
-				funcs:     make(map[string][]apiFuncType),
-			},
-		}
-
-		groups := make(map[string]group)
-		for typename, cfg := range config.Group {
-			gtype, ok := groupTypeContainer()[typename]
-			if !ok {
-				return fmt.Errorf("%s group type is not valid", typename)
-			}
-
-			if !gtype.Implements(reflect.TypeOf((*group)(nil)).Elem()) {
-				return fmt.Errorf("%s is not implement proper interface", typename)
-			}
-			ptrvalue := reflect.New(gtype.Elem())
-			instance := ptrvalue.Interface().(group)
-			if err := instance.Initialize(sub, cfg); err != nil {
-				return err
-			}
-			groups[typename] = instance
-		}
-
-		sub.groups = groups
-		sub.apiFuncs.normalize()
-
-		tv.subTaverns = append(tv.subTaverns, sub)
+		groups[typename] = instance
 	}

+	tv.groups = groups
+	tv.apiFuncs.normalize()
+
 	return nil
 }

 func (tv *Tavern) RegisterHandlers(ctx context.Context, serveMux *http.ServeMux, prefix string) error {
-	for _, sub := range tv.subTaverns {
-		tv.wsh.RegisterReceiver(sub.region, sub)
-		var pattern string
-		if sub.region == "default" {
-			pattern = gocommon.MakeHttpHandlerPattern(prefix, "api")
-		} else {
-			pattern = gocommon.MakeHttpHandlerPattern(prefix, sub.region, "api")
-		}
-		serveMux.HandleFunc(pattern, sub.api)
-	}
+	tv.wsh.RegisterReceiver(tv)
+	pattern := gocommon.MakeHttpHandlerPattern(prefix, "api")
+	serveMux.HandleFunc(pattern, tv.api)

 	return nil
 }

-func (sub *subTavern) OnClientMessageReceived(sender *wshandler.Sender, messageType wshandler.WebSocketMessageType, body io.Reader) {
+func (tv *Tavern) OnClientMessageReceived(sender *wshandler.Sender, messageType wshandler.WebSocketMessageType, body io.Reader) {
 	if messageType == wshandler.Connected {
 		logger.Println("OnClientMessageReceived : connected ", sender.Accid.Hex())
-		sub.redison.HSet(sub.redison.Context(), sender.Accid.Hex(), "_ts", time.Now().UTC().Unix()).Result()
-		for _, gt := range sub.groups {
+		tv.redison.HSet(tv.redison.Context(), sender.Accid.Hex(), "_ts", time.Now().UTC().Unix()).Result()
+		for _, gt := range tv.groups {
 			gt.ClientMessageReceived(sender, messageType, nil)
 		}
 	} else if messageType == wshandler.Disconnected {
 		var rooms []string
 		dec := json.NewDecoder(body)
 		if err := dec.Decode(&rooms); err == nil {
-			for _, gt := range sub.groups {
+			for _, gt := range tv.groups {
 				gt.ClientMessageReceived(sender, messageType, rooms)
 			}
 		}
-		sub.redison.Del(sub.redison.Context(), sender.Accid.Hex()).Result()
+		tv.redison.Del(tv.redison.Context(), sender.Accid.Hex()).Result()
 		logger.Println("OnClientMessageReceived : disconnected ", sender.Accid.Hex())
 	} else if messageType == wshandler.BinaryMessage {
 		var commandline []any
@ -230,13 +204,13 @@ func (sub *subTavern) OnClientMessageReceived(sender *wshandler.Sender, messageT
 			args := commandline[1:]
 			switch cmd {
 			case "EnterChannel":
-				sub.wsh.EnterRoom(sub.region, args[0].(string), sender.Accid)
+				tv.wsh.EnterRoom(args[0].(string), sender.Accid)

 			case "LeaveChannel":
-				sub.wsh.LeaveRoom(sub.region, args[0].(string), sender.Accid)
+				tv.wsh.LeaveRoom(args[0].(string), sender.Accid)

 			default:
-				for _, gt := range sub.groups {
+				for _, gt := range tv.groups {
 					gt.ClientMessageReceived(sender, messageType, commandline)
 				}
 			}
@ -244,29 +218,29 @@ func (sub *subTavern) OnClientMessageReceived(sender *wshandler.Sender, messageT
 	}
 }

-func (sub *subTavern) OnRoomCreated(region, name string) {
-	cnt, err := sub.redison.IncrBy(sub.redison.Context(), "_ref_"+name, 1).Result()
+func (tv *Tavern) OnRoomCreated(name string) {
+	cnt, err := tv.redison.IncrBy(tv.redison.Context(), "_ref_"+name, 1).Result()
 	if err != nil && !errors.Is(err, redis.Nil) {
 		logger.Println("OnRoomCreated JSONSet failed :", err)
 		return
 	}

 	if cnt == 1 {
-		sub.redison.JSONSet(name, "$", map[string]any{}, gocommon.RedisonSetOptionNX)
+		tv.redison.JSONSet(name, "$", map[string]any{}, gocommon.RedisonSetOptionNX)
 	}
 }

-func (sub *subTavern) OnRoomDestroyed(region, name string) {
-	cnt, err := sub.redison.IncrBy(sub.redison.Context(), "_ref_"+name, -1).Result()
+func (tv *Tavern) OnRoomDestroyed(name string) {
+	cnt, err := tv.redison.IncrBy(tv.redison.Context(), "_ref_"+name, -1).Result()
 	if err != nil {
 		logger.Println("OnRoomDestroyed JSONNumIncrBy failed :", err)
 	} else if cnt == 0 {
-		sub.redison.Del(sub.redison.Context(), "_ref_"+name)
-		sub.redison.JSONDel(name, "$")
+		tv.redison.Del(tv.redison.Context(), "_ref_"+name)
+		tv.redison.JSONDel(name, "$")
 	}
 }

-func (sub *subTavern) api(w http.ResponseWriter, r *http.Request) {
+func (tv *Tavern) api(w http.ResponseWriter, r *http.Request) {
 	defer func() {
 		s := recover()
 		if s != nil {
@ -294,5 +268,5 @@ func (sub *subTavern) api(w http.ResponseWriter, r *http.Request) {
 		return
 	}

-	sub.apiFuncs.call(operation, w, r)
+	tv.apiFuncs.call(operation, w, r)
 }