Merge branch 'master' of https://github.com/duanhf2012/origin

cluster/cluster.go

@@ -145,3 +145,8 @@ func GetRpcClient(nodeId int,serviceMethod string) ([]*rpc.Client,error) {
 func GetRpcServer() *rpc.Server{
 	return &cluster.rpcServer
 }
+
+func (slf *Cluster) IsNodeConnected (nodeId int) bool {
+	pClient := slf.GetRpcClient(nodeId)
+	return pClient!=nil && pClient.IsConnected()
+}

event/event.go

@@ -10,7 +10,7 @@ import (
 const Default_EventChannelLen = 10000
 
 //事件接受器
-type EventReciver func(event *Event) error
+type EventReciverFunc func(event *Event)
 
 type Event struct {
 	Type EventType
@@ -19,10 +19,12 @@ type Event struct {
 
 type IEventProcessor interface {
 	NotifyEvent(*Event)
-	OnEventHandler(event *Event) error
+
 	SetEventReciver(eventProcessor IEventProcessor)
 	GetEventReciver() IEventProcessor
 	SetEventChanNum(num int32) bool
+	RegEventReciverFunc(eventType EventType,reciverFunc EventReciverFunc)
+	UnRegEventReciverFun(eventType EventType)
 }
 
 type EventProcessor struct {
@@ -32,6 +34,18 @@ type EventProcessor struct {
 
 	eventChanNumLocker sync.RWMutex
 	eventChanNum int32
+	mapEventReciverFunc map[EventType]EventReciverFunc
+}
+
+func (slf *EventProcessor) RegEventReciverFunc(eventType EventType,reciverFunc EventReciverFunc){
+	if slf.mapEventReciverFunc == nil {
+		slf.mapEventReciverFunc = map[EventType]EventReciverFunc{}
+	}
+	slf.mapEventReciverFunc[eventType] = reciverFunc
+}
+
+func (slf *EventProcessor) UnRegEventReciverFun(eventType EventType){
+	delete(slf.mapEventReciverFunc,eventType)
 }
 
 func (slf *EventProcessor) NotifyEvent(pEvent *Event) {
@@ -41,9 +55,7 @@ func (slf *EventProcessor) NotifyEvent(pEvent *Event) {
 	slf.EventChan <-pEvent
 }
 
-func (slf *EventProcessor) OnEventHandler(event *Event) error{
-	return nil
-}
+
 
 func (slf *EventProcessor) GetEventChan() chan *Event{
 	slf.eventChanNumLocker.Lock()
@@ -85,7 +97,7 @@ type IHttpEventData interface {
 	Handle()
 }
 
-func (slf *EventProcessor) EventHandler(processor IEventProcessor,ev *Event) {
+func (slf *EventProcessor) EventHandler(ev *Event) {
 	defer func() {
 		if r := recover(); r != nil {
 			buf := make([]byte, 4096)
@@ -99,7 +111,11 @@ func (slf *EventProcessor) EventHandler(processor IEventProcessor,ev *Event) {
 		return
 	}
 
-	processor.OnEventHandler(ev)
+	if fun,ok := slf.mapEventReciverFunc[ev.Type];ok == false{
+		return
+	}else{
+		fun(ev)
+	}
 }
 
 func (slf *EventProcessor) innerEventHandler(ev *Event) bool {

example/GateService/GateService.go

@@ -9,6 +9,8 @@ import (
 	"github.com/duanhf2012/origin/node"
 	"github.com/duanhf2012/origin/service"
 	"github.com/duanhf2012/origin/sysservice"
+	"github.com/duanhf2012/origin/util/timer"
+	"net/http"
 )
 
 type GateService struct {
@@ -29,9 +31,17 @@ func (slf *GateService) OnInit() error{
 	slf.httpRouter.GET("/get/query", slf.HttpTest)
 	slf.httpRouter.POST("/post/query", slf.HttpTestPost)
 	slf.httpRouter.SetServeFile(sysservice.METHOD_GET,"/img/head/","d:/img")
+
+	pCronExpr,_ := timer.NewCronExpr("0 * * * * *")
+	slf.CronFunc(pCronExpr,slf.Test)
+
 	return nil
 }
 
+func (slf *GateService) Test(){
+	fmt.Print("xxxxx\n")
+}
+
 func (slf *GateService) HttpTest(session *sysservice.HttpSession) {
 	session.SetHeader("a","b")
 	session.Write([]byte("this is a test"))
@@ -57,10 +67,11 @@ func (slf *GateService) HttpTestPost(session *sysservice.HttpSession) {
 
 	testa.AA = 100
 	testa.BB = "this is a test"
-	session.WriteJson("asdasda")
+	session.WriteJsonDone(http.StatusOK,"asdasda")
 }
 
 func (slf *GateService) OnEventHandler(ev *event.Event) error{
+
 	if ev.Type == event.Sys_Event_Tcp_RecvPack {
 		pPack := ev.Data.(*sysservice.TcpPack)
 		slf.processor.Route(ev.Data,pPack.ClientId)

network/tcp_conn.go

@@ -111,3 +111,7 @@ func (tcpConn *TCPConn) ReadMsg() ([]byte, error) {
 func (tcpConn *TCPConn) WriteMsg(args ...[]byte) error {
 	return tcpConn.msgParser.Write(tcpConn, args...)
 }
+
+func (tcpConn *TCPConn) IsConnected() bool {
+	return tcpConn.closeFlag == false
+}

rpc/client.go

@@ -219,3 +219,7 @@ func (slf *Client) Run(){
 
 func (slf *Client) OnClose(){
 }
+
+func (slf *Client) IsConnected() bool {
+	return slf.conn!=nil && slf.conn.IsConnected()==true
+}

service/service.go

@@ -2,7 +2,6 @@ package service
 
 import (
 	"fmt"
-	"github.com/duanhf2012/origin/event"
 	"github.com/duanhf2012/origin/log"
 	"github.com/duanhf2012/origin/profiler"
 	"github.com/duanhf2012/origin/rpc"
@@ -130,7 +129,7 @@ func (slf *Service) Run() {
 			if slf.profiler!=nil {
 				analyzer = slf.profiler.Push(fmt.Sprintf("Event_%d", int(ev.Type)))
 			}
-			slf.EventHandler(slf.this.(event.IEventProcessor),ev)
+			slf.EventHandler(ev)
 			if analyzer!=nil {
 				analyzer.Pop()
 				analyzer = nil

sysservice/httpservice.go

@@ -121,6 +121,7 @@ func NewHttpHttpRouter(eventReciver event.IEventProcessor) IHttpRouter {
 
 
 func (slf *HttpSession) Query(key string) (string, bool) {
+
 	if slf.mapParam == nil {
 		slf.mapParam = make(map[string]string)
 
@@ -158,6 +159,14 @@ func (slf *HttpSession) AddHeader(key, value string) {
 	slf.w.Header().Add(key,value)
 }
 
+func (slf *HttpSession) GetHeader(key string) string{
+	return slf.r.Header.Get(key)
+}
+
+func (slf *HttpSession) DelHeader(key string) {
+	slf.r.Header.Del(key)
+}
+
 func (slf *HttpSession) WriteStatusCode(statusCode int){
 	slf.statusCode = statusCode
 }
@@ -166,15 +175,18 @@ func (slf *HttpSession) Write(msg []byte) {
 	slf.msg = msg
 }
 
-func (slf *HttpSession) WriteJson(msgJson interface{}) error {
+func (slf *HttpSession) WriteJsonDone(statusCode int,msgJson interface{}) error {
 	msg, err := json.Marshal(msgJson)
 	if err == nil {
 		slf.Write(msg)
 	}
 
+	slf.Done()
 	return err
 }
 
+
+
 func (slf *HttpSession) flush() {
 	slf.w.WriteHeader(slf.statusCode)
 	if slf.msg!=nil {
@@ -182,7 +194,7 @@ func (slf *HttpSession) flush() {
 	}
 }
 
-func (slf *HttpSession) done(){
+func (slf *HttpSession) Done(){
 	slf.sessionDone <- slf
 }
 
@@ -249,7 +261,7 @@ func (slf *HttpRouter) Router(session *HttpSession){
 	if slf.httpFiltrateList!=nil {
 		for _,fun := range slf.httpFiltrateList{
 			if fun(session) == false {
-				session.done()
+				//session.done()
 				return
 			}
 		}
@@ -268,7 +280,7 @@ func (slf *HttpRouter) Router(session *HttpSession){
 		}
 
 		v.httpHandle(session)
-		session.done()
+		//session.done()
 		return
 	}
 
@@ -276,13 +288,13 @@ func (slf *HttpRouter) Router(session *HttpSession){
 		idx := strings.Index(urlPath, k)
 		if idx != -1 {
 			session.fileData = v
-			session.done()
+			session.Done()
 			return
 		}
 	}
 
 	session.WriteStatusCode(http.StatusNotFound)
-	session.done()
+	session.Done()
 }
 
 func (slf *HttpService) SetHttpRouter(httpRouter IHttpRouter) {