新增服务间事件通知机制

event/event.go

@@ -10,91 +10,190 @@ import (
 const Default_EventChannelLen = 10000
 
 //事件接受器
-type EventReciverFunc func(event *Event)
+type EventCallBack func(event *Event)
+
 
 type Event struct {
 	Type EventType
 	Data interface{}
 }
 
-type IEventProcessor interface {
+type IEventHandler interface {
+
+	GetEventProcessor() IEventProcessor  //获得事件
 	NotifyEvent(*Event)
 
-	SetEventReciver(eventProcessor IEventProcessor)
+	//注册了事件
-	GetEventReciver() IEventProcessor
+	addRegInfo(eventType EventType,eventProcessor IEventProcessor)
-	SetEventChanNum(num int32) bool
+	removeRegInfo(eventType EventType,eventProcessor IEventProcessor)
-	RegEventReciverFunc(eventType EventType,reciverFunc EventReciverFunc)
+
-	UnRegEventReciverFun(eventType EventType)
 }
 
+type IEventProcessor interface {
+	RegEventReciverFunc(eventType EventType,reciver IEventHandler,callback EventCallBack)
+	UnRegEventReciverFun(eventType EventType,reciver IEventHandler)
+	SetEventChannel(channelNum int) bool
+
+	castEvent(event *Event) //广播事件
+	pushEvent(event *Event)
+	addBindEvent(eventType EventType,reciver IEventHandler,callback EventCallBack)
+	addListen(eventType EventType,reciver IEventHandler)
+	removeBindEvent(eventType EventType,reciver IEventHandler)
+	removeListen(eventType EventType,reciver IEventHandler)
+}
+
+type EventHandler struct {
+	//已经注册的事件类型
+	eventProcessor IEventProcessor
+
+	//已经注册的事件
+	locker sync.RWMutex
+	mapRegEvent map[EventType]map[IEventProcessor]interface{}  //向其他事件处理器监听的事件类型
+}
+
+
 type EventProcessor struct {
-	//事件管道
+	eventChannel chan *Event
-	EventChan chan *Event
-	eventReciver IEventProcessor
 
-	eventChanNumLocker sync.RWMutex
+	locker sync.RWMutex
-	eventChanNum int32
+	mapListenerEvent map[EventType]map[IEventProcessor]int           //监听者信息
-	mapEventReciverFunc map[EventType]EventReciverFunc
+	mapBindHandlerEvent map[EventType]map[IEventHandler]EventCallBack//收到事件处理
 }
 
-func (slf *EventProcessor) RegEventReciverFunc(eventType EventType,reciverFunc EventReciverFunc){
+func (slf *EventHandler) addRegInfo(eventType EventType,eventProcessor IEventProcessor){
-	if slf.mapEventReciverFunc == nil {
+	slf.locker.Lock()
-		slf.mapEventReciverFunc = map[EventType]EventReciverFunc{}
+	defer slf.locker.Unlock()
-	}
+	if slf.mapRegEvent == nil {
-	slf.mapEventReciverFunc[eventType] = reciverFunc
+		slf.mapRegEvent = map[EventType]map[IEventProcessor]interface{}{}
-}
-
-func (slf *EventProcessor) UnRegEventReciverFun(eventType EventType){
-	delete(slf.mapEventReciverFunc,eventType)
-}
-
-func (slf *EventProcessor) NotifyEvent(pEvent *Event) {
-	if len(slf.EventChan) >= int(slf.eventChanNum) {
-		log.Error("event queue is full!")
-	}
-	slf.EventChan <-pEvent
-}
-
-
-
-func (slf *EventProcessor) GetEventChan() chan *Event{
-	slf.eventChanNumLocker.Lock()
-	defer  slf.eventChanNumLocker.Unlock()
-
-	if slf.eventChanNum == 0 {
-		slf.eventChanNum = Default_EventChannelLen
 	}
 
-	if slf.EventChan == nil {
+	if _,ok := slf.mapRegEvent[eventType] ;ok == false{
-		slf.EventChan = make(chan *Event,slf.eventChanNum)
+		slf.mapRegEvent[eventType] = map[IEventProcessor]interface{}{}
 	}
-
+	slf.mapRegEvent[eventType][eventProcessor] = nil
-	return slf.EventChan
 }
 
-//不允许重复设置
+func (slf *EventHandler) removeRegInfo(eventType EventType,eventProcessor IEventProcessor){
-func (slf *EventProcessor) SetEventChanNum(num int32) bool {
+	if _,ok :=slf.mapRegEvent[eventType];ok == true {
-	slf.eventChanNumLocker.Lock()
+		delete(slf.mapRegEvent[eventType],eventProcessor)
-	defer  slf.eventChanNumLocker.Unlock()
+	}
-	if slf.eventChanNum>0 {
+}
+
+func (slf *EventHandler) GetEventProcessor() IEventProcessor{
+	return slf.eventProcessor
+}
+
+func (slf *EventHandler) NotifyEvent(ev *Event){
+	slf.GetEventProcessor().castEvent(ev)
+}
+
+func (slf *EventHandler) Init(processor IEventProcessor){
+	slf.eventProcessor = processor
+}
+
+
+func (slf *EventProcessor) SetEventChannel(channelNum int) bool{
+	slf.locker.Lock()
+	defer slf.locker.Unlock()
+	if slf.eventChannel!=nil {
 		return false
 	}
 
-	slf.eventChanNum = num
+	if channelNum == 0 {
+		channelNum = Default_EventChannelLen
+	}
+
+	slf.eventChannel = make(chan *Event,channelNum)
 	return true
 }
 
-func (slf *EventProcessor) SetEventReciver(eventProcessor IEventProcessor){
+func (slf *EventProcessor) addBindEvent(eventType EventType,reciver IEventHandler,callback EventCallBack){
-	slf.eventReciver = eventProcessor
+	//mapBindHandlerEvent map[EventType]map[IEventHandler]EventCallBack//收到事件处理
+	slf.locker.Lock()
+	defer slf.locker.Unlock()
+	if slf.mapBindHandlerEvent == nil {
+		slf.mapBindHandlerEvent = map[EventType]map[IEventHandler]EventCallBack{}
+	}
+
+	if _,ok := slf.mapBindHandlerEvent[eventType]; ok == false {
+		slf.mapBindHandlerEvent[eventType] = map[IEventHandler]EventCallBack{}
+	}
+
+	slf.mapBindHandlerEvent[eventType][reciver] = callback
 }
 
+func (slf *EventProcessor) addListen(eventType EventType,reciver IEventHandler){
+	slf.locker.Lock()
+	defer slf.locker.Unlock()
 
-func (slf *EventProcessor) GetEventReciver() IEventProcessor{
+	//mapListenerEvent map[EventType]map[IEventProcessor]int
-	return slf.eventReciver
+	if slf.mapListenerEvent == nil {
+		slf.mapListenerEvent = map[EventType]map[IEventProcessor]int{}
+	}
+
+	if _,ok :=slf.mapListenerEvent[eventType];ok == false{
+		slf.mapListenerEvent[eventType] = map[IEventProcessor]int{}
+	}
+
+	slf.mapListenerEvent[eventType][reciver.GetEventProcessor()] += 1
 }
 
-type IHttpEventData interface {
+func (slf *EventProcessor) removeBindEvent(eventType EventType,reciver IEventHandler){
-	Handle()
+	slf.locker.Lock()
+	defer slf.locker.Unlock()
+	if _,ok := slf.mapBindHandlerEvent[eventType];ok == true{
+		delete(slf.mapBindHandlerEvent[eventType],reciver)
+	}
+}
+
+func (slf *EventProcessor) removeListen(eventType EventType,reciver IEventHandler){
+	slf.locker.Lock()
+	defer slf.locker.Unlock()
+	if _,ok := slf.mapListenerEvent[eventType];ok == true{
+		slf.mapListenerEvent[eventType][reciver.GetEventProcessor()]-=1
+		if slf.mapListenerEvent[eventType][reciver.GetEventProcessor()] <= 0 {
+			delete(slf.mapListenerEvent[eventType],reciver.GetEventProcessor())
+		}
+	}
+}
+
+func (slf *EventProcessor) RegEventReciverFunc(eventType EventType,reciver IEventHandler,callback EventCallBack){
+	//记录reciver自己注册过的事件
+	reciver.addRegInfo(eventType,slf)
+	//记录当前所属IEventProcessor注册的回调
+	reciver.GetEventProcessor().addBindEvent(eventType,reciver,callback)
+	//将注册加入到监听中
+	slf.addListen(eventType,reciver)
+}
+
+func (slf *EventProcessor) UnRegEventReciverFun(eventType EventType,reciver IEventHandler) {
+	slf.removeListen(eventType,reciver)
+	reciver.GetEventProcessor().removeBindEvent(eventType,reciver)
+	reciver.removeRegInfo(eventType,slf)
+}
+
+func (slf *EventHandler) desctory(){
+	for eventTyp,mapEventProcess := range slf.mapRegEvent {
+		if mapEventProcess == nil {
+			continue
+		}
+
+		//map[IEventProcessor]interface{}
+		for eventProcess,_ := range mapEventProcess {
+			eventProcess.UnRegEventReciverFun(eventTyp,slf)
+		}
+	}
+}
+
+func (slf *EventProcessor) GetEventChan() chan *Event{
+	slf.locker.Lock()
+	defer slf.locker.Unlock()
+
+	if slf.eventChannel == nil {
+		slf.eventChannel =make(chan *Event,Default_EventChannelLen)
+	}
+
+	return slf.eventChannel
 }
 
 func (slf *EventProcessor) EventHandler(ev *Event) {
@@ -107,25 +206,40 @@ func (slf *EventProcessor) EventHandler(ev *Event) {
 		}
 	}()
 
-	if slf.innerEventHandler(ev) == true {
+	mapCallBack,ok := slf.mapBindHandlerEvent[ev.Type]
+	if ok == false {
 		return
 	}
-
+	for _,callback := range mapCallBack {
-	if fun,ok := slf.mapEventReciverFunc[ev.Type];ok == false{
+		callback(ev)
-		return
-	}else{
-		fun(ev)
 	}
 }
 
-func (slf *EventProcessor) innerEventHandler(ev *Event) bool {
+
-	switch ev.Type {
+
-	case Sys_Event_Http_Event:
+
-		ev.Data.(IHttpEventData).Handle()
+func (slf *EventProcessor) pushEvent(event *Event){
-		return true
+	if len(slf.eventChannel)>=cap(slf.eventChannel){
+		log.Error("event process channel is full.")
+		return
 	}
 
-	return false
+	slf.eventChannel<-event
 }
 
+func (slf *EventProcessor) castEvent(event *Event){
+	if slf.mapListenerEvent == nil{
+		log.Error("mapListenerEvent not init!")
+		return
+	}
 
+	processor,ok :=slf.mapListenerEvent[event.Type]
+	if ok == false || processor == nil{
+		log.Debug("event type %d not listen.",event.Type)
+		return
+	}
+
+	for proc,_ := range processor {
+		proc.pushEvent(event)
+	}
+}

event/eventtype.go

@@ -4,11 +4,8 @@ type EventType int
 
 //大于Sys_Event_User_Define给用户定义
 const (
-	Sys_Event_Tcp_Connected EventType= 1
+	Sys_Event_Tcp       EventType = 5
-	Sys_Event_Tcp_DisConnected EventType= 2
+	Sys_Event_Http_Event EventType = 4
-	Sys_Event_Tcp_RecvPack EventType = 3
-	Sys_Event_Tcp_PackException EventType = 4
-	Sys_Event_Http_Event EventType = 5
 
 	Sys_Event_User_Define EventType = 1000
 )

example/GateService/GateService.go

@@ -3,8 +3,6 @@ package GateService
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/duanhf2012/origin/event"
-	"github.com/duanhf2012/origin/network"
 	"github.com/duanhf2012/origin/network/processor"
 	"github.com/duanhf2012/origin/node"
 	"github.com/duanhf2012/origin/service"
@@ -15,18 +13,20 @@ import (
 
 type GateService struct {
 	service.Service
-	processor network.Processor
+	processor *processor.PBProcessor
 	httpRouter sysservice.IHttpRouter
 }
 
 func (slf *GateService) OnInit() error{
 	tcpervice := node.GetService("TcpService").(*sysservice.TcpService)
 	slf.processor = &processor.PBProcessor{}
-	tcpervice.SetProcessor(slf.processor)
+	slf.processor.RegisterDisConnected(slf.OnDisconnected)
+	slf.processor.RegisterConnected(slf.OnConnected)
+	tcpervice.SetProcessor(slf.processor,slf.GetEventHandler())
 
 	httpervice := node.GetService("HttpService").(*sysservice.HttpService)
-	slf.httpRouter = sysservice.NewHttpHttpRouter(slf)
+	slf.httpRouter = sysservice.NewHttpHttpRouter()
-	httpervice.SetHttpRouter(slf.httpRouter)
+	httpervice.SetHttpRouter(slf.httpRouter,slf.GetEventHandler())
 
 	slf.httpRouter.GET("/get/query", slf.HttpTest)
 	slf.httpRouter.POST("/post/query", slf.HttpTestPost)
@@ -70,21 +70,6 @@ func (slf *GateService) HttpTestPost(session *sysservice.HttpSession) {
 	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)
-	}else if ev.Type == event.Sys_Event_Tcp_Connected {
-		pPack := ev.Data.(*sysservice.TcpPack)
-		slf.OnConnected(pPack.ClientId)
-	}else if ev.Type == event.Sys_Event_Tcp_DisConnected {
-		pPack := ev.Data.(*sysservice.TcpPack)
-		slf.OnDisconnected(pPack.ClientId)
-	}
-	return nil
-}
-
 func (slf *GateService) OnConnected(clientid uint64){
 	fmt.Printf("client id %d connected",clientid)
 }

example/main.go

@@ -235,7 +235,7 @@ func main(){
 
 	tcpService := &sysservice.TcpService{}
 	gateService := &GateService.GateService{}
-	tcpService.SetEventReciver(gateService)
+
 
 	httpService := &sysservice.HttpService{}
 

network/processor.go

@@ -3,7 +3,13 @@ package network
 
 type Processor interface {
 	// must goroutine safe
-	Route(msg interface{}, userData interface{}) error
+	MsgRoute(msg interface{}, userData interface{}) error
+	//must goroutine safe
+	UnknownMsgRoute(msg interface{}, userData interface{})
+	// connect event
+	ConnectedRoute(userData interface{})
+	DisConnectedRoute(userData interface{})
+
 	// must goroutine safe
 	Unmarshal(data []byte) (interface{}, error)
 	// must goroutine safe

network/processor/pbprocessor.go

@@ -13,10 +13,17 @@ type MessageInfo struct {
 }
 
 type MessageHandler func(clientid uint64,msg proto.Message)
+type ConnectHandler func(clientid uint64)
+type UnknownMessageHandler func(clientid uint64,msg []byte)
+
 const MsgTypeSize = 2
 type PBProcessor struct {
 	mapMsg map[uint16]MessageInfo
 	LittleEndian bool
+
+	unknownMessageHandler UnknownMessageHandler
+	connectHandler ConnectHandler
+	disconnectHandler ConnectHandler
 }
 
 func NewPBProcessor() *PBProcessor {
@@ -44,7 +51,7 @@ func (slf *PBPackInfo) GetMsg() proto.Message {
 }
 
 // must goroutine safe
-func (slf *PBProcessor ) Route(msg interface{},userdata interface{}) error{
+func (slf *PBProcessor ) MsgRoute(msg interface{},userdata interface{}) error{
 	pPackInfo := msg.(*PBPackInfo)
 	v,ok := slf.mapMsg[pPackInfo.typ]
 	if ok == false {
@@ -117,3 +124,31 @@ func (slf *PBProcessor) MakeMsg(msgType uint16,protoMsg proto.Message) *PBPackIn
 func (slf *PBProcessor) MakeRawMsg(msgType uint16,msg []byte) *PBPackInfo {
 	return &PBPackInfo{typ:msgType,rawMsg:msg}
 }
+
+func (slf *PBProcessor) UnknownMsgRoute(msg interface{}, userData interface{}){
+	slf.unknownMessageHandler(userData.(uint64),msg.([]byte))
+}
+
+// connect event
+func (slf *PBProcessor) ConnectedRoute(userData interface{}){
+	slf.connectHandler(userData.(uint64))
+}
+
+func (slf *PBProcessor) DisConnectedRoute(userData interface{}){
+	slf.disconnectHandler(userData.(uint64))
+}
+
+func (slf *PBProcessor) RegisterUnknownMsg(unknownMessageHandler UnknownMessageHandler){
+	slf.unknownMessageHandler = unknownMessageHandler
+}
+
+func (slf *PBProcessor) RegisterConnected(connectHandler ConnectHandler){
+	slf.connectHandler = connectHandler
+}
+
+func (slf *PBProcessor) RegisterDisConnected(disconnectHandler ConnectHandler){
+	slf.disconnectHandler = disconnectHandler
+}
+
+
+

service/module.go

@@ -26,8 +26,9 @@ type IModule interface {
 	OnRelease()
 	getBaseModule() IModule
 	GetService() IService
-	GetEventChan() chan *event.Event
 	GetModuleName() string
+	GetEventProcessor()event.IEventProcessor
+	NotifyEvent(ev *event.Event)
 }
 
 
@@ -50,8 +51,8 @@ type Module struct {
 	descendants map[int64]IModule//始祖的后裔们
 
 	//事件管道
-	event.EventProcessor
 	moduleName string
+	eventHandler event.EventHandler
 	//eventChan chan *SEvent
 }
 
@@ -96,6 +97,7 @@ func (slf *Module) AddModule(module IModule) (int64,error){
 	pAddModule.dispatcher = slf.GetAncestor().getBaseModule().(*Module).dispatcher
 	pAddModule.ancestor = slf.ancestor
 	pAddModule.moduleName = reflect.Indirect(reflect.ValueOf(module)).Type().Name()
+	pAddModule.eventHandler.Init(slf.eventHandler.GetEventProcessor())
 	err := module.OnInit()
 	if err != nil {
 		return 0,err
@@ -202,3 +204,14 @@ func (slf *Module) GetService() IService {
 	return slf.GetAncestor().(IService)
 }
 
+func (slf *Module) GetEventProcessor() event.IEventProcessor{
+	return slf.eventHandler.GetEventProcessor()
+}
+
+func (slf *Module) NotifyEvent(ev *event.Event){
+	slf.eventHandler.NotifyEvent(ev)
+}
+
+func (slf *Module) GetEventHandler() event.IEventHandler{
+	return &slf.eventHandler
+}

service/service.go

@@ -2,6 +2,7 @@ package service
 
 import (
 	"fmt"
+	"github.com/duanhf2012/origin/event"
 	"github.com/duanhf2012/origin/log"
 	"github.com/duanhf2012/origin/profiler"
 	"github.com/duanhf2012/origin/rpc"
@@ -42,7 +43,7 @@ type Service struct {
 	serviceCfg interface{}
 	gorouterNum int32
 	startStatus bool
-
+	eventProcessor event.EventProcessor //事件接收者
 	profiler *profiler.Profiler //性能分析器
 }
 
@@ -70,6 +71,7 @@ func (slf *Service) Init(iservice IService,getClientFun rpc.FuncRpcClient,getSer
 	slf.descendants = map[int64]IModule{}
 	slf.serviceCfg = serviceCfg
 	slf.gorouterNum = 1
+	slf.eventHandler.Init(&slf.eventProcessor)
 	slf.this.OnInit()
 }
 
@@ -101,7 +103,7 @@ func (slf *Service) Run() {
 	for{
 		rpcRequestChan := slf.GetRpcRequestChan()
 		rpcResponeCallBack := slf.GetRpcResponeChan()
-		eventChan := slf.GetEventChan()
+		eventChan := slf.eventProcessor.GetEventChan()
 		var analyzer *profiler.Analyzer
 		select {
 		case <- closeSig:
@@ -129,7 +131,7 @@ func (slf *Service) Run() {
 			if slf.profiler!=nil {
 				analyzer = slf.profiler.Push(fmt.Sprintf("Event_%d", int(ev.Type)))
 			}
-			slf.EventHandler(ev)
+			slf.eventProcessor.EventHandler(ev)
 			if analyzer!=nil {
 				analyzer.Pop()
 				analyzer = nil
@@ -195,3 +197,11 @@ func (slf *Service) GetProfiler() *profiler.Profiler{
 }
 
 
+
+func (slf *Service) RegEventReciverFunc(eventType event.EventType,reciver event.IEventHandler,callback event.EventCallBack){
+	slf.eventProcessor.RegEventReciverFunc(eventType,reciver,callback)
+}
+
+func (slf *Service) UnRegEventReciverFun(eventType event.EventType,reciver event.IEventHandler){
+	slf.eventProcessor.UnRegEventReciverFun(eventType,reciver)
+}

sysservice/httpservice.go

@@ -57,7 +57,6 @@ type IHttpRouter interface {
 	POST(url string, handle HttpHandle) bool
 	Router(session *HttpSession)
 
-	PutHttpSession(httpSession *HttpSession)
 	SetServeFile(method HTTP_METHOD, urlpath string, dirname string) error
 	SetFormFileKey(formFileKey string)
 	GetFormFileKey()string
@@ -67,7 +66,7 @@ type IHttpRouter interface {
 type HttpRouter struct {
 	pathRouter map[HTTP_METHOD] map[string] routerMatchData //url地址，对应本service地址
 	serveFileData  map[string] *routerServeFileData
-	eventReciver event.IEventProcessor
+	//eventReciver event.IEventHandler
 	httpFiltrateList [] HttpFiltrate
 
 	formFileKey string
@@ -104,9 +103,9 @@ type HttpService struct {
 
 
 
-func NewHttpHttpRouter(eventReciver event.IEventProcessor) IHttpRouter {
+func NewHttpHttpRouter() IHttpRouter {
 	httpRouter := &HttpRouter{}
-	httpRouter.eventReciver = eventReciver
+	//httpRouter.eventReciver = eventHandler
 	httpRouter.pathRouter =map[HTTP_METHOD] map[string] routerMatchData{}
 	httpRouter.serveFileData = map[string] *routerServeFileData{}
 	httpRouter.formFileKey = "file"
@@ -185,8 +184,6 @@ func (slf *HttpSession) WriteJsonDone(statusCode int,msgJson interface{}) error
 	return err
 }
 
-
-
 func (slf *HttpSession) flush() {
 	slf.w.WriteHeader(slf.statusCode)
 	if slf.msg!=nil {
@@ -235,9 +232,6 @@ func (slf *HttpRouter) GetFormFileKey()string{
 	return slf.formFileKey
 }
 
-func (slf *HttpRouter) PutHttpSession(httpSession *HttpSession){
-	slf.eventReciver.NotifyEvent(&event.Event{Type:event.Sys_Event_Http_Event,Data:httpSession})
-}
 
 func (slf *HttpRouter) GET(url string, handle HttpHandle) bool {
 	return slf.regRouter(METHOD_GET, url, handle)
@@ -297,8 +291,14 @@ func (slf *HttpRouter) Router(session *HttpSession){
 	session.Done()
 }
 
-func (slf *HttpService) SetHttpRouter(httpRouter IHttpRouter) {
+
+func (slf *HttpService) HttpEventHandler(ev *event.Event)  {
+	ev.Data.(*HttpSession).Handle()
+}
+
+func (slf *HttpService) SetHttpRouter(httpRouter IHttpRouter,eventHandler event.IEventHandler) {
 	slf.httpRouter = httpRouter
+	slf.RegEventReciverFunc(event.Sys_Event_Http_Event,eventHandler,slf.HttpEventHandler)
 }
 
 
@@ -521,7 +521,7 @@ func (slf *HttpService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 	session.body = body
 
-	slf.httpRouter.PutHttpSession(session)
+	slf.GetEventHandler().NotifyEvent(&event.Event{Type:event.Sys_Event_Http_Event,Data:session})
 	ticker := time.NewTicker(slf.processTimeout)
 	select {
 	case <-ticker.C:

sysservice/tcpservice.go

@@ -71,8 +71,41 @@ func (slf *TcpService) OnInit() error{
 	return nil
 }
 
-func (slf *TcpService) SetProcessor(process network.Processor){
+
+
+
+type TcpPackType int8
+const(
+	TPT_Connected TcpPackType = 0
+	TPT_DisConnected TcpPackType = 1
+	TPT_Pack TcpPackType = 2
+	TPT_UnknownPack TcpPackType = 3
+)
+
+type TcpPack struct {
+	Type TcpPackType //0表示连接 1表示断开 2表示数据
+	MsgProcessor network.Processor
+	ClientId uint64
+	Data interface{}
+}
+
+
+func (slf *TcpService) TcpEventHandler(ev *event.Event) {
+		pack := ev.Data.(*TcpPack)
+		if pack.Type == TPT_Connected {
+			pack.MsgProcessor.ConnectedRoute(pack.ClientId)
+		}else if pack.Type == TPT_DisConnected {
+			pack.MsgProcessor.DisConnectedRoute(pack.ClientId)
+		} else if pack.Type == TPT_UnknownPack{
+			pack.MsgProcessor.UnknownMsgRoute(pack.Data,pack.ClientId)
+		} else if pack.Type == TPT_Pack {
+			pack.MsgProcessor.MsgRoute(pack.Data, pack.ClientId)
+		}
+}
+
+func (slf *TcpService) SetProcessor(process network.Processor,handler event.IEventHandler){
 	slf.process = process
+	slf.RegEventReciverFunc(event.Sys_Event_Tcp,handler,slf.TcpEventHandler)
 }
 
 func (slf *TcpService) NewClient(conn *network.TCPConn) network.Agent {
@@ -103,17 +136,14 @@ type Client struct {
 	tcpService *TcpService
 }
 
-type TcpPack struct {
+
-	ClientId uint64
-	Data interface{}
-}
 
 func (slf *Client) GetId() uint64 {
 	return slf.id
 }
 
 func (slf *Client) Run() {
-	slf.tcpService.GetEventReciver().NotifyEvent(&event.Event{Type:event.Sys_Event_Tcp_Connected,Data:&TcpPack{ClientId:slf.id}})
+	slf.tcpService.NotifyEvent(&event.Event{Type:event.Sys_Event_Tcp,Data:&TcpPack{ClientId:slf.id,Type:TPT_Connected,MsgProcessor:slf.tcpService.process}})
 	for{
 		bytes,err := slf.tcpConn.ReadMsg()
 		if err != nil {
@@ -122,17 +152,15 @@ func (slf *Client) Run() {
 		}
 		data,err:=slf.tcpService.process.Unmarshal(bytes)
 		if err != nil {
-			slf.tcpService.GetEventReciver().NotifyEvent(&event.Event{Type:event.Sys_Event_Tcp_PackException,Data:&TcpPack{ClientId:slf.id,Data:bytes}})
+			slf.tcpService.NotifyEvent(&event.Event{Type:event.Sys_Event_Tcp,Data:&TcpPack{ClientId:slf.id,Type:TPT_UnknownPack,Data:bytes,MsgProcessor:slf.tcpService.process}})
-			//log.Debug("process.Unmarshal is error:%+v",err)
 			continue
 		}
-
+		slf.tcpService.NotifyEvent(&event.Event{Type:event.Sys_Event_Tcp,Data:&TcpPack{ClientId:slf.id,Type:TPT_Pack,Data:data,MsgProcessor:slf.tcpService.process}})
-		slf.tcpService.GetEventReciver().NotifyEvent(&event.Event{Type:event.Sys_Event_Tcp_RecvPack,Data:&TcpPack{ClientId:slf.id,Data:data}})
 	}
 }
 
 func (slf *Client) OnClose(){
-	slf.tcpService.GetEventReciver().NotifyEvent(&event.Event{Type:event.Sys_Event_Tcp_DisConnected,Data:&TcpPack{ClientId:slf.id}})
+	slf.tcpService.NotifyEvent(&event.Event{Type:event.Sys_Event_Tcp,Data:&TcpPack{ClientId:slf.id,Type:TPT_DisConnected,MsgProcessor:slf.tcpService.process}})
 	slf.tcpService.mapClientLocker.Lock()
 	defer slf.tcpService.mapClientLocker.Unlock()
 	delete (slf.tcpService.mapClient,slf.GetId())