withchao
8 months ago
parent
f60272a1d2
commit
c2b035839d
@ -0,0 +1,209 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"flag"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"path"
|
||||||
|
"path/filepath"
|
||||||
|
"reflect"
|
||||||
|
"runtime"
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
"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/pkg/common/config"
|
||||||
|
"github.com/openimsdk/tools/discovery"
|
||||||
|
"github.com/openimsdk/tools/discovery/standalone"
|
||||||
|
"github.com/openimsdk/tools/utils/datautil"
|
||||||
|
"github.com/spf13/viper"
|
||||||
|
"google.golang.org/grpc"
|
||||||
|
"gopkg.in/yaml.v3"
|
||||||
|
)
|
||||||
|
|
||||||
|
func main() {
|
||||||
|
var configPath string
|
||||||
|
flag.StringVar(&configPath, "c", "/Users/chao/Desktop/code/open-im-server/config", "config path")
|
||||||
|
flag.Parse()
|
||||||
|
cmd := newCmds(configPath)
|
||||||
|
putCmd1(cmd, auth.Start)
|
||||||
|
putCmd1(cmd, conversation.Start)
|
||||||
|
putCmd1(cmd, relation.Start)
|
||||||
|
putCmd1(cmd, group.Start)
|
||||||
|
putCmd1(cmd, msg.Start)
|
||||||
|
putCmd1(cmd, third.Start)
|
||||||
|
putCmd1(cmd, user.Start)
|
||||||
|
putCmd1(cmd, push.Start)
|
||||||
|
putCmd2(cmd, msggateway.Start)
|
||||||
|
putCmd2(cmd, msgtransfer.Start)
|
||||||
|
putCmd2(cmd, api.Start)
|
||||||
|
ctx := context.Background()
|
||||||
|
if err := cmd.run(ctx); err != nil {
|
||||||
|
fmt.Println(err)
|
||||||
|
}
|
||||||
|
fmt.Println("success")
|
||||||
|
}
|
||||||
|
|
||||||
|
func getTypePath(typ reflect.Type) string {
|
||||||
|
return path.Join(typ.PkgPath(), typ.Name())
|
||||||
|
}
|
||||||
|
|
||||||
|
func newCmds(confPath string) *cmds {
|
||||||
|
return &cmds{confPath: confPath}
|
||||||
|
}
|
||||||
|
|
||||||
|
type cmds struct {
|
||||||
|
confPath string
|
||||||
|
cmds []cmdName
|
||||||
|
conf map[string][]byte
|
||||||
|
}
|
||||||
|
|
||||||
|
func (x *cmds) readConfig() error {
|
||||||
|
skip := []string{
|
||||||
|
config.DiscoveryConfigFilename,
|
||||||
|
}
|
||||||
|
if x.conf == nil {
|
||||||
|
x.conf = make(map[string][]byte)
|
||||||
|
}
|
||||||
|
vof := reflect.ValueOf(&config.AllConfig{}).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
|
||||||
|
}
|
||||||
|
}
|
||||||
|
itemConf := field.Addr().Interface()
|
||||||
|
name := itemConf.(interface{ GetConfigFileName() string }).GetConfigFileName()
|
||||||
|
if datautil.Contain(name, skip...) {
|
||||||
|
x.conf[getTypePath(field.Type())] = nil
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
data, err := os.ReadFile(filepath.Join(x.confPath, name))
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
x.conf[getTypePath(field.Type())] = data
|
||||||
|
}
|
||||||
|
val := config.Discovery{Enable: discovery.Standalone}
|
||||||
|
var buf bytes.Buffer
|
||||||
|
if err := yaml.NewEncoder(&buf).Encode(&val); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
x.conf[getTypePath(reflect.TypeOf(val))] = buf.Bytes()
|
||||||
|
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 := getTypePath(field.Type())
|
||||||
|
confData, ok := x.conf[pkt]
|
||||||
|
if !ok {
|
||||||
|
if typeField.Name == "FcmConfigPath" && field.Kind() == reflect.String {
|
||||||
|
field.SetString(x.confPath)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
return fmt.Errorf("config field %s %s not found", vof.Type().Name(), typeField.Name)
|
||||||
|
}
|
||||||
|
if confData == nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
val := field.Addr().Interface()
|
||||||
|
v := viper.New()
|
||||||
|
v.SetConfigType("yaml")
|
||||||
|
if err := v.ReadConfig(bytes.NewReader(confData)); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
fn := func(conf *mapstructure.DecoderConfig) {
|
||||||
|
conf.TagName = config.StructTagName
|
||||||
|
}
|
||||||
|
if err := v.Unmarshal(val, fn); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (x *cmds) add(name string, fn func(ctx context.Context) error) {
|
||||||
|
x.cmds = append(x.cmds, cmdName{Name: name, Func: fn})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (x *cmds) run(ctx context.Context) error {
|
||||||
|
if x.conf == nil {
|
||||||
|
if err := x.readConfig(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for _, cmd := range x.cmds {
|
||||||
|
fmt.Println("start", cmd.Name)
|
||||||
|
if err := cmd.Func(ctx); err != nil {
|
||||||
|
fmt.Println("start failed", cmd.Name, err)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
fmt.Println("start ok", cmd.Name)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
type cmdName struct {
|
||||||
|
Name string
|
||||||
|
Func func(ctx context.Context) error
|
||||||
|
}
|
||||||
|
|
||||||
|
func getFuncPacketName(fn any) string {
|
||||||
|
name := path.Base(runtime.FuncForPC(reflect.ValueOf(fn).Pointer()).Name())
|
||||||
|
if index := strings.Index(name, "."); index >= 0 {
|
||||||
|
name = name[:index]
|
||||||
|
}
|
||||||
|
return name
|
||||||
|
}
|
||||||
|
|
||||||
|
func putCmd1[C any](cmd *cmds, fn func(ctx context.Context, config *C, client discovery.Conn, server grpc.ServiceRegistrar) error) {
|
||||||
|
cmd.add(getFuncPacketName(fn), 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())
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func putCmd2[C any](cmd *cmds, fn func(ctx context.Context, index int, config *C) error) {
|
||||||
|
cmd.add(getFuncPacketName(fn), func(ctx context.Context) error {
|
||||||
|
var conf C
|
||||||
|
if err := cmd.parseConf(&conf); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return fn(ctx, 0, &conf)
|
||||||
|
})
|
||||||
|
}
|
||||||
@ -1,134 +0,0 @@
|
|||||||
package service
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"fmt"
|
|
||||||
"sync"
|
|
||||||
"testing"
|
|
||||||
|
|
||||||
"github.com/openimsdk/protocol/group"
|
|
||||||
"github.com/openimsdk/protocol/user"
|
|
||||||
"google.golang.org/grpc"
|
|
||||||
"google.golang.org/grpc/codes"
|
|
||||||
"google.golang.org/grpc/status"
|
|
||||||
"google.golang.org/protobuf/proto"
|
|
||||||
)
|
|
||||||
|
|
||||||
var (
|
|
||||||
_ user.UnimplementedUserServer
|
|
||||||
_ group.UnimplementedGroupServer
|
|
||||||
)
|
|
||||||
|
|
||||||
func TestName1(t *testing.T) {
|
|
||||||
cc := newStandaloneConn()
|
|
||||||
user.RegisterUserServer(cc.Registry(), &user.UnimplementedUserServer{})
|
|
||||||
group.RegisterGroupServer(cc.Registry(), &group.UnimplementedGroupServer{})
|
|
||||||
ctx := context.Background()
|
|
||||||
resp, err := user.NewUserClient(cc).GetUserStatus(ctx, &user.GetUserStatusReq{UserID: "imAdmin", UserIDs: []string{"10000", "20000"}})
|
|
||||||
if err != nil {
|
|
||||||
t.Error(err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
t.Log(resp)
|
|
||||||
}
|
|
||||||
|
|
||||||
func newStandaloneConn() *standaloneConn {
|
|
||||||
return &standaloneConn{
|
|
||||||
registry: newStandaloneRegistry(),
|
|
||||||
serializer: NewProtoSerializer(),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
type standaloneConn struct {
|
|
||||||
registry *standaloneRegistry
|
|
||||||
serializer Serializer
|
|
||||||
}
|
|
||||||
|
|
||||||
func (x *standaloneConn) Registry() grpc.ServiceRegistrar {
|
|
||||||
return x.registry
|
|
||||||
}
|
|
||||||
|
|
||||||
func (x *standaloneConn) Invoke(ctx context.Context, method string, args any, reply any, opts ...grpc.CallOption) error {
|
|
||||||
handler := x.registry.getMethod(method)
|
|
||||||
if handler == nil {
|
|
||||||
return fmt.Errorf("service %s not found", method)
|
|
||||||
}
|
|
||||||
resp, err := handler(ctx, args, nil)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
tmp, err := x.serializer.Marshal(resp)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
return x.serializer.Unmarshal(tmp, reply)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (x *standaloneConn) NewStream(ctx context.Context, desc *grpc.StreamDesc, method string, opts ...grpc.CallOption) (grpc.ClientStream, error) {
|
|
||||||
return nil, status.Errorf(codes.Unimplemented, "method stream not implemented")
|
|
||||||
}
|
|
||||||
|
|
||||||
type serverHandler func(ctx context.Context, req any, interceptor grpc.UnaryServerInterceptor) (any, error)
|
|
||||||
|
|
||||||
func newStandaloneRegistry() *standaloneRegistry {
|
|
||||||
return &standaloneRegistry{
|
|
||||||
methods: make(map[string]serverHandler),
|
|
||||||
serializer: NewProtoSerializer(),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
type standaloneRegistry struct {
|
|
||||||
lock sync.RWMutex
|
|
||||||
methods map[string]serverHandler
|
|
||||||
serializer Serializer
|
|
||||||
}
|
|
||||||
|
|
||||||
func (x *standaloneConn) emptyDec(req any) error {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (x *standaloneRegistry) RegisterService(desc *grpc.ServiceDesc, impl any) {
|
|
||||||
x.lock.Lock()
|
|
||||||
defer x.lock.Unlock()
|
|
||||||
for i := range desc.Methods {
|
|
||||||
method := desc.Methods[i]
|
|
||||||
name := fmt.Sprintf("/%s/%s", desc.ServiceName, method.MethodName)
|
|
||||||
if _, ok := x.methods[name]; ok {
|
|
||||||
panic(fmt.Errorf("service %s already registered, method %s", desc.ServiceName, method.MethodName))
|
|
||||||
}
|
|
||||||
x.methods[name] = func(ctx context.Context, req any, interceptor grpc.UnaryServerInterceptor) (any, error) {
|
|
||||||
return method.Handler(impl, ctx, func(in any) error {
|
|
||||||
tmp, err := x.serializer.Marshal(req)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
return x.serializer.Unmarshal(tmp, in)
|
|
||||||
}, interceptor)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (x *standaloneRegistry) getMethod(name string) serverHandler {
|
|
||||||
x.lock.RLock()
|
|
||||||
defer x.lock.RUnlock()
|
|
||||||
return x.methods[name]
|
|
||||||
}
|
|
||||||
|
|
||||||
type Serializer interface {
|
|
||||||
Marshal(any) ([]byte, error)
|
|
||||||
Unmarshal([]byte, any) error
|
|
||||||
}
|
|
||||||
|
|
||||||
func NewProtoSerializer() Serializer {
|
|
||||||
return protoSerializer{}
|
|
||||||
}
|
|
||||||
|
|
||||||
type protoSerializer struct{}
|
|
||||||
|
|
||||||
func (protoSerializer) Marshal(in any) ([]byte, error) {
|
|
||||||
return proto.Marshal(in.(proto.Message))
|
|
||||||
}
|
|
||||||
|
|
||||||
func (protoSerializer) Unmarshal(b []byte, out any) error {
|
|
||||||
return proto.Unmarshal(b, out.(proto.Message))
|
|
||||||
}
|
|
||||||
@ -1,34 +0,0 @@
|
|||||||
package service
|
|
||||||
|
|
||||||
//
|
|
||||||
//import (
|
|
||||||
// "context"
|
|
||||||
// "fmt"
|
|
||||||
// "sync"
|
|
||||||
//
|
|
||||||
// "google.golang.org/grpc"
|
|
||||||
//)
|
|
||||||
//
|
|
||||||
//type DiscoveryRegistry struct {
|
|
||||||
// lock sync.RWMutex
|
|
||||||
// services map[string]grpc.ClientConnInterface
|
|
||||||
//}
|
|
||||||
//
|
|
||||||
//func (x *DiscoveryRegistry) RegisterService(desc *grpc.ServiceDesc, impl any) {
|
|
||||||
// fmt.Println("RegisterService", desc, impl)
|
|
||||||
//}
|
|
||||||
//
|
|
||||||
//func (x *DiscoveryRegistry) GetConns(ctx context.Context, serviceName string, opts ...grpc.DialOption) ([]grpc.ClientConnInterface, error) {
|
|
||||||
// //TODO implement me
|
|
||||||
// panic("implement me")
|
|
||||||
//}
|
|
||||||
//
|
|
||||||
//func (x *DiscoveryRegistry) GetConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (grpc.ClientConnInterface, error) {
|
|
||||||
// //TODO implement me
|
|
||||||
// panic("implement me")
|
|
||||||
//}
|
|
||||||
//
|
|
||||||
//func (x *DiscoveryRegistry) IsSelfNode(cc grpc.ClientConnInterface) bool {
|
|
||||||
//
|
|
||||||
// return false
|
|
||||||
//}
|
|
||||||
@ -1,14 +0,0 @@
|
|||||||
package service
|
|
||||||
|
|
||||||
import (
|
|
||||||
"fmt"
|
|
||||||
|
|
||||||
"google.golang.org/grpc"
|
|
||||||
)
|
|
||||||
|
|
||||||
type GrpcServer struct {
|
|
||||||
}
|
|
||||||
|
|
||||||
func (x *GrpcServer) RegisterService(desc *grpc.ServiceDesc, impl any) {
|
|
||||||
fmt.Println("RegisterService", desc, impl)
|
|
||||||
}
|
|
||||||
Loading…
Reference in new issue