commit
17f9eb40fd
@ -0,0 +1,419 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"os/signal"
|
||||
"path"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"runtime"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/mitchellh/mapstructure"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/api"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/msggateway"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/msgtransfer"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/rpc/auth"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/rpc/conversation"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/rpc/group"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/rpc/msg"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/rpc/relation"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/rpc/third"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/rpc/user"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/tools/cron"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
|
||||
"github.com/openimsdk/open-im-server/v3/version"
|
||||
"github.com/openimsdk/tools/discovery"
|
||||
"github.com/openimsdk/tools/discovery/standalone"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"github.com/openimsdk/tools/system/program"
|
||||
"github.com/openimsdk/tools/utils/datautil"
|
||||
"github.com/openimsdk/tools/utils/network"
|
||||
"github.com/spf13/viper"
|
||||
"google.golang.org/grpc"
|
||||
)
|
||||
|
||||
func init() {
|
||||
config.SetStandalone()
|
||||
prommetrics.RegistryAll()
|
||||
}
|
||||
|
||||
func main() {
|
||||
var configPath string
|
||||
flag.StringVar(&configPath, "c", "", "config path")
|
||||
flag.Parse()
|
||||
if configPath == "" {
|
||||
_, _ = fmt.Fprintln(os.Stderr, "config path is empty")
|
||||
os.Exit(1)
|
||||
return
|
||||
}
|
||||
cmd := newCmds(configPath)
|
||||
putCmd(cmd, false, auth.Start)
|
||||
putCmd(cmd, false, conversation.Start)
|
||||
putCmd(cmd, false, relation.Start)
|
||||
putCmd(cmd, false, group.Start)
|
||||
putCmd(cmd, false, msg.Start)
|
||||
putCmd(cmd, false, third.Start)
|
||||
putCmd(cmd, false, user.Start)
|
||||
putCmd(cmd, false, push.Start)
|
||||
putCmd(cmd, true, msggateway.Start)
|
||||
putCmd(cmd, true, msgtransfer.Start)
|
||||
putCmd(cmd, true, api.Start)
|
||||
putCmd(cmd, true, cron.Start)
|
||||
ctx := context.Background()
|
||||
if err := cmd.run(ctx); err != nil {
|
||||
_, _ = fmt.Fprintf(os.Stderr, "server exit %s", err)
|
||||
os.Exit(1)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func newCmds(confPath string) *cmds {
|
||||
return &cmds{confPath: confPath}
|
||||
}
|
||||
|
||||
type cmdName struct {
|
||||
Name string
|
||||
Func func(ctx context.Context) error
|
||||
Block bool
|
||||
}
|
||||
type cmds struct {
|
||||
confPath string
|
||||
cmds []cmdName
|
||||
config config.AllConfig
|
||||
conf map[string]reflect.Value
|
||||
}
|
||||
|
||||
func (x *cmds) getTypePath(typ reflect.Type) string {
|
||||
return path.Join(typ.PkgPath(), typ.Name())
|
||||
}
|
||||
|
||||
func (x *cmds) initDiscovery() {
|
||||
x.config.Discovery.Enable = "standalone"
|
||||
vof := reflect.ValueOf(&x.config.Discovery.RpcService).Elem()
|
||||
tof := reflect.TypeOf(&x.config.Discovery.RpcService).Elem()
|
||||
num := tof.NumField()
|
||||
for i := 0; i < num; i++ {
|
||||
field := tof.Field(i)
|
||||
if !field.IsExported() {
|
||||
continue
|
||||
}
|
||||
if field.Type.Kind() != reflect.String {
|
||||
continue
|
||||
}
|
||||
vof.Field(i).SetString(field.Name)
|
||||
}
|
||||
}
|
||||
|
||||
func (x *cmds) initAllConfig() error {
|
||||
x.conf = make(map[string]reflect.Value)
|
||||
vof := reflect.ValueOf(&x.config).Elem()
|
||||
num := vof.NumField()
|
||||
for i := 0; i < num; i++ {
|
||||
field := vof.Field(i)
|
||||
for ptr := true; ptr; {
|
||||
if field.Kind() == reflect.Ptr {
|
||||
field = field.Elem()
|
||||
} else {
|
||||
ptr = false
|
||||
}
|
||||
}
|
||||
x.conf[x.getTypePath(field.Type())] = field
|
||||
val := field.Addr().Interface()
|
||||
name := val.(interface{ GetConfigFileName() string }).GetConfigFileName()
|
||||
confData, err := os.ReadFile(filepath.Join(x.confPath, name))
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
v := viper.New()
|
||||
v.SetConfigType("yaml")
|
||||
if err := v.ReadConfig(bytes.NewReader(confData)); err != nil {
|
||||
return err
|
||||
}
|
||||
opt := func(conf *mapstructure.DecoderConfig) {
|
||||
conf.TagName = config.StructTagName
|
||||
}
|
||||
if err := v.Unmarshal(val, opt); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
x.initDiscovery()
|
||||
x.config.Redis.Disable = false
|
||||
x.config.LocalCache = config.LocalCache{}
|
||||
config.InitNotification(&x.config.Notification)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *cmds) parseConf(conf any) error {
|
||||
vof := reflect.ValueOf(conf)
|
||||
for {
|
||||
if vof.Kind() == reflect.Ptr {
|
||||
vof = vof.Elem()
|
||||
} else {
|
||||
break
|
||||
}
|
||||
}
|
||||
tof := vof.Type()
|
||||
numField := vof.NumField()
|
||||
for i := 0; i < numField; i++ {
|
||||
typeField := tof.Field(i)
|
||||
if !typeField.IsExported() {
|
||||
continue
|
||||
}
|
||||
field := vof.Field(i)
|
||||
pkt := x.getTypePath(field.Type())
|
||||
val, ok := x.conf[pkt]
|
||||
if !ok {
|
||||
switch field.Interface().(type) {
|
||||
case config.Index:
|
||||
case config.Path:
|
||||
field.SetString(x.confPath)
|
||||
case config.AllConfig:
|
||||
field.Set(reflect.ValueOf(x.config))
|
||||
case *config.AllConfig:
|
||||
field.Set(reflect.ValueOf(&x.config))
|
||||
default:
|
||||
return fmt.Errorf("config field %s %s not found", vof.Type().Name(), typeField.Name)
|
||||
}
|
||||
continue
|
||||
}
|
||||
field.Set(val)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *cmds) add(name string, block bool, fn func(ctx context.Context) error) {
|
||||
x.cmds = append(x.cmds, cmdName{Name: name, Block: block, Func: fn})
|
||||
}
|
||||
|
||||
func (x *cmds) initLog() error {
|
||||
conf := x.config.Log
|
||||
if err := log.InitLoggerFromConfig(
|
||||
"openim-server",
|
||||
program.GetProcessName(),
|
||||
"", "",
|
||||
conf.RemainLogLevel,
|
||||
conf.IsStdout,
|
||||
conf.IsJson,
|
||||
conf.StorageLocation,
|
||||
conf.RemainRotationCount,
|
||||
conf.RotationTime,
|
||||
strings.TrimSpace(version.Version),
|
||||
conf.IsSimplify,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
|
||||
}
|
||||
|
||||
func (x *cmds) run(ctx context.Context) error {
|
||||
if len(x.cmds) == 0 {
|
||||
return fmt.Errorf("no command to run")
|
||||
}
|
||||
if err := x.initAllConfig(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := x.initLog(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancelCause(ctx)
|
||||
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
log.ZError(ctx, "context server exit cause", context.Cause(ctx))
|
||||
}()
|
||||
|
||||
if prometheus := x.config.API.Prometheus; prometheus.Enable {
|
||||
var (
|
||||
port int
|
||||
err error
|
||||
)
|
||||
if !prometheus.AutoSetPorts {
|
||||
port, err = datautil.GetElemByIndex(prometheus.Ports, 0)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
ip, err := network.GetLocalIP()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
|
||||
if err != nil {
|
||||
return fmt.Errorf("prometheus listen %d error %w", port, err)
|
||||
}
|
||||
defer listener.Close()
|
||||
log.ZDebug(ctx, "prometheus start", "addr", listener.Addr())
|
||||
target, err := json.Marshal(prommetrics.BuildDefaultTarget(ip, listener.Addr().(*net.TCPAddr).Port))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := standalone.GetKeyValue().SetKey(ctx, prommetrics.BuildDiscoveryKey(prommetrics.APIKeyName), target); err != nil {
|
||||
return err
|
||||
}
|
||||
go func() {
|
||||
err := prommetrics.Start(listener)
|
||||
if err == nil {
|
||||
err = fmt.Errorf("http done")
|
||||
}
|
||||
cancel(fmt.Errorf("prometheus %w", err))
|
||||
}()
|
||||
}
|
||||
|
||||
go func() {
|
||||
sigs := make(chan os.Signal, 1)
|
||||
signal.Notify(sigs, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case val := <-sigs:
|
||||
log.ZDebug(ctx, "recv signal", "signal", val.String())
|
||||
cancel(fmt.Errorf("signal %s", val.String()))
|
||||
}
|
||||
}()
|
||||
|
||||
for i := range x.cmds {
|
||||
cmd := x.cmds[i]
|
||||
if cmd.Block {
|
||||
continue
|
||||
}
|
||||
if err := cmd.Func(ctx); err != nil {
|
||||
cancel(fmt.Errorf("server %s exit %w", cmd.Name, err))
|
||||
return err
|
||||
}
|
||||
go func() {
|
||||
if cmd.Block {
|
||||
cancel(fmt.Errorf("server %s exit", cmd.Name))
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
var wait cmdManger
|
||||
for i := range x.cmds {
|
||||
cmd := x.cmds[i]
|
||||
if !cmd.Block {
|
||||
continue
|
||||
}
|
||||
wait.Start(cmd.Name)
|
||||
go func() {
|
||||
defer wait.Shutdown(cmd.Name)
|
||||
if err := cmd.Func(ctx); err != nil {
|
||||
cancel(fmt.Errorf("server %s exit %w", cmd.Name, err))
|
||||
return
|
||||
}
|
||||
cancel(fmt.Errorf("server %s exit", cmd.Name))
|
||||
}()
|
||||
}
|
||||
<-ctx.Done()
|
||||
exitCause := context.Cause(ctx)
|
||||
log.ZWarn(ctx, "notification of service closure", exitCause)
|
||||
done := wait.Wait()
|
||||
timeout := time.NewTimer(time.Second * 10)
|
||||
defer timeout.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-timeout.C:
|
||||
log.ZWarn(ctx, "server exit timeout", nil, "running", wait.Running())
|
||||
return exitCause
|
||||
case _, ok := <-done:
|
||||
if ok {
|
||||
log.ZWarn(ctx, "waiting for the service to exit", nil, "running", wait.Running())
|
||||
} else {
|
||||
log.ZInfo(ctx, "all server exit done")
|
||||
return exitCause
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func putCmd[C any](cmd *cmds, block bool, fn func(ctx context.Context, config *C, client discovery.Conn, server grpc.ServiceRegistrar) error) {
|
||||
name := path.Base(runtime.FuncForPC(reflect.ValueOf(fn).Pointer()).Name())
|
||||
if index := strings.Index(name, "."); index >= 0 {
|
||||
name = name[:index]
|
||||
}
|
||||
cmd.add(name, block, func(ctx context.Context) error {
|
||||
var conf C
|
||||
if err := cmd.parseConf(&conf); err != nil {
|
||||
return err
|
||||
}
|
||||
return fn(ctx, &conf, standalone.GetDiscoveryConn(), standalone.GetServiceRegistrar())
|
||||
})
|
||||
}
|
||||
|
||||
type cmdManger struct {
|
||||
lock sync.Mutex
|
||||
done chan struct{}
|
||||
count int
|
||||
names map[string]struct{}
|
||||
}
|
||||
|
||||
func (x *cmdManger) Start(name string) {
|
||||
x.lock.Lock()
|
||||
defer x.lock.Unlock()
|
||||
if x.names == nil {
|
||||
x.names = make(map[string]struct{})
|
||||
}
|
||||
if x.done == nil {
|
||||
x.done = make(chan struct{}, 1)
|
||||
}
|
||||
if _, ok := x.names[name]; ok {
|
||||
panic(fmt.Errorf("cmd %s already exists", name))
|
||||
}
|
||||
x.count++
|
||||
x.names[name] = struct{}{}
|
||||
}
|
||||
|
||||
func (x *cmdManger) Shutdown(name string) {
|
||||
x.lock.Lock()
|
||||
defer x.lock.Unlock()
|
||||
if _, ok := x.names[name]; !ok {
|
||||
panic(fmt.Errorf("cmd %s not exists", name))
|
||||
}
|
||||
delete(x.names, name)
|
||||
x.count--
|
||||
if x.count == 0 {
|
||||
close(x.done)
|
||||
} else {
|
||||
select {
|
||||
case x.done <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (x *cmdManger) Wait() <-chan struct{} {
|
||||
x.lock.Lock()
|
||||
defer x.lock.Unlock()
|
||||
if x.count == 0 || x.done == nil {
|
||||
tmp := make(chan struct{})
|
||||
close(tmp)
|
||||
return tmp
|
||||
}
|
||||
return x.done
|
||||
}
|
||||
|
||||
func (x *cmdManger) Running() []string {
|
||||
x.lock.Lock()
|
||||
defer x.lock.Unlock()
|
||||
names := make([]string, 0, len(x.names))
|
||||
for name := range x.names {
|
||||
names = append(names, name)
|
||||
}
|
||||
return names
|
||||
}
|
@ -1,12 +1,13 @@
|
||||
package tools
|
||||
package cron
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/protocol/msg"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"github.com/openimsdk/tools/mcontext"
|
||||
"os"
|
||||
"time"
|
||||
)
|
||||
|
||||
func (c *cronServer) deleteMsg() {
|
@ -1,12 +1,13 @@
|
||||
package tools
|
||||
package cron
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/protocol/third"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"github.com/openimsdk/tools/mcontext"
|
||||
"os"
|
||||
"time"
|
||||
)
|
||||
|
||||
func (c *cronServer) clearS3() {
|
@ -1,12 +1,13 @@
|
||||
package tools
|
||||
package cron
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
pbconversation "github.com/openimsdk/protocol/conversation"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"github.com/openimsdk/tools/mcontext"
|
||||
"os"
|
||||
"time"
|
||||
)
|
||||
|
||||
func (c *cronServer) clearUserMsg() {
|
@ -0,0 +1,11 @@
|
||||
package config
|
||||
|
||||
var standalone bool
|
||||
|
||||
func SetStandalone() {
|
||||
standalone = true
|
||||
}
|
||||
|
||||
func Standalone() bool {
|
||||
return standalone
|
||||
}
|
@ -1,31 +0,0 @@
|
||||
package prommetrics
|
||||
|
||||
import "fmt"
|
||||
|
||||
const (
|
||||
APIKeyName = "api"
|
||||
MessageTransferKeyName = "message-transfer"
|
||||
)
|
||||
|
||||
type Target struct {
|
||||
Target string `json:"target"`
|
||||
Labels map[string]string `json:"labels"`
|
||||
}
|
||||
|
||||
type RespTarget struct {
|
||||
Targets []string `json:"targets"`
|
||||
Labels map[string]string `json:"labels"`
|
||||
}
|
||||
|
||||
func BuildDiscoveryKey(name string) string {
|
||||
return fmt.Sprintf("%s/%s/%s", "openim", "prometheus_discovery", name)
|
||||
}
|
||||
|
||||
func BuildDefaultTarget(host string, ip int) Target {
|
||||
return Target{
|
||||
Target: fmt.Sprintf("%s:%d", host, ip),
|
||||
Labels: map[string]string{
|
||||
"namespace": "default",
|
||||
},
|
||||
}
|
||||
}
|
@ -1,15 +0,0 @@
|
||||
// Copyright © 2024 OpenIM. All rights reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package prommetrics // import "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
|
@ -1,15 +0,0 @@
|
||||
// Copyright © 2024 OpenIM. All rights reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package redispubsub // import "github.com/openimsdk/open-im-server/v3/pkg/common/redispubsub"
|
@ -1,30 +0,0 @@
|
||||
// Copyright © 2024 OpenIM. All rights reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package redispubsub
|
||||
|
||||
import "github.com/redis/go-redis/v9"
|
||||
|
||||
type Publisher struct {
|
||||
client redis.UniversalClient
|
||||
channel string
|
||||
}
|
||||
|
||||
func NewPublisher(client redis.UniversalClient, channel string) *Publisher {
|
||||
return &Publisher{client: client, channel: channel}
|
||||
}
|
||||
|
||||
func (p *Publisher) Publish(message string) error {
|
||||
return p.client.Publish(ctx, p.channel, message).Err()
|
||||
}
|
@ -1,49 +0,0 @@
|
||||
// Copyright © 2024 OpenIM. All rights reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package redispubsub
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
var ctx = context.Background()
|
||||
|
||||
type Subscriber struct {
|
||||
client redis.UniversalClient
|
||||
channel string
|
||||
}
|
||||
|
||||
func NewSubscriber(client redis.UniversalClient, channel string) *Subscriber {
|
||||
return &Subscriber{client: client, channel: channel}
|
||||
}
|
||||
|
||||
func (s *Subscriber) OnMessage(ctx context.Context, callback func(string)) error {
|
||||
messageChannel := s.client.Subscribe(ctx, s.channel).Channel()
|
||||
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case msg := <-messageChannel:
|
||||
callback(msg.Payload)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return nil
|
||||
}
|
@ -1,15 +0,0 @@
|
||||
// Copyright © 2024 OpenIM. All rights reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package startrpc // import "github.com/openimsdk/open-im-server/v3/pkg/common/startrpc"
|
@ -0,0 +1,50 @@
|
||||
package mcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||
"github.com/openimsdk/tools/s3/minio"
|
||||
)
|
||||
|
||||
func NewMinioCache(cache database.Cache) minio.Cache {
|
||||
return &minioCache{
|
||||
cache: cache,
|
||||
expireTime: time.Hour * 24 * 7,
|
||||
}
|
||||
}
|
||||
|
||||
type minioCache struct {
|
||||
cache database.Cache
|
||||
expireTime time.Duration
|
||||
}
|
||||
|
||||
func (g *minioCache) getObjectImageInfoKey(key string) string {
|
||||
return cachekey.GetObjectImageInfoKey(key)
|
||||
}
|
||||
|
||||
func (g *minioCache) getMinioImageThumbnailKey(key string, format string, width int, height int) string {
|
||||
return cachekey.GetMinioImageThumbnailKey(key, format, width, height)
|
||||
}
|
||||
|
||||
func (g *minioCache) DelObjectImageInfoKey(ctx context.Context, keys ...string) error {
|
||||
ks := make([]string, 0, len(keys))
|
||||
for _, key := range keys {
|
||||
ks = append(ks, g.getObjectImageInfoKey(key))
|
||||
}
|
||||
return g.cache.Del(ctx, ks)
|
||||
}
|
||||
|
||||
func (g *minioCache) DelImageThumbnailKey(ctx context.Context, key string, format string, width int, height int) error {
|
||||
return g.cache.Del(ctx, []string{g.getMinioImageThumbnailKey(key, format, width, height)})
|
||||
}
|
||||
|
||||
func (g *minioCache) GetImageObjectKeyInfo(ctx context.Context, key string, fn func(ctx context.Context) (*minio.ImageInfo, error)) (*minio.ImageInfo, error) {
|
||||
return getCache[*minio.ImageInfo](ctx, g.cache, g.getObjectImageInfoKey(key), g.expireTime, fn)
|
||||
}
|
||||
|
||||
func (g *minioCache) GetThumbnailKey(ctx context.Context, key string, format string, width int, height int, minioCache func(ctx context.Context) (string, error)) (string, error) {
|
||||
return getCache[string](ctx, g.cache, g.getMinioImageThumbnailKey(key, format, width, height), g.expireTime, minioCache)
|
||||
}
|
@ -0,0 +1,132 @@
|
||||
package mcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/localcache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/localcache/lru"
|
||||
"github.com/openimsdk/tools/errs"
|
||||
"github.com/openimsdk/tools/utils/datautil"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
var (
|
||||
memMsgCache lru.LRU[string, *model.MsgInfoModel]
|
||||
initMemMsgCache sync.Once
|
||||
)
|
||||
|
||||
func NewMsgCache(cache database.Cache, msgDocDatabase database.Msg) cache.MsgCache {
|
||||
initMemMsgCache.Do(func() {
|
||||
memMsgCache = lru.NewLayLRU[string, *model.MsgInfoModel](1024*8, time.Hour, time.Second*10, localcache.EmptyTarget{}, nil)
|
||||
})
|
||||
return &msgCache{
|
||||
cache: cache,
|
||||
msgDocDatabase: msgDocDatabase,
|
||||
memMsgCache: memMsgCache,
|
||||
}
|
||||
}
|
||||
|
||||
type msgCache struct {
|
||||
cache database.Cache
|
||||
msgDocDatabase database.Msg
|
||||
memMsgCache lru.LRU[string, *model.MsgInfoModel]
|
||||
}
|
||||
|
||||
func (x *msgCache) getSendMsgKey(id string) string {
|
||||
return cachekey.GetSendMsgKey(id)
|
||||
}
|
||||
|
||||
func (x *msgCache) SetSendMsgStatus(ctx context.Context, id string, status int32) error {
|
||||
return x.cache.Set(ctx, x.getSendMsgKey(id), strconv.Itoa(int(status)), time.Hour*24)
|
||||
}
|
||||
|
||||
func (x *msgCache) GetSendMsgStatus(ctx context.Context, id string) (int32, error) {
|
||||
key := x.getSendMsgKey(id)
|
||||
res, err := x.cache.Get(ctx, []string{key})
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
val, ok := res[key]
|
||||
if !ok {
|
||||
return 0, errs.Wrap(redis.Nil)
|
||||
}
|
||||
status, err := strconv.Atoi(val)
|
||||
if err != nil {
|
||||
return 0, errs.WrapMsg(err, "GetSendMsgStatus strconv.Atoi error", "val", val)
|
||||
}
|
||||
return int32(status), nil
|
||||
}
|
||||
|
||||
func (x *msgCache) getMsgCacheKey(conversationID string, seq int64) string {
|
||||
return cachekey.GetMsgCacheKey(conversationID, seq)
|
||||
|
||||
}
|
||||
|
||||
func (x *msgCache) GetMessageBySeqs(ctx context.Context, conversationID string, seqs []int64) ([]*model.MsgInfoModel, error) {
|
||||
if len(seqs) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
keys := make([]string, 0, len(seqs))
|
||||
keySeq := make(map[string]int64, len(seqs))
|
||||
for _, seq := range seqs {
|
||||
key := x.getMsgCacheKey(conversationID, seq)
|
||||
keys = append(keys, key)
|
||||
keySeq[key] = seq
|
||||
}
|
||||
res, err := x.memMsgCache.GetBatch(keys, func(keys []string) (map[string]*model.MsgInfoModel, error) {
|
||||
findSeqs := make([]int64, 0, len(keys))
|
||||
for _, key := range keys {
|
||||
seq, ok := keySeq[key]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
findSeqs = append(findSeqs, seq)
|
||||
}
|
||||
res, err := x.msgDocDatabase.FindSeqs(ctx, conversationID, seqs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
kv := make(map[string]*model.MsgInfoModel)
|
||||
for i := range res {
|
||||
msg := res[i]
|
||||
if msg == nil || msg.Msg == nil || msg.Msg.Seq <= 0 {
|
||||
continue
|
||||
}
|
||||
key := x.getMsgCacheKey(conversationID, msg.Msg.Seq)
|
||||
kv[key] = msg
|
||||
}
|
||||
return kv, nil
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return datautil.Values(res), nil
|
||||
}
|
||||
|
||||
func (x msgCache) DelMessageBySeqs(ctx context.Context, conversationID string, seqs []int64) error {
|
||||
if len(seqs) == 0 {
|
||||
return nil
|
||||
}
|
||||
for _, seq := range seqs {
|
||||
x.memMsgCache.Del(x.getMsgCacheKey(conversationID, seq))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *msgCache) SetMessageBySeqs(ctx context.Context, conversationID string, msgs []*model.MsgInfoModel) error {
|
||||
for i := range msgs {
|
||||
msg := msgs[i]
|
||||
if msg == nil || msg.Msg == nil || msg.Msg.Seq <= 0 {
|
||||
continue
|
||||
}
|
||||
x.memMsgCache.Set(x.getMsgCacheKey(conversationID, msg.Msg.Seq), msg)
|
||||
}
|
||||
return nil
|
||||
}
|
@ -0,0 +1,82 @@
|
||||
package mcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
||||
)
|
||||
|
||||
var (
|
||||
globalOnlineCache cache.OnlineCache
|
||||
globalOnlineOnce sync.Once
|
||||
)
|
||||
|
||||
func NewOnlineCache() cache.OnlineCache {
|
||||
globalOnlineOnce.Do(func() {
|
||||
globalOnlineCache = &onlineCache{
|
||||
user: make(map[string]map[int32]struct{}),
|
||||
}
|
||||
})
|
||||
return globalOnlineCache
|
||||
}
|
||||
|
||||
type onlineCache struct {
|
||||
lock sync.RWMutex
|
||||
user map[string]map[int32]struct{}
|
||||
}
|
||||
|
||||
func (x *onlineCache) GetOnline(ctx context.Context, userID string) ([]int32, error) {
|
||||
x.lock.RLock()
|
||||
defer x.lock.RUnlock()
|
||||
pSet, ok := x.user[userID]
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
res := make([]int32, 0, len(pSet))
|
||||
for k := range pSet {
|
||||
res = append(res, k)
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (x *onlineCache) SetUserOnline(ctx context.Context, userID string, online, offline []int32) error {
|
||||
x.lock.Lock()
|
||||
defer x.lock.Unlock()
|
||||
pSet, ok := x.user[userID]
|
||||
if ok {
|
||||
for _, p := range offline {
|
||||
delete(pSet, p)
|
||||
}
|
||||
}
|
||||
if len(online) > 0 {
|
||||
if !ok {
|
||||
pSet = make(map[int32]struct{})
|
||||
x.user[userID] = pSet
|
||||
}
|
||||
for _, p := range online {
|
||||
pSet[p] = struct{}{}
|
||||
}
|
||||
}
|
||||
if len(pSet) == 0 {
|
||||
delete(x.user, userID)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *onlineCache) GetAllOnlineUsers(ctx context.Context, cursor uint64) (map[string][]int32, uint64, error) {
|
||||
if cursor != 0 {
|
||||
return nil, 0, nil
|
||||
}
|
||||
x.lock.RLock()
|
||||
defer x.lock.RUnlock()
|
||||
res := make(map[string][]int32)
|
||||
for k, v := range x.user {
|
||||
pSet := make([]int32, 0, len(v))
|
||||
for p := range v {
|
||||
pSet = append(pSet, p)
|
||||
}
|
||||
res[k] = pSet
|
||||
}
|
||||
return res, 0, nil
|
||||
}
|
@ -0,0 +1,79 @@
|
||||
package mcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||
)
|
||||
|
||||
func NewSeqConversationCache(sc database.SeqConversation) cache.SeqConversationCache {
|
||||
return &seqConversationCache{
|
||||
sc: sc,
|
||||
}
|
||||
}
|
||||
|
||||
type seqConversationCache struct {
|
||||
sc database.SeqConversation
|
||||
}
|
||||
|
||||
func (x *seqConversationCache) Malloc(ctx context.Context, conversationID string, size int64) (int64, error) {
|
||||
return x.sc.Malloc(ctx, conversationID, size)
|
||||
}
|
||||
|
||||
func (x *seqConversationCache) SetMinSeq(ctx context.Context, conversationID string, seq int64) error {
|
||||
return x.sc.SetMinSeq(ctx, conversationID, seq)
|
||||
}
|
||||
|
||||
func (x *seqConversationCache) GetMinSeq(ctx context.Context, conversationID string) (int64, error) {
|
||||
return x.sc.GetMinSeq(ctx, conversationID)
|
||||
}
|
||||
|
||||
func (x *seqConversationCache) GetMaxSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error) {
|
||||
res := make(map[string]int64)
|
||||
for _, conversationID := range conversationIDs {
|
||||
seq, err := x.GetMinSeq(ctx, conversationID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
res[conversationID] = seq
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (x *seqConversationCache) GetMaxSeqsWithTime(ctx context.Context, conversationIDs []string) (map[string]database.SeqTime, error) {
|
||||
res := make(map[string]database.SeqTime)
|
||||
for _, conversationID := range conversationIDs {
|
||||
seq, err := x.GetMinSeq(ctx, conversationID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
res[conversationID] = database.SeqTime{Seq: seq}
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (x *seqConversationCache) GetMaxSeq(ctx context.Context, conversationID string) (int64, error) {
|
||||
return x.sc.GetMaxSeq(ctx, conversationID)
|
||||
}
|
||||
|
||||
func (x *seqConversationCache) GetMaxSeqWithTime(ctx context.Context, conversationID string) (database.SeqTime, error) {
|
||||
seq, err := x.GetMinSeq(ctx, conversationID)
|
||||
if err != nil {
|
||||
return database.SeqTime{}, err
|
||||
}
|
||||
return database.SeqTime{Seq: seq}, nil
|
||||
}
|
||||
|
||||
func (x *seqConversationCache) SetMinSeqs(ctx context.Context, seqs map[string]int64) error {
|
||||
for conversationID, seq := range seqs {
|
||||
if err := x.sc.SetMinSeq(ctx, conversationID, seq); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *seqConversationCache) GetCacheMaxSeqWithTime(ctx context.Context, conversationIDs []string) (map[string]database.SeqTime, error) {
|
||||
return x.GetMaxSeqsWithTime(ctx, conversationIDs)
|
||||
}
|
@ -0,0 +1,98 @@
|
||||
package mcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||
"github.com/openimsdk/tools/errs"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
func NewThirdCache(cache database.Cache) cache.ThirdCache {
|
||||
return &thirdCache{
|
||||
cache: cache,
|
||||
}
|
||||
}
|
||||
|
||||
type thirdCache struct {
|
||||
cache database.Cache
|
||||
}
|
||||
|
||||
func (c *thirdCache) getGetuiTokenKey() string {
|
||||
return cachekey.GetGetuiTokenKey()
|
||||
}
|
||||
|
||||
func (c *thirdCache) getGetuiTaskIDKey() string {
|
||||
return cachekey.GetGetuiTaskIDKey()
|
||||
}
|
||||
|
||||
func (c *thirdCache) getUserBadgeUnreadCountSumKey(userID string) string {
|
||||
return cachekey.GetUserBadgeUnreadCountSumKey(userID)
|
||||
}
|
||||
|
||||
func (c *thirdCache) getFcmAccountTokenKey(account string, platformID int) string {
|
||||
return cachekey.GetFcmAccountTokenKey(account, platformID)
|
||||
}
|
||||
|
||||
func (c *thirdCache) get(ctx context.Context, key string) (string, error) {
|
||||
res, err := c.cache.Get(ctx, []string{key})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if val, ok := res[key]; ok {
|
||||
return val, nil
|
||||
}
|
||||
return "", errs.Wrap(redis.Nil)
|
||||
}
|
||||
|
||||
func (c *thirdCache) SetFcmToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) (err error) {
|
||||
return errs.Wrap(c.cache.Set(ctx, c.getFcmAccountTokenKey(account, platformID), fcmToken, time.Duration(expireTime)*time.Second))
|
||||
}
|
||||
|
||||
func (c *thirdCache) GetFcmToken(ctx context.Context, account string, platformID int) (string, error) {
|
||||
return c.get(ctx, c.getFcmAccountTokenKey(account, platformID))
|
||||
}
|
||||
|
||||
func (c *thirdCache) DelFcmToken(ctx context.Context, account string, platformID int) error {
|
||||
return c.cache.Del(ctx, []string{c.getFcmAccountTokenKey(account, platformID)})
|
||||
}
|
||||
|
||||
func (c *thirdCache) IncrUserBadgeUnreadCountSum(ctx context.Context, userID string) (int, error) {
|
||||
return c.cache.Incr(ctx, c.getUserBadgeUnreadCountSumKey(userID), 1)
|
||||
}
|
||||
|
||||
func (c *thirdCache) SetUserBadgeUnreadCountSum(ctx context.Context, userID string, value int) error {
|
||||
return c.cache.Set(ctx, c.getUserBadgeUnreadCountSumKey(userID), strconv.Itoa(value), 0)
|
||||
}
|
||||
|
||||
func (c *thirdCache) GetUserBadgeUnreadCountSum(ctx context.Context, userID string) (int, error) {
|
||||
str, err := c.get(ctx, c.getUserBadgeUnreadCountSumKey(userID))
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
val, err := strconv.Atoi(str)
|
||||
if err != nil {
|
||||
return 0, errs.WrapMsg(err, "strconv.Atoi", "str", str)
|
||||
}
|
||||
return val, nil
|
||||
}
|
||||
|
||||
func (c *thirdCache) SetGetuiToken(ctx context.Context, token string, expireTime int64) error {
|
||||
return c.cache.Set(ctx, c.getGetuiTokenKey(), token, time.Duration(expireTime)*time.Second)
|
||||
}
|
||||
|
||||
func (c *thirdCache) GetGetuiToken(ctx context.Context) (string, error) {
|
||||
return c.get(ctx, c.getGetuiTokenKey())
|
||||
}
|
||||
|
||||
func (c *thirdCache) SetGetuiTaskID(ctx context.Context, taskID string, expireTime int64) error {
|
||||
return c.cache.Set(ctx, c.getGetuiTaskIDKey(), taskID, time.Duration(expireTime)*time.Second)
|
||||
}
|
||||
|
||||
func (c *thirdCache) GetGetuiTaskID(ctx context.Context) (string, error) {
|
||||
return c.get(ctx, c.getGetuiTaskIDKey())
|
||||
}
|
@ -0,0 +1,130 @@
|
||||
package mcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||
"github.com/openimsdk/tools/errs"
|
||||
"github.com/openimsdk/tools/log"
|
||||
)
|
||||
|
||||
func NewTokenCacheModel(cache database.Cache, accessExpire int64) cache.TokenModel {
|
||||
c := &tokenCache{cache: cache}
|
||||
c.accessExpire = c.getExpireTime(accessExpire)
|
||||
return c
|
||||
}
|
||||
|
||||
type tokenCache struct {
|
||||
cache database.Cache
|
||||
accessExpire time.Duration
|
||||
}
|
||||
|
||||
func (x *tokenCache) getTokenKey(userID string, platformID int, token string) string {
|
||||
return cachekey.GetTokenKey(userID, platformID) + ":" + token
|
||||
|
||||
}
|
||||
|
||||
func (x *tokenCache) SetTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error {
|
||||
return x.cache.Set(ctx, x.getTokenKey(userID, platformID, token), strconv.Itoa(flag), x.accessExpire)
|
||||
}
|
||||
|
||||
// SetTokenFlagEx set token and flag with expire time
|
||||
func (x *tokenCache) SetTokenFlagEx(ctx context.Context, userID string, platformID int, token string, flag int) error {
|
||||
return x.SetTokenFlag(ctx, userID, platformID, token, flag)
|
||||
}
|
||||
|
||||
func (x *tokenCache) GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error) {
|
||||
prefix := x.getTokenKey(userID, platformID, "")
|
||||
m, err := x.cache.Prefix(ctx, prefix)
|
||||
if err != nil {
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
mm := make(map[string]int)
|
||||
for k, v := range m {
|
||||
state, err := strconv.Atoi(v)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "token value is not int", err, "value", v, "userID", userID, "platformID", platformID)
|
||||
continue
|
||||
}
|
||||
mm[strings.TrimPrefix(k, prefix)] = state
|
||||
}
|
||||
return mm, nil
|
||||
}
|
||||
|
||||
func (x *tokenCache) GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error) {
|
||||
prefix := cachekey.UidPidToken + userID + ":"
|
||||
tokens, err := x.cache.Prefix(ctx, prefix)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
res := make(map[int]map[string]int)
|
||||
for key, flagStr := range tokens {
|
||||
flag, err := strconv.Atoi(flagStr)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "token value is not int", err, "key", key, "value", flagStr, "userID", userID)
|
||||
continue
|
||||
}
|
||||
arr := strings.SplitN(strings.TrimPrefix(key, prefix), ":", 2)
|
||||
if len(arr) != 2 {
|
||||
log.ZError(ctx, "token value is not int", err, "key", key, "value", flagStr, "userID", userID)
|
||||
continue
|
||||
}
|
||||
platformID, err := strconv.Atoi(arr[0])
|
||||
if err != nil {
|
||||
log.ZError(ctx, "token value is not int", err, "key", key, "value", flagStr, "userID", userID)
|
||||
continue
|
||||
}
|
||||
token := arr[1]
|
||||
if token == "" {
|
||||
log.ZError(ctx, "token value is not int", err, "key", key, "value", flagStr, "userID", userID)
|
||||
continue
|
||||
}
|
||||
tk, ok := res[platformID]
|
||||
if !ok {
|
||||
tk = make(map[string]int)
|
||||
res[platformID] = tk
|
||||
}
|
||||
tk[token] = flag
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (x *tokenCache) SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error {
|
||||
for token, flag := range m {
|
||||
err := x.SetTokenFlag(ctx, userID, platformID, token, flag)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *tokenCache) BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]any) error {
|
||||
for prefix, tokenFlag := range tokens {
|
||||
for token, flag := range tokenFlag {
|
||||
flagStr := fmt.Sprintf("%v", flag)
|
||||
if err := x.cache.Set(ctx, prefix+":"+token, flagStr, x.accessExpire); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *tokenCache) DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error {
|
||||
keys := make([]string, 0, len(fields))
|
||||
for _, token := range fields {
|
||||
keys = append(keys, x.getTokenKey(userID, platformID, token))
|
||||
}
|
||||
return x.cache.Del(ctx, keys)
|
||||
}
|
||||
|
||||
func (x *tokenCache) getExpireTime(t int64) time.Duration {
|
||||
return time.Hour * 24 * time.Duration(t)
|
||||
}
|
@ -0,0 +1,63 @@
|
||||
package mcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||
"github.com/openimsdk/tools/log"
|
||||
)
|
||||
|
||||
func getCache[V any](ctx context.Context, cache database.Cache, key string, expireTime time.Duration, fn func(ctx context.Context) (V, error)) (V, error) {
|
||||
getDB := func() (V, bool, error) {
|
||||
res, err := cache.Get(ctx, []string{key})
|
||||
if err != nil {
|
||||
var val V
|
||||
return val, false, err
|
||||
}
|
||||
var val V
|
||||
if str, ok := res[key]; ok {
|
||||
if json.Unmarshal([]byte(str), &val) != nil {
|
||||
return val, false, err
|
||||
}
|
||||
return val, true, nil
|
||||
}
|
||||
return val, false, nil
|
||||
}
|
||||
dbVal, ok, err := getDB()
|
||||
if err != nil {
|
||||
return dbVal, err
|
||||
}
|
||||
if ok {
|
||||
return dbVal, nil
|
||||
}
|
||||
lockValue, err := cache.Lock(ctx, key, time.Minute)
|
||||
if err != nil {
|
||||
return dbVal, err
|
||||
}
|
||||
defer func() {
|
||||
if err := cache.Unlock(ctx, key, lockValue); err != nil {
|
||||
log.ZError(ctx, "unlock cache key", err, "key", key, "value", lockValue)
|
||||
}
|
||||
}()
|
||||
dbVal, ok, err = getDB()
|
||||
if err != nil {
|
||||
return dbVal, err
|
||||
}
|
||||
if ok {
|
||||
return dbVal, nil
|
||||
}
|
||||
val, err := fn(ctx)
|
||||
if err != nil {
|
||||
return val, err
|
||||
}
|
||||
data, err := json.Marshal(val)
|
||||
if err != nil {
|
||||
return val, err
|
||||
}
|
||||
if err := cache.Set(ctx, key, string(data), expireTime); err != nil {
|
||||
return val, err
|
||||
}
|
||||
return val, nil
|
||||
}
|
@ -1,15 +0,0 @@
|
||||
// Copyright © 2023 OpenIM. All rights reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package redis
|
@ -0,0 +1,59 @@
|
||||
package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
|
||||
"github.com/openimsdk/tools/s3/minio"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
func NewMinioCache(rdb redis.UniversalClient) minio.Cache {
|
||||
rc := newRocksCacheClient(rdb)
|
||||
return &minioCacheRedis{
|
||||
BatchDeleter: rc.GetBatchDeleter(),
|
||||
rcClient: rc,
|
||||
expireTime: time.Hour * 24 * 7,
|
||||
}
|
||||
}
|
||||
|
||||
type minioCacheRedis struct {
|
||||
cache.BatchDeleter
|
||||
rcClient *rocksCacheClient
|
||||
expireTime time.Duration
|
||||
}
|
||||
|
||||
func (g *minioCacheRedis) getObjectImageInfoKey(key string) string {
|
||||
return cachekey.GetObjectImageInfoKey(key)
|
||||
}
|
||||
|
||||
func (g *minioCacheRedis) getMinioImageThumbnailKey(key string, format string, width int, height int) string {
|
||||
return cachekey.GetMinioImageThumbnailKey(key, format, width, height)
|
||||
}
|
||||
|
||||
func (g *minioCacheRedis) DelObjectImageInfoKey(ctx context.Context, keys ...string) error {
|
||||
ks := make([]string, 0, len(keys))
|
||||
for _, key := range keys {
|
||||
ks = append(ks, g.getObjectImageInfoKey(key))
|
||||
}
|
||||
return g.BatchDeleter.ExecDelWithKeys(ctx, ks)
|
||||
}
|
||||
|
||||
func (g *minioCacheRedis) DelImageThumbnailKey(ctx context.Context, key string, format string, width int, height int) error {
|
||||
return g.BatchDeleter.ExecDelWithKeys(ctx, []string{g.getMinioImageThumbnailKey(key, format, width, height)})
|
||||
|
||||
}
|
||||
|
||||
func (g *minioCacheRedis) GetImageObjectKeyInfo(ctx context.Context, key string, fn func(ctx context.Context) (*minio.ImageInfo, error)) (*minio.ImageInfo, error) {
|
||||
info, err := getCache(ctx, g.rcClient, g.getObjectImageInfoKey(key), g.expireTime, fn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return info, nil
|
||||
}
|
||||
|
||||
func (g *minioCacheRedis) GetThumbnailKey(ctx context.Context, key string, format string, width int, height int, minioCache func(ctx context.Context) (string, error)) (string, error) {
|
||||
return getCache(ctx, g.rcClient, g.getMinioImageThumbnailKey(key, format, width, height), g.expireTime, minioCache)
|
||||
}
|
@ -0,0 +1,16 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Cache interface {
|
||||
Get(ctx context.Context, key []string) (map[string]string, error)
|
||||
Prefix(ctx context.Context, prefix string) (map[string]string, error)
|
||||
Set(ctx context.Context, key string, value string, expireAt time.Duration) error
|
||||
Incr(ctx context.Context, key string, value int) (int, error)
|
||||
Del(ctx context.Context, key []string) error
|
||||
Lock(ctx context.Context, key string, duration time.Duration) (string, error)
|
||||
Unlock(ctx context.Context, key string, value string) error
|
||||
}
|
@ -0,0 +1,183 @@
|
||||
package mgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/tools/db/mongoutil"
|
||||
"github.com/openimsdk/tools/errs"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
)
|
||||
|
||||
func NewCacheMgo(db *mongo.Database) (*CacheMgo, error) {
|
||||
coll := db.Collection(database.CacheName)
|
||||
_, err := coll.Indexes().CreateMany(context.Background(), []mongo.IndexModel{
|
||||
{
|
||||
Keys: bson.D{
|
||||
{Key: "key", Value: 1},
|
||||
},
|
||||
Options: options.Index().SetUnique(true),
|
||||
},
|
||||
{
|
||||
Keys: bson.D{
|
||||
{Key: "expire_at", Value: 1},
|
||||
},
|
||||
Options: options.Index().SetExpireAfterSeconds(0),
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
return &CacheMgo{coll: coll}, nil
|
||||
}
|
||||
|
||||
type CacheMgo struct {
|
||||
coll *mongo.Collection
|
||||
}
|
||||
|
||||
func (x *CacheMgo) findToMap(res []model.Cache, now time.Time) map[string]string {
|
||||
kv := make(map[string]string)
|
||||
for _, re := range res {
|
||||
if re.ExpireAt != nil && re.ExpireAt.Before(now) {
|
||||
continue
|
||||
}
|
||||
kv[re.Key] = re.Value
|
||||
}
|
||||
return kv
|
||||
|
||||
}
|
||||
|
||||
func (x *CacheMgo) Get(ctx context.Context, key []string) (map[string]string, error) {
|
||||
if len(key) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
now := time.Now()
|
||||
res, err := mongoutil.Find[model.Cache](ctx, x.coll, bson.M{
|
||||
"key": bson.M{"$in": key},
|
||||
"$or": []bson.M{
|
||||
{"expire_at": bson.M{"$gt": now}},
|
||||
{"expire_at": nil},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return x.findToMap(res, now), nil
|
||||
}
|
||||
|
||||
func (x *CacheMgo) Prefix(ctx context.Context, prefix string) (map[string]string, error) {
|
||||
now := time.Now()
|
||||
res, err := mongoutil.Find[model.Cache](ctx, x.coll, bson.M{
|
||||
"key": bson.M{"$regex": "^" + prefix},
|
||||
"$or": []bson.M{
|
||||
{"expire_at": bson.M{"$gt": now}},
|
||||
{"expire_at": nil},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return x.findToMap(res, now), nil
|
||||
}
|
||||
|
||||
func (x *CacheMgo) Set(ctx context.Context, key string, value string, expireAt time.Duration) error {
|
||||
cv := &model.Cache{
|
||||
Key: key,
|
||||
Value: value,
|
||||
}
|
||||
if expireAt > 0 {
|
||||
now := time.Now().Add(expireAt)
|
||||
cv.ExpireAt = &now
|
||||
}
|
||||
opt := options.Update().SetUpsert(true)
|
||||
return mongoutil.UpdateOne(ctx, x.coll, bson.M{"key": key}, bson.M{"$set": cv}, false, opt)
|
||||
}
|
||||
|
||||
func (x *CacheMgo) Incr(ctx context.Context, key string, value int) (int, error) {
|
||||
pipeline := mongo.Pipeline{
|
||||
{
|
||||
{"$set", bson.M{
|
||||
"value": bson.M{
|
||||
"$toString": bson.M{
|
||||
"$add": bson.A{
|
||||
bson.M{"$toInt": "$value"},
|
||||
value,
|
||||
},
|
||||
},
|
||||
},
|
||||
}},
|
||||
},
|
||||
}
|
||||
opt := options.FindOneAndUpdate().SetReturnDocument(options.After)
|
||||
res, err := mongoutil.FindOneAndUpdate[model.Cache](ctx, x.coll, bson.M{"key": key}, pipeline, opt)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return strconv.Atoi(res.Value)
|
||||
}
|
||||
|
||||
func (x *CacheMgo) Del(ctx context.Context, key []string) error {
|
||||
if len(key) == 0 {
|
||||
return nil
|
||||
}
|
||||
_, err := x.coll.DeleteMany(ctx, bson.M{"key": bson.M{"$in": key}})
|
||||
return err
|
||||
}
|
||||
|
||||
func (x *CacheMgo) lockKey(key string) string {
|
||||
return "LOCK_" + key
|
||||
}
|
||||
|
||||
func (x *CacheMgo) Lock(ctx context.Context, key string, duration time.Duration) (string, error) {
|
||||
tmp, err := uuid.NewUUID()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if duration <= 0 || duration > time.Minute*10 {
|
||||
duration = time.Minute * 10
|
||||
}
|
||||
cv := &model.Cache{
|
||||
Key: x.lockKey(key),
|
||||
Value: tmp.String(),
|
||||
ExpireAt: nil,
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
|
||||
defer cancel()
|
||||
wait := func() error {
|
||||
timeout := time.NewTimer(time.Millisecond * 100)
|
||||
defer timeout.Stop()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-timeout.C:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
for {
|
||||
if err := mongoutil.DeleteOne(ctx, x.coll, bson.M{"key": key, "expire_at": bson.M{"$lt": time.Now()}}); err != nil {
|
||||
return "", err
|
||||
}
|
||||
expireAt := time.Now().Add(duration)
|
||||
cv.ExpireAt = &expireAt
|
||||
if err := mongoutil.InsertMany[*model.Cache](ctx, x.coll, []*model.Cache{cv}); err != nil {
|
||||
if mongo.IsDuplicateKeyError(err) {
|
||||
if err := wait(); err != nil {
|
||||
return "", err
|
||||
}
|
||||
continue
|
||||
}
|
||||
return "", err
|
||||
}
|
||||
return cv.Value, nil
|
||||
}
|
||||
}
|
||||
|
||||
func (x *CacheMgo) Unlock(ctx context.Context, key string, value string) error {
|
||||
return mongoutil.DeleteOne(ctx, x.coll, bson.M{"key": x.lockKey(key), "value": value})
|
||||
}
|
@ -0,0 +1,133 @@
|
||||
package mgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/tools/db/mongoutil"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
)
|
||||
|
||||
func TestName1111(t *testing.T) {
|
||||
coll := Mongodb().Collection("temp")
|
||||
|
||||
//updatePipeline := mongo.Pipeline{
|
||||
// {
|
||||
// {"$set", bson.M{
|
||||
// "age": bson.M{
|
||||
// "$toString": bson.M{
|
||||
// "$add": bson.A{
|
||||
// bson.M{"$toInt": "$age"},
|
||||
// 1,
|
||||
// },
|
||||
// },
|
||||
// },
|
||||
// }},
|
||||
// },
|
||||
//}
|
||||
|
||||
pipeline := mongo.Pipeline{
|
||||
{
|
||||
{"$set", bson.M{
|
||||
"value": bson.M{
|
||||
"$toString": bson.M{
|
||||
"$add": bson.A{
|
||||
bson.M{"$toInt": "$value"},
|
||||
1,
|
||||
},
|
||||
},
|
||||
},
|
||||
}},
|
||||
},
|
||||
}
|
||||
|
||||
opt := options.FindOneAndUpdate().SetUpsert(false).SetReturnDocument(options.After)
|
||||
res, err := mongoutil.FindOneAndUpdate[model.Cache](context.Background(), coll, bson.M{"key": "123456"}, pipeline, opt)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
t.Log(res)
|
||||
}
|
||||
|
||||
func TestName33333(t *testing.T) {
|
||||
c, err := NewCacheMgo(Mongodb())
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
if err := c.Set(context.Background(), "123456", "123456", time.Hour); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
if err := c.Set(context.Background(), "123666", "123666", time.Hour); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
res1, err := c.Get(context.Background(), []string{"123456"})
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
t.Log(res1)
|
||||
|
||||
res2, err := c.Prefix(context.Background(), "123")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
t.Log(res2)
|
||||
}
|
||||
|
||||
func TestName1111aa(t *testing.T) {
|
||||
|
||||
c, err := NewCacheMgo(Mongodb())
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
var count int
|
||||
|
||||
key := "123456"
|
||||
|
||||
doFunc := func() {
|
||||
value, err := c.Lock(context.Background(), key, time.Second*30)
|
||||
if err != nil {
|
||||
t.Log("Lock error", err)
|
||||
return
|
||||
}
|
||||
tmp := count
|
||||
tmp++
|
||||
count = tmp
|
||||
t.Log("count", tmp)
|
||||
if err := c.Unlock(context.Background(), key, value); err != nil {
|
||||
t.Log("Unlock error", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if _, err := c.Lock(context.Background(), key, time.Second*10); err != nil {
|
||||
t.Log(err)
|
||||
return
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 32; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for i := 0; i < 100; i++ {
|
||||
doFunc()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
|
||||
}
|
||||
|
||||
func TestName111111a(t *testing.T) {
|
||||
arr := strings.SplitN("1:testkakskdask:1111", ":", 2)
|
||||
t.Log(arr)
|
||||
}
|
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue