支持自定义日志，优化数据返回

cluster/cluster.go

@@ -93,6 +93,7 @@ func (slf *Cluster) Start() {
 	slf.rpcServer.Start(slf.localNodeInfo.ListenAddr)
 }
 
+
 func GetCluster() *Cluster{
 	return &cluster
 }

example/main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"fmt"
 	"github.com/duanhf2012/origin/example/GateService"
+	"github.com/duanhf2012/origin/log"
 	"github.com/duanhf2012/origin/node"
 	"github.com/duanhf2012/origin/service"
 	"github.com/duanhf2012/origin/sysmodule"
@@ -91,6 +92,7 @@ func (slf *Module4) OnRelease() {
 
 func (slf *TestServiceCall) OnInit() error {
 	slf.AfterFunc(time.Second*1,slf.Run)
+	slf.AfterFunc(time.Second*1,slf.Test)
 	moduleid1,_ = slf.AddModule(&Module1{})
 	moduleid2,_ = slf.AddModule(&Module2{})
 	fmt.Print(moduleid1,moduleid2)
@@ -118,38 +120,47 @@ type Param struct {
 	Pa []string
 }
 
-
+var index int
-func  (slf *TestServiceCall) Run(){
+func (slf *TestServiceCall) Test(){
-	//var ret int
+	index += 1
-	//var input int = 10000
-	//bT := time.Now()            // 开始时间
-
-	//err := slf.Call("TestServiceCall.RPC_Test",&ret,&input)
 	var param Param
 	param.A = 2342342341
 	param.B = "xxxxxxxxxxxxxxxxxxxxxxx"
 	param.Pa = []string{"ccccc","asfsdfsdaf","bbadfsdf","ewrwefasdf","safsadfka;fksd"}
-	/*
+	param.Index = index
+	slf.AsyncCall("TestService1.RPC_Test",&param, func(reply *Param, err error) {
+		fmt.Print(reply,"\n")
+	})
+	slf.AfterFunc(time.Second*1,slf.Test)
+}
+
+func  (slf *TestServiceCall) Run(){
+	//var ret int
+	var input int = 1000000
+	//bT := time.Now()            // 开始时间
+
+	//err := slf.Call("TestServiceCall.RPC_Test",&ret,&input)
 	for i:=input;i>=0;i--{
+		var param Param
+		param.A = 2342342341
+		param.B = "xxxxxxxxxxxxxxxxxxxxxxx"
+		param.Pa = []string{"ccccc","asfsdfsdaf","bbadfsdf","ewrwefasdf","safsadfka;fksd"}
 		param.Index = i
+		if param.Index == 0 {
+			fmt.Print(".......................\n")
+		}
 		slf.AsyncCall("TestService1.RPC_Test",&param, func(reply *Param, err error) {
-			if reply.Index == 0 || err != nil{
+			log.Debug(" index %d ,err %+v",reply.Index,err)
-				eT := time.Since(bT)      // 从开始到当前所消耗的时间
-				fmt.Print(err,eT.Milliseconds())
-				fmt.Print("xxxx..................",eT,err,"\n")
-			}
-			//fmt.Print(*reply,"\n",err)
 		})
 
 	}
-*/
+
 	fmt.Print("finsh....")
-
-
 }
 
 func (slf *TestService1) RPC_Test(a *Param,b *Param) error {
-	*a = *b
+	//*a = *b
+	*b = *a
 	return nil
 }
 

node/node.go

@@ -4,8 +4,10 @@ import (
 	"fmt"
 	"github.com/duanhf2012/origin/cluster"
 	"github.com/duanhf2012/origin/console"
+	"github.com/duanhf2012/origin/log"
 	"github.com/duanhf2012/origin/service"
 	"io/ioutil"
+	syslog "log"
 	"os"
 	"os/signal"
 	"strconv"
@@ -84,6 +86,7 @@ func initNode(id int){
 }
 
 func Start() {
+	SetSysLog("debug","./",syslog.Lshortfile|syslog.LstdFlags)
 	console.RegisterCommand("start",startNode)
 	console.RegisterCommand("stop",stopNode)
 	err := console.Run(os.Args)
@@ -132,3 +135,8 @@ func GetService(servicename string) service.IService {
 func SetConfigDir(configdir string){
 	cluster.SetConfigDir(configdir)
 }
+
+func SetSysLog(strLevel string, pathname string, flag int){
+	logs,_:= log.New(strLevel,pathname,flag)
+	log.Export(logs)
+}

rpc/client.go

@@ -70,7 +70,6 @@ func (slf *Client) AsycGo(rpcHandler IRpcHandler,mutiCoroutine bool,serviceMetho
 	request.Seq = slf.startSeq
 	slf.pending[call.Seq] = call
 	slf.pendingLock.Unlock()
-
 	request.ServiceMethod = serviceMethod
 	var herr error
 	request.InParam,herr = processor.Marshal(args)
@@ -163,8 +162,8 @@ func (slf *Client) Run(){
 	for {
 		bytes,err := slf.conn.ReadMsg()
 		if err != nil {
-			slf.Close()
+			log.Error("rpcClient %s ReadMsg error:%+v",slf.Addr,err)
-			slf.Start()
+			return
 		}
 		//1.解析head
 		respone := &RpcResponse{}
@@ -198,12 +197,7 @@ func (slf *Client) Run(){
 
 		}
 	}
-
 }
 
 func (slf *Client) OnClose(){
-	if slf.blocalhost== false{
-		//关闭时，重新连接
-		slf.Start()
-	}
 }

rpc/rpchandler.go

@@ -179,13 +179,16 @@ func (slf *RpcHandler) HandlerRpcRequest(request *RpcRequest) {
 		v.iparam = request.localParam
 	}
 
-
+	var oParam reflect.Value
 	paramList = append(paramList,reflect.ValueOf(slf.GetRpcHandler())) //接受者
 	if request.localReply!=nil {
-		v.oParam = reflect.ValueOf(request.localReply)
+		oParam = reflect.ValueOf(request.localReply)
+	}else{
+		oParam = reflect.New(v.oParam.Type().Elem())
 	}
+
 	paramList = append(paramList,reflect.ValueOf(v.iparam))
-	paramList = append(paramList,v.oParam) //输出参数
+	paramList = append(paramList,oParam) //输出参数
 
 	returnValues := v.method.Func.Call(paramList)
 	errInter := returnValues[0].Interface()
@@ -194,7 +197,7 @@ func (slf *RpcHandler) HandlerRpcRequest(request *RpcRequest) {
 	}
 
 	if request.requestHandle!=nil {
-		request.requestHandle(v.oParam.Interface(), err)
+		request.requestHandle(oParam.Interface(), err)
 	}
 }